SilvaEngine Utility

A high-performance utility library providing JSON serialization, AWS integration, and utility functions with comprehensive performance monitoring.

Features

• High-Performance JSON Handling: Unified JSON serialization with automatic fallback from orjson to standard library
• SQLAlchemy Integration: Smart serialization of SQLAlchemy models with relationship handling and circular reference prevention
• Hybrid Cache System: Redis-primary caching with disk fallback, optimized for AWS Lambda serverless environments
• AWS Integration: Simplified Lambda invocation, SQS messaging, and GraphQL query execution
• Performance Monitoring: Built-in performance tracking for JSON operations with detailed metrics
• Datetime Handling: Advanced datetime parsing with Pendulum integration and caching
• Type Safety: Comprehensive type handling for Decimal, datetime, bytes, and custom objects
• Pagination: Shared offset/limit pagination utilities (cursor encoding/decoding, Relay Connection builder) for banyan business modules
• Idempotency: Shared IdempotencyRepository / check_and_store / store_result / purge_expired plus idempotent_mutation decorator for banyan Mutation deduplication
• Auth Helpers: Shared get_operator_id / require_operator_id / OperatorIdRequiredError for banyan Mutation operator-id guards

Installation

# Basic installation
pip install silvaengine-utility

# With performance optimizations (recommended)
pip install silvaengine-utility[performance]

# For development
pip install silvaengine-utility[dev]

# All features
pip install silvaengine-utility[all]

Quick Start

JSON Operations

from silvaengine_utility import Serializer

# Serialize data with optimized performance
data = {"user": "john", "timestamp": datetime.now(), "amount": Decimal("100.50")}
json_string = Serializer.json_dumps(data)

# Deserialize with automatic type conversion
parsed_data = Serializer.json_loads(json_string)

# Performance monitoring
Serializer.reset_json_performance_stats()
# ... perform operations ...
stats = Serializer.get_json_performance_stats()
print(f"Operations: {stats['json_dumps']['count']}")

Cache Operations

from silvaengine_utility import hybrid_cache, authorization_cache, method_cache

# Basic function caching
@hybrid_cache(ttl=300)
def expensive_function(param):
    return complex_calculation(param)

# Authorization caching for Lambda functions
@authorization_cache(ttl=300)
def _dynamic_authorization(self, event, context, action):
    return authorize_user(event, action)

# Method caching with custom cache names
@method_cache(ttl=1800, cache_name="settings")
def get_settings(setting_key):
    return settings_lookup(setting_key)

AWS Lambda Integration

# Invoke AWS Lambda function
result = Utility.invoke_funct_on_aws_lambda(
    logger=logger,
    endpoint_id="my-endpoint",
    funct="process_data",
    params={"data": "value"},
    execute_mode="local_for_all",  # For local testing
    aws_lambda=lambda_client
)

# Execute GraphQL queries
response = Utility.execute_graphql_query(
    logger=logger,
    endpoint_id="graphql-endpoint",
    funct="execute_query",
    query="{ users { id name } }",
    variables={"limit": 10}
)

Utility Functions

# Network utilities
is_in_subnet = Utility.in_subnet("192.168.1.100", "192.168.1.0/24")

# Dynamic imports
func = Utility.import_dynamically(
    module_name="mymodule",
    function_name="myfunction",
    class_name="MyClass"
)

# Database session creation
session = Utility.create_database_session({
    "user": "username",
    "password": "password",
    "host": "localhost",
    "port": 3306,
    "schema": "mydb"
})

Core Components

JSON Handler

The HighPerformanceJSONHandler provides:

• Unified Serialization: Single handler for all object types
• Performance Optimization: Uses orjson when available, falls back gracefully
• SQLAlchemy Support: Deep serialization of models with relationship handling
• Type Intelligence: Smart handling of Decimal (int for whole numbers, float for decimals)
• Datetime Consistency: ISO format serialization with automatic parsing
• Depth Control: Prevents infinite recursion with configurable depth limits
• Performance Monitoring: Automatic tracking of all operations

Performance Monitor

Built-in monitoring provides:

# Get comprehensive statistics
stats = Utility.get_json_performance_stats()
# {
#   "json_dumps": {
#     "count": 1250,
#     "total_time": 0.125,
#     "avg_time": 0.0001,
#     "errors": 0
#   },
#   "json_loads": { ... }
# }

# Get human-readable summary
summary = Utility.get_json_performance_summary()

# Reset metrics
Utility.reset_json_performance_stats()

AWS Integration

Execution Modes

• "local_for_all": Execute all functions locally
• "local_for_sqs": Execute SQS functions locally, others on AWS
• "local_for_aws_lambda": Execute Lambda functions locally when possible

SQS Integration

# Send message to SQS queue
Utility.invoke_funct_on_aws_lambda(
    logger=logger,
    endpoint_id="endpoint",
    funct="process",
    params={"data": "value"},
    message_group_id="group-1",
    task_queue=sqs_queue
)

Datetime Handling

Advanced datetime processing with caching:

from silvaengine_utility.datetime_handler import PendulumDateTimeHandler

# Parse ISO datetime strings
dt = PendulumDateTimeHandler.parse_datetime_in_json("2023-01-01T12:00:00Z")

# Get library info
info = Utility.get_library_info()
# {
#   "json": {"library": "orjson", "high_performance": true},
#   "datetime": {"library": "pendulum", "version": "2.1.0"}
# }

Pagination

Shared offset/limit pagination utilities for banyan business modules:

from silvaengine_utility.pagination import (
    encode_offset_cursor,
    decode_cursor,
    build_connection,
    DEFAULT_PAGE_SIZE,
    MAX_PAGE_LIMIT,
)

# Encode offset as base64 cursor
cursor = encode_offset_cursor(20)  # "MjA="

# Decode cursor back to offset
offset = decode_cursor(cursor)  # 20

# Build a Relay Connection payload from rows + total
connection = build_connection(rows, total, first=10, after=cursor)
# { "edges": [...], "pageInfo": { "hasNextPage": True, "endCursor": ... }, "totalCount": 100 }

Idempotency

Shared Mutation deduplication for banyan business modules. Uses tenant_idempotency_record (composite key partition_key + idempotency_key) with INSERT ... ON CONFLICT DO NOTHING for concurrent-safe deduplication:

from silvaengine_utility.idempotency import (
    IdempotencyRepository,
    check_and_store,
    store_result,
    purge_expired,
    idempotent_mutation,
)

# Low-level repository (for custom flows)
existing = check_and_store(partition_key, idempotency_key, mutation_name)
if existing is not None:
    return existing
result = mutate(...)
store_result(partition_key, idempotency_key, mutation_name, result)

# Decorator (recommended for graphene Mutations)
@idempotent_mutation("CreateAgent")
def mutate(root, info, **kwargs):
    ...
    return {"node": agent}

The idempotent_mutation decorator:

• Retrieves idempotency_key from kwargs only (no positional-arg fallback, which previously mis-bound id/session_id as the key)
• On check_and_store exception: logs ERROR and proceeds (degrades to non-idempotent for this request, does not block business)
• On store_result exception: logs ERROR and returns the computed result (business succeeded; idempotency not persisted, client retry will re-execute)
• Preserves the wrapped function's return structure exactly

Auth Helpers

Shared operator-id retrieval and guard for banyan business module Mutations:

from silvaengine_utility.auth import (
    get_operator_id,
    require_operator_id,
    OperatorIdRequiredError,
)

# Read-only (returns None when missing, no fallback)
operator_id = get_operator_id(info)

# Guard (raises when missing) — use at the entry of write Mutations
operator_id = require_operator_id(info, "DeleteAgent")

get_operator_id reads info.context["user_id"] then info.context["operator_id"], returns Optional[str] (no fallback — historical or "system" fallbacks are preserved at call sites, semantics unchanged). require_operator_id raises OperatorIdRequiredError(mutation_name) when both are missing. Guards are not enforced by the shared helper — adding them at each Mutation remains the business module's responsibility.`

Configuration

Optional Dependencies

• orjson: High-performance JSON serialization (3-5x faster)
• pendulum: Advanced datetime handling with caching
• graphql-core: GraphQL query support

Environment Variables

The library automatically detects and uses available performance libraries:

• If orjson is available: Uses for high-performance serialization
• If pendulum is available: Uses for datetime parsing with caching
• Falls back to standard library gracefully

API Reference

Utility Class

JSON Operations

• json_dumps(data, **kwargs) - Serialize to JSON with performance monitoring
• json_loads(data, parser_number=True, validate=True) - Parse JSON with type conversion
• is_json_string(string) - Validate JSON string

Performance Monitoring

• get_json_performance_stats() - Get detailed performance metrics
• get_json_performance_summary() - Get human-readable summary
• reset_json_performance_stats() - Reset performance counters
• get_library_info() - Get information about loaded performance libraries

AWS Integration

• invoke_funct_on_aws_lambda(...) - Invoke Lambda with local fallback support
• invoke_funct_on_local(...) - Execute configured local functions
• execute_graphql_query(...) - Execute GraphQL queries with normalization
• fetch_graphql_schema(...) - Fetch and parse GraphQL schemas

Utilities

• in_subnet(ip, subnet) - Check IP subnet membership
• import_dynamically(...) - Dynamic module/function imports
• create_database_session(settings) - Create SQLAlchemy database sessions
• convert_camel_to_underscore(text) - Convert camelCase to snake_case
• format_error(error) - Format GraphQL and generic errors

Hybrid Cache System

A high-performance caching system optimized for AWS Lambda serverless environments with Redis primary cache and disk fallback.

Cache Features

• Hybrid Architecture: Redis primary with automatic disk fallback
• Serverless Optimized: Designed for AWS Lambda cold/warm starts
• Decorator-Based: Easy-to-use decorators for common caching patterns
• Multiple Cache Types: Specialized decorators for different use cases
• Graceful Degradation: Works even when Redis is unavailable
• Performance Monitoring: Built-in cache statistics and management

Cache Usage

Basic Function Caching

from silvaengine_utility import hybrid_cache

@hybrid_cache(ttl=300)  # 5 minutes
def expensive_calculation(x: int, y: int) -> int:
    # Expensive operation
    return x * y + complex_computation()

# First call - executes function
result1 = expensive_calculation(5, 10)

# Second call - returns cached result
result2 = expensive_calculation(5, 10)  # Fast!

Authorization Caching

from silvaengine_utility import authorization_cache

class AuthHandler:
    @authorization_cache(ttl=300)
    def _dynamic_authorization(self, event, context, action):
        # Expensive authorization logic
        return authorize_user(event, action)

Method Caching (Class Methods)

from silvaengine_utility import method_cache

class DataService:
    @method_cache(ttl=1800, cache_name="settings")  # 30 min for settings
    def get_settings(self, setting_key: str) -> dict:
        return expensive_settings_lookup(setting_key)

    @method_cache(ttl=600, cache_name="database")   # 10 min for DB queries
    def get_user_data(self, user_id: str) -> dict:
        return database.query(user_id)

    @method_cache(ttl=300, cache_name="validation") # 5 min for validation
    def validate_permission(self, user_id: str, action: str) -> bool:
        return expensive_permission_check(user_id, action)

Cache Configuration

Environment Variables

# Redis Configuration (optional)
REDIS_HOST=your-redis-host.com
REDIS_PORT=6379
REDIS_PASSWORD=your-password
REDIS_DB=0

# Cache Directory
CACHE_DIR=/tmp/silvaengine_cache

Cache Names

Different cache instances for different use cases:

• default - General purpose caching
• authorization - Authorization results
• method - General method caching (default)
• settings - Configuration data (via method_cache)
• database - Database query results (via method_cache)
• validation - Permission validation (via method_cache)

Cache Management

# Clear specific function cache
my_function.cache_clear()

# Delete specific cache entry
my_function.cache_delete(arg1, arg2)

# Get cache statistics
stats = my_function.cache_stats()
print(stats)
# Output: {
#   'cache_name': 'default',
#   'redis_available': True,
#   'disk_available': True,
#   'disk_path': '/tmp/silvaengine_cache/default'
# }

Performance Benefits

Lambda Cold Start Optimization

# Without caching
def get_settings(setting_id):
    return expensive_db_lookup(setting_id)  # 200ms every time

# With caching
@method_cache(ttl=1800, cache_name="settings")  # 30 minutes
def get_settings(setting_id):
    return expensive_db_lookup(setting_id)  # 200ms first time, ~1ms after

Redis + Disk Hybrid Performance

Scenario	Redis	Disk	No Cache
Cold Start	N/A	~5ms	200ms
Warm Start (Redis)	~1ms	N/A	200ms
Warm Start (Disk)	N/A	~3ms	200ms
Redis Down	N/A	~3ms	200ms

This integration can improve Lambda performance by 80-95% for cached operations while maintaining full functionality when caches are unavailable.

Development

Running Tests

# Install development dependencies
pip install -e .[dev]

# Run tests with coverage
pytest --cov=silvaengine_utility --cov-report=term-missing

# Run performance benchmarks
pytest -m performance

Code Quality

# Format code
black silvaengine_utility/

# Lint code
flake8 silvaengine_utility/

# Type checking
mypy silvaengine_utility/

Performance

Benchmarks

With orjson enabled:

• JSON serialization: 3-5x faster than standard library
• JSON deserialization: 2-3x faster than standard library
• SQLAlchemy model serialization: Optimized with depth limiting

Memory Usage

• Efficient SQLAlchemy relationship handling
• LRU caching for datetime parsing
• Depth-limited recursion prevents memory exhaustion

Migration from v0.0.5

The library maintains backward compatibility while providing enhanced performance:

1. JSON Operations: Existing json_dumps/json_loads calls work unchanged
2. AWS Integration: Replace test_mode with execute_mode (old parameter still works with deprecation warning)
3. Performance: Install [performance] extras for optimal speed
4. Monitoring: New performance monitoring is automatically enabled

License

MIT License - see LICENSE file for details.

Support

• Repository: https://github.com/silvaengine/silvaengine-utility
• Issues: https://github.com/silvaengine/silvaengine-utility/issues
• Documentation: https://silvaengine-utility.readthedocs.io