Skip to main content

Python Customization

Keep generated packages replaceable. Domain shape belongs in KSML; trusted platform state and infrastructure belong in UserContext; HTTP/CLI input mapping belongs in handwritten application code.

Context factory​

Create one async factory per deployment profile:

@dataclass(frozen=True)
class TrustedRequest:
    tenant_id: int
    actor_id: str
    permissions: frozenset[str]

async def request_context(infra, trusted: TrustedRequest) -> UserContext:
    context = UserContext()
    # Register generated metadata, async provider, request policy and audit sinks.
    # Inject trusted identity and authorization here.
    return context

Handlers should never accept a provider argument beside the context, and client JSON must never initialize trusted resources.

The concrete runtime composition is:

def configured_runtime(
    module: RuntimeModule,
    schema_provider,
    request_policy,
    trusted_tenant: str,
    app_audit_sink,
) -> UserContext:
    if not trusted_tenant.strip():
        raise ValueError("trusted tenant is required")
    return (
        module.into_context()
        .with_schema_provider(schema_provider)
        .with_request_policy(request_policy)
        .with_app_audit_event_sink(app_audit_sink)
        .insert_resource("trusted_tenant", trusted_tenant)
    )


async def readiness(context: UserContext) -> None:
    context.require_resource("request_policy")
    context.require_resource("schema_provider")
    if not context.require_resource("trusted_tenant").strip():
        raise ValueError("trusted tenant is required")
    await context.ensure_schema()

Attach OpenTelemetry separately with context.with_runtime_telemetry(telemetry). Telemetry setup and ordinary health requests must not call ensure_schema() implicitly.

Explicit dynamic-query adapter​

Normalize and validate JSON before calling a generated Request. Use a closed mapping:

def apply_filter(request, item):
    match item["field"], item["operator"]:
        case "order_number", "contains":
            return request.with_order_number_containing(require_string(item["value"]))
        case _:
            raise InvalidQuery("unsupported field or operator")

Reject unknown fields, excessive IN lists, reversed/wrongly typed ranges, deep paths, forbidden sorts, negative offsets and oversized pages. Build rows, count, aggregate, and facets from the same normalized filter object.

Validation and behavior​

Implement the exact generated checker/behavior interfaces. Use them for entity-local normalization and invariants. Use request policy for tenant scope, permissions, purpose approval and platform limits. Test each extension through the public generated API so registration mistakes are visible.

Audit and application events​

Every mutation carries audit_as. Preserve the runtime row/mutation audit event as immutable evidence. Register a separate application sink for outbox or observability; mask configured sensitive fields before serialization and attribute the event to the trusted actor. An application log is not a replacement for runtime audit.

Async provider lifecycle​

Create pools at application startup, attach them through context initialization, and close them at shutdown. Do not create a connection per generated Request. Provider qualification must include real persistence, reconnect, optimistic locking, native aggregates, stable pagination, and exact relation Top-N.

Packaging after release​

Once published, pin the official PyPI runtime and provider extras in project metadata, record hashes in the lock file, and test the wheel in a clean environment. Until those coordinates are announced, documentation and examples must not claim a public install path.

If a customization needs edits in generated Q.py, E.py, models/, or requests/, move the change to the model, generator, or runtime extension point and regenerate.

Runtime infrastructure​

Python UserContext provides process-local cache and keyed locks plus optional Remote Cache/Remote Lock provider resources. Redis adapters are present, but the current implementations convert cache failures to misses/no-ops and the lock uses an unowned shared value with unconditional delete. Do not use that lock adapter as the sole distributed critical-section guard without an owner-safe atomic replacement. TeaQL currently claims no Python Nacos or Consul adapter. See Cache, Lock, and Cloud Runtime Infrastructure.

Worked Example: Local Cache and Lock​

async def currency_view(context: UserContext, tenant: str, code: str):
    cache_key = f"{tenant}:currency:{code}"
    cached = context.get_from_local_cache(cache_key, CurrencyView)
    if cached is not None:
        return cached

    lock_key = f"{tenant}:currency-refresh:{code}"
    if not context.try_local_lock(lock_key, timeout_millis=0, expire_millis=5_000):
        raise RuntimeError("currency refresh is busy")
    try:
        cached = context.get_from_local_cache(cache_key, CurrencyView)
        if cached is not None:
            return cached
        value = await load_authorized_currency_view(context, code)
        context.put_to_local_cache(cache_key, value, time_to_live_in_seconds=300)
        return value
    finally:
        context.unlock_local(lock_key)

Call remove_from_local_cache(cache_key) after an audited Mutation commits. For multiple processes, replace the current Redis lock adapter with an owner-safe atomic provider and require it in readiness; Local Lock cannot coordinate worker processes.