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:
    ctx = UserContext()
    # Register generated metadata, async provider, request policy and audit sinks.
    # Inject trusted identity and authorization here.
    return ctx

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

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.