Manage authenticated identities and data-masking policies across asynchronous boundaries with zero state pollution.
The Challenge
In a high-concurrency asynchronous application, tracking the "current user" or active security policies is notoriously difficult. Developers often face:
1. Argument Drilling: Passing a user object through five layers of service and repository calls just to log an ID or filter a query.
2. State Leakage: Using standard global variables or thread-locals, which can leak data between concurrent requests in an async/await environment.
3. Inconsistent Validation: Manually checking if a user ID is a valid UUID or if a list of restricted fields is mutable, leading to runtime errors in deep architectural layers.
The ZCore Elegance
ZCore utilizes Python's contextvars to provide a coroutine-safe execution context. The request_context manager allows you to bind a user identity and a set of data-masking restrictions at the start of a request. These variables are then globally accessible to any downstream function (like a Search Engine or Repository) without being explicitly passed as arguments.
from zcore.context import request_context, get_current_user_id
# 1. Bind context in a middleware or entry point
with request_context(user_id="...", fields=["products.cost"]):
# 2. Anywhere in your deep service/repository logic:
uid = get_current_user_id()
# uid is resolved automatically from the coroutine stack
# You must pass the user through every single function
async def get_products(user_id: uuid.UUID, restricted: list[str]):
return await repo.find_all(user_id, restricted)
async def find_all(user_id: uuid.UUID, restricted: list[str]):
# This becomes a maintenance nightmare in large apps
query = select(Product).where(Product.owner_id == user_id)
# ...
Boundaries & Integration
ZCore Context is a low-level utility that bridges the gap between Web and Infrastructure layers.
- Native ContextVars: ZCore uses standard
contextvars.ContextVar. This means it integrates perfectly with standard Python logging (viastructlog) and any other library that supports thecontextvarsprotocol. - Scoped Middleware: The
ScopedDependencyMiddlewareautomatically wraps every HTTP request in an empty context, ensuring a clean slate for every transaction. - Bypass: If you prefer explicit dependency injection, you can continue to pass
userobjects manually. ZCore's context is an additional capability, not a restrictive requirement.
The Execution Flow
Under-the-Hood Spec
1. Token-Based Restoration Safety
ZCore context managers (user_context, request_context) utilize the contextvars.Token system [context/context.py]. When a context is set, a token is generated. Upon exiting the block, the state is explicitly reset to its previous value using this token. This guarantees that even if a request crashes or a task is recycled, the security context never "bleeds" into the next execution.
2. Immutable Restriction Sets
When you set restricted_fields, ZCore automatically converts the input into a frozenset [context/context.py]. This ensures that once fields are restricted for a request, they cannot be modified by downstream services, preventing accidental or malicious "unmasking" of sensitive data during a complex transaction.
3. Immediate UUID Integrity
The set_current_user_id function performs immediate type coercion [context/context.py]. If a string is provided, it is validated and converted to a uuid.UUID instance instantly. This ensures that downstream database queries never fail due to string-to-UUID comparison mismatches, moving the validation error to the very boundary of the request.
Security Note
Context variables are isolated per asynchronous task. If you spawn a background task (e.g., asyncio.create_task), you must manually re-bind the context if the task requires the user's identity.