Core¶

The application bootstrap, DI container, config, routing, service providers, the application-level event dispatcher, view rendering, and the framework's exception/validation primitives.

application ¶

The Zeython application: an ASGI app assembled from service providers.

Application ¶

Application(
    config: Config | None = None,
    *,
    base_path: str | Path | None = None,
)

The central object: owns the container, config, router, and providers.

Application is itself a valid ASGI callable, so uvicorn main:app works directly once you construct one and register at least a router.

Source code in src/zeython/application.py

def __init__(self, config: Config | None = None, *, base_path: str | Path | None = None) -> None:
    self.base_path = Path(base_path) if base_path is not None else Path.cwd()
    self.config = config or Config.load(self.base_path)
    _configure_default_logging(self.config)
    self.container = Container()
    self.container.instance(Config, self.config)
    self.container.instance(Container, self.container)

    self.router = Router()
    self.container.instance(Router, self.router)

    self._providers: list[ServiceProvider] = []
    self._middleware: list[Middleware] = []
    self._booted = False
    self._asgi: Starlette | None = None

providers `property` ¶

providers: list[ServiceProvider]

Service providers registered so far, in registration order.

run ¶

run(
    *, host: str | None = None, port: int | None = None
) -> None

Run with uvicorn. For auto-reload during development, use the zeython serve CLI instead.

Source code in src/zeython/application.py

def run(self, *, host: str | None = None, port: int | None = None) -> None:
    """Run with uvicorn. For auto-reload during development, use the `zeython serve` CLI instead."""
    import uvicorn

    uvicorn.run(self, host=host or self.config.host, port=port or self.config.port)

container ¶

A minimal, type-hint-driven dependency injection container.

Zeython's :class:Container resolves dependencies by inspecting constructor and function type annotations, in the spirit of Laravel's service container. Bindings can be a plain instance, a factory callable, or a class to autowire directly.

BindingResolutionError ¶

Bases: Exception

Raised when the container cannot resolve a requested binding.

Container ¶

Container()

A small service container supporting binding, singletons, and autowiring.

Source code in src/zeython/container.py

def __init__(self) -> None:
    self._bindings: dict[Abstract, _Binding] = {}
    self._instances: dict[Abstract, Any] = {}

bind ¶

bind(
    abstract: Abstract,
    factory: Factory | None = None,
    *,
    shared: bool = False,
) -> None

Register a binding. If factory is omitted, abstract must be a concrete class.

Source code in src/zeython/container.py

def bind(self, abstract: Abstract, factory: Factory | None = None, *, shared: bool = False) -> None:
    """Register a binding. If ``factory`` is omitted, ``abstract`` must be a concrete class."""
    resolved_factory = factory or abstract
    if not callable(resolved_factory):
        raise TypeError(f"Binding for {abstract!r} must provide a callable factory")
    self._bindings[abstract] = _Binding(resolved_factory, shared)
    self._instances.pop(abstract, None)

singleton ¶

singleton(
    abstract: Abstract, factory: Factory | None = None
) -> None

Register a binding that is instantiated once and reused.

Source code in src/zeython/container.py

def singleton(self, abstract: Abstract, factory: Factory | None = None) -> None:
    """Register a binding that is instantiated once and reused."""
    self.bind(abstract, factory, shared=True)

instance ¶

instance(abstract: Abstract, value: Any) -> Any

Register an already-constructed instance under abstract.

Source code in src/zeython/container.py

def instance(self, abstract: Abstract, value: Any) -> Any:
    """Register an already-constructed instance under ``abstract``."""
    self._instances[abstract] = value
    return value

make ¶

make(abstract: Abstract, **overrides: Any) -> Any

Resolve abstract to a concrete instance, autowiring its dependencies.

Source code in src/zeython/container.py

def make(self, abstract: Abstract, **overrides: Any) -> Any:
    """Resolve ``abstract`` to a concrete instance, autowiring its dependencies."""
    if abstract in self._instances:
        return self._instances[abstract]

    binding = self._bindings.get(abstract)
    factory = binding.factory if binding else abstract

    if not callable(factory):
        raise BindingResolutionError(
            f"Cannot resolve unbound abstract type {abstract!r}"
        )

    instance = self.call(factory, **overrides)

    if binding is not None and binding.shared:
        self._instances[abstract] = instance

    return instance

call ¶

call(fn: Callable[..., T], **overrides: Any) -> T

Call fn, resolving any missing arguments from the container.

Source code in src/zeython/container.py

def call(self, fn: Callable[..., T], **overrides: Any) -> T:
    """Call ``fn``, resolving any missing arguments from the container."""
    signature = inspect.signature(fn)
    kwargs: dict[str, Any] = {}

    for name, param in signature.parameters.items():
        if name in overrides:
            kwargs[name] = overrides[name]
            continue

        if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
            continue

        annotation = param.annotation

        if annotation is inspect.Parameter.empty:
            if param.default is inspect.Parameter.empty:
                raise BindingResolutionError(
                    f"Cannot resolve parameter '{name}' of {fn!r}: no type hint and no default"
                )
            continue

        if self.has(annotation):
            kwargs[name] = self.make(annotation)
        elif inspect.isclass(annotation) and not _is_builtin_scalar(annotation):
            try:
                kwargs[name] = self.make(annotation)
            except BindingResolutionError:
                if param.default is inspect.Parameter.empty:
                    raise
        elif param.default is inspect.Parameter.empty:
            raise BindingResolutionError(
                f"Cannot resolve parameter '{name}' of {fn!r}: "
                f"no binding registered for {annotation!r}"
            )

    return fn(**kwargs)

flush ¶

flush() -> None

Remove all bindings and cached instances.

Source code in src/zeython/container.py

def flush(self) -> None:
    """Remove all bindings and cached instances."""
    self._bindings.clear()
    self._instances.clear()

config ¶

Environment-driven configuration for Zeython applications.

Config ¶

Config(values: dict[str, Any], base_path: Path)

Layered configuration backed by the environment and .env files.

Values resolve in this order (highest priority first): real process environment variables, then the loaded .env file, then explicit defaults passed to :meth:get.

Source code in src/zeython/config.py

def __init__(self, values: dict[str, Any], base_path: Path) -> None:
    self._values = values
    self.base_path = base_path

get ¶

get(key: str, default: Any = None) -> Any

Dot-path lookup, e.g. config.get("database.url") reads DATABASE_URL.

Source code in src/zeython/config.py

def get(self, key: str, default: Any = None) -> Any:
    """Dot-path lookup, e.g. ``config.get("database.url")`` reads ``DATABASE_URL``."""
    env_key = key.replace(".", "_").upper()
    return self._values.get(env_key, default)

providers ¶

Service providers: the seam where cross-cutting concerns hook into boot.

ServiceProvider ¶

ServiceProvider(app: Application)

Base class for registering and booting application services.

register() runs for every provider before any provider's boot() runs, so bindings you depend on in boot() are guaranteed to exist regardless of registration order.

Source code in src/zeython/providers.py

def __init__(self, app: Application) -> None:
    self.app = app
    self.container = app.container
    self.config = app.config

register ¶

register() -> None

Bind services into the container. Override in subclasses.

Source code in src/zeython/providers.py

def register(self) -> None:
    """Bind services into the container. Override in subclasses."""

boot ¶

boot() -> None

Run once all providers have registered. Override in subclasses.

Source code in src/zeython/providers.py

def boot(self) -> None:
    """Run once all providers have registered. Override in subclasses."""

DatabaseServiceProvider ¶

DatabaseServiceProvider(app: Application)

Bases: ServiceProvider

Wires up the async :class:~zeython.db.Database and its request-scoped session.

DATABASE_POOL_SIZE/DATABASE_MAX_OVERFLOW are passed straight through to SQLAlchemy's connection pool when set -- unset by default, so nothing changes for an in-memory SQLite URL (:memory:), whose default pool doesn't accept them at all. See docs/database.md#connection-pooling.

DATABASE_READ_URL, if set, binds a read replica -- database.read_replica() opens a session against it instead of the primary. See docs/database.md#read-replicas.

Source code in src/zeython/providers.py

def __init__(self, app: Application) -> None:
    self.app = app
    self.container = app.container
    self.config = app.config

RouteServiceProvider ¶

RouteServiceProvider(
    app: Application, modules: tuple[str, ...] = ()
)

Bases: ServiceProvider

Imports route modules for their side effect of registering routes on the app.

Source code in src/zeython/providers.py

def __init__(self, app: Application, modules: tuple[str, ...] = ()) -> None:
    super().__init__(app)
    self.modules = modules

ViewServiceProvider ¶

ViewServiceProvider(app: Application)

Bases: ServiceProvider

Binds a :class:~zeython.views.Views instance for server-rendered HTML.

Looks for templates in resources/views under the app's base path by default; override with VIEWS_PATH in .env or the views.path config key.

Source code in src/zeython/providers.py

def __init__(self, app: Application) -> None:
    self.app = app
    self.container = app.container
    self.config = app.config

CorsServiceProvider ¶

CorsServiceProvider(app: Application)

Bases: ServiceProvider

Opt-in CORS support, configured via .env.

CORS_ORIGINS — comma-separated list of allowed origins (default: none)
CORS_ALLOW_CREDENTIALS — default false
CORS_ALLOW_METHODS — comma-separated, default *
CORS_ALLOW_HEADERS — comma-separated, default *

Source code in src/zeython/providers.py

def __init__(self, app: Application) -> None:
    self.app = app
    self.container = app.container
    self.config = app.config

events ¶

Application-level events: decoupled listeners reacting to a domain event (OrderPlaced, UserRegistered, ...) without the code that raises it needing to know who's listening.

Deliberately separate from :class:zeython.db.Observer -- an Observer reacts to one model's own lifecycle (creating, updated, ...); an event here can be anything your application defines, dispatched from anywhere (a controller, a job, a scheduled task), with any number of independent listeners reacting to it without editing the code that dispatches it. A common pattern is dispatching an event from a model hook (created()) once the write itself is the model's own concern but what happens next (send a receipt, notify a webhook, update a search index) isn't.

EventDispatcher ¶

EventDispatcher(*, container: Container | None = None)

Maps an event type to the listeners registered for it.

Source code in src/zeython/events.py

def __init__(self, *, container: Container | None = None) -> None:
    self._listeners: dict[type, list[Listener]] = defaultdict(list)
    self._container = container

listen ¶

listen(event_type: type[E], listener: Listener) -> None

Register listener to run whenever an instance of event_type is dispatched.

Source code in src/zeython/events.py

def listen(self, event_type: type[E], listener: Listener) -> None:
    """Register ``listener`` to run whenever an instance of ``event_type`` is dispatched."""
    self._listeners[event_type].append(listener)

on ¶

on(event_type: type[E]) -> Callable[[Listener], Listener]

Decorator form of :meth:listen::

@dispatcher.on(OrderPlaced) async def send_receipt(event: OrderPlaced) -> None: ...

Source code in src/zeython/events.py

def on(self, event_type: type[E]) -> Callable[[Listener], Listener]:
    """Decorator form of :meth:`listen`::

        @dispatcher.on(OrderPlaced)
        async def send_receipt(event: OrderPlaced) -> None:
            ...
    """

    def decorator(listener: Listener) -> Listener:
        self.listen(event_type, listener)
        return listener

    return decorator

listeners_for ¶

listeners_for(event_type: type) -> list[Listener]

The listeners currently registered for event_type, in registration order.

Source code in src/zeython/events.py

def listeners_for(self, event_type: type) -> list[Listener]:
    """The listeners currently registered for ``event_type``, in registration order."""
    return list(self._listeners.get(event_type, ()))

dispatch `async` ¶

dispatch(event: object) -> None

Call every listener registered for type(event), in registration order.

A listener's own exception is logged and reported (see :mod:zeython.error_monitoring), not raised -- one broken listener (a bad webhook call, a typo in an audit-log write) shouldn't stop the others from running, the same way a failed Slack notification shouldn't also silently swallow the receipt email.

Source code in src/zeython/events.py

async def dispatch(self, event: object) -> None:
    """Call every listener registered for ``type(event)``, in registration order.

    A listener's own exception is logged and reported (see
    :mod:`zeython.error_monitoring`), not raised -- one broken listener
    (a bad webhook call, a typo in an audit-log write) shouldn't stop
    the others from running, the same way a failed Slack notification
    shouldn't also silently swallow the receipt email.
    """
    for listener in self._listeners.get(type(event), ()):
        try:
            await self._invoke(listener, event)
        except Exception as exc:
            report_exception(exc, listener=getattr(listener, "__qualname__", repr(listener)))
            logger.exception("Event listener %r raised while handling %r", listener, event)

EventServiceProvider ¶

EventServiceProvider(app: Application)

Bases: ServiceProvider

Binds an :class:EventDispatcher into the container.

Register your own listeners by subclassing and overriding boot() (calling super().boot() first, so the dispatcher exists) -- registration happens once, at startup, the same way route modules and other providers wire themselves up::

class AppEventServiceProvider(EventServiceProvider):
    def boot(self) -> None:
        super().boot()
        dispatcher = self.container.make(EventDispatcher)
        dispatcher.listen(OrderPlaced, send_receipt_email)
        dispatcher.listen(OrderPlaced, notify_fulfillment_webhook)

Source code in src/zeython/providers.py

def __init__(self, app: Application) -> None:
    self.app = app
    self.container = app.container
    self.config = app.config

emit `async` ¶

emit(request: Request, event: object) -> None

Dispatch event to every listener registered for its type.

Uses whichever :class:EventDispatcher is bound in the container (see :class:EventServiceProvider). Outside of a request -- a job, a scheduled task, a model hook -- dispatch directly against a resolved dispatcher instead::

await app.container.make(EventDispatcher).dispatch(event)

Source code in src/zeython/events.py

async def emit(request: Request, event: object) -> None:
    """Dispatch ``event`` to every listener registered for its type.

    Uses whichever :class:`EventDispatcher` is bound in the container (see
    :class:`EventServiceProvider`). Outside of a request -- a job, a
    scheduled task, a model hook -- dispatch directly against a resolved
    dispatcher instead::

        await app.container.make(EventDispatcher).dispatch(event)
    """
    dispatcher: EventDispatcher = request.app.state.container.make(EventDispatcher)
    await dispatcher.dispatch(event)

routing ¶

Ergonomic routing built on top of Starlette's proven route matching.

Controller ¶

Marker base class for class-based controllers used with :meth:Router.resource.

Router ¶

Router(prefix: str = '')

Collects routes and exposes Laravel/FastAPI-style decorator sugar.

A Router compiles down to a plain list of Starlette BaseRoute objects, so nesting via :meth:include is just a Mount and gets the same battle-tested path matching as everything else built on Starlette.

Source code in src/zeython/routing.py

def __init__(self, prefix: str = "") -> None:
    self.prefix = prefix.rstrip("/")
    self.routes: list[BaseRoute] = []

websocket ¶

websocket(
    path: str, *, name: str | None = None
) -> Callable[[Endpoint], Endpoint]

Register a WebSocket handler: async def handler(websocket: WebSocket) -> None.

See :mod:zeython.websockets and docs/websockets.md.

Source code in src/zeython/routing.py

def websocket(self, path: str, *, name: str | None = None) -> Callable[[Endpoint], Endpoint]:
    """Register a WebSocket handler: ``async def handler(websocket: WebSocket) -> None``.

    See :mod:`zeython.websockets` and docs/websockets.md.
    """

    def decorator(endpoint: Endpoint) -> Endpoint:
        self.routes.append(WebSocketRoute(self._full_path(path), endpoint, name=name or endpoint.__name__))
        return endpoint

    return decorator

include ¶

include(router: Router, *, prefix: str = '') -> None

Mount another router's routes under an optional additional prefix.

Source code in src/zeython/routing.py

def include(self, router: Router, *, prefix: str = "") -> None:
    """Mount another router's routes under an optional additional prefix."""
    self.routes.append(Mount(prefix or "/", routes=router.routes))

mount ¶

mount(
    path: str, app: Any, *, name: str | None = None
) -> None

Mount an arbitrary ASGI app (e.g. starlette.staticfiles.StaticFiles) at a path prefix.

Source code in src/zeython/routing.py

def mount(self, path: str, app: Any, *, name: str | None = None) -> None:
    """Mount an arbitrary ASGI app (e.g. ``starlette.staticfiles.StaticFiles``) at a path prefix."""
    self.routes.append(Mount(path, app=app, name=name))

resource ¶

resource(
    path: str,
    controller_cls: type[Controller],
    *,
    only: Iterable[str] | None = None,
) -> None

Register RESTful CRUD routes bound to a controller's methods.

Maps: index->GET path, store->POST path, show->GET path/{id}, update->PUT/PATCH path/{id}, destroy->DELETE path/{id}.

Source code in src/zeython/routing.py

def resource(self, path: str, controller_cls: type[Controller], *, only: Iterable[str] | None = None) -> None:
    """Register RESTful CRUD routes bound to a controller's methods.

    Maps: index->GET path, store->POST path, show->GET path/{id},
    update->PUT/PATCH path/{id}, destroy->DELETE path/{id}.
    """
    controller = controller_cls()
    action_map: dict[str, tuple[str, str]] = {
        "index": ("GET", ""),
        "store": ("POST", ""),
        "show": ("GET", "/{id}"),
        "update": ("PUT", "/{id}"),
        "destroy": ("DELETE", "/{id}"),
    }
    allowed = set(only) if only is not None else set(action_map)

    for action, (method, suffix) in action_map.items():
        if action not in allowed or not hasattr(controller, action):
            continue
        handler = getattr(controller, action)
        route_path = self._full_path(f"{path.rstrip('/')}{suffix}")
        self.routes.append(
            Route(route_path, handler, methods=[method], name=f"{path.strip('/')}.{action}")
        )

views ¶

Server-rendered HTML views, by convention read from resources/views/.

Views ¶

Views(directory: str | Path)

Thin wrapper around Starlette's Jinja2 integration.

Bound into the container by :class:~zeython.providers.ViewServiceProvider under the resources/views directory by default. Use the module-level :func:render helper from inside a controller/handler for Flask-style ergonomics.

Source code in src/zeython/views.py

def __init__(self, directory: str | Path) -> None:
    self.directory = Path(directory)
    self._templates = Jinja2Templates(directory=str(self.directory))

render ¶

render(
    request: Request,
    name: str,
    context: dict[str, Any] | None = None,
    *,
    status_code: int = 200,
) -> HTMLResponse

Render name using the application's registered :class:Views instance.

Usage inside a controller::

from zeython.views import render

async def show(self, request):
    return render(request, "posts/show.html", {"post": post})

Source code in src/zeython/views.py

def render(
    request: Request,
    name: str,
    context: dict[str, Any] | None = None,
    *,
    status_code: int = 200,
) -> HTMLResponse:
    """Render ``name`` using the application's registered :class:`Views` instance.

    Usage inside a controller::

        from zeython.views import render

        async def show(self, request):
            return render(request, "posts/show.html", {"post": post})
    """
    from zeython.container import Container

    container: Container = request.app.state.container
    views = container.make(Views)
    return views.render(request, name, context, status_code=status_code)

exceptions ¶

HTTP-aware exception hierarchy with a default JSON error handler.

HTTPException ¶

HTTPException(
    detail: str | None = None,
    *,
    headers: dict[str, str] | None = None,
)

Bases: Exception

Base class for exceptions that should be rendered as HTTP responses.

Source code in src/zeython/exceptions.py

def __init__(self, detail: str | None = None, *, headers: dict[str, str] | None = None) -> None:
    self.detail = detail or self.default_detail
    self.headers = headers or {}
    super().__init__(self.detail)

validation ¶

Declarative validation rules for :class:zeython.db.Model fields.

Rule ¶

Rule(check: Check, message: str)

A single named validation rule with a default error message.

Source code in src/zeython/validation.py

def __init__(self, check: Check, message: str) -> None:
    self.check = check
    self.message = message

validate ¶

validate(
    data: dict[str, Any], rules: dict[str, list[Rule]]
) -> dict[str, list[str]]

Run declarative rules against a plain dict -- the same rule sets you'd write for :attr:zeython.db.Model.__rules__, applied to a request payload, query params, or any other dict that isn't (or isn't yet) a model instance. Does not raise; raise ValidationException(errors) yourself if that's what you want when errors is non-empty.

Model.validate() is this function applied to a model instance's own field values -- kept in sync with it deliberately, so a rule set means the same thing whether it's checked against a model or a plain dict.

Source code in src/zeython/validation.py

def validate(data: dict[str, Any], rules: dict[str, list[Rule]]) -> dict[str, list[str]]:
    """Run declarative rules against a plain dict -- the same rule sets you'd
    write for :attr:`zeython.db.Model.__rules__`, applied to a request
    payload, query params, or any other dict that isn't (or isn't yet) a
    model instance. Does not raise; raise ``ValidationException(errors)``
    yourself if that's what you want when ``errors`` is non-empty.

    ``Model.validate()`` is this function applied to a model instance's own
    field values -- kept in sync with it deliberately, so a rule set means
    the same thing whether it's checked against a model or a plain dict.
    """
    errors: dict[str, list[str]] = {}
    for field, field_rules in rules.items():
        value = data.get(field)
        for rule in field_rules:
            if not rule(value):
                errors.setdefault(field, []).append(rule.message)
    return errors

Core¶

application ¶

Application ¶

providers property ¶

run ¶

container ¶

BindingResolutionError ¶

Container ¶

bind ¶

singleton ¶

instance ¶

make ¶

call ¶

flush ¶

config ¶

Config ¶

get ¶

providers ¶

ServiceProvider ¶

register ¶

boot ¶

DatabaseServiceProvider ¶

RouteServiceProvider ¶

ViewServiceProvider ¶

CorsServiceProvider ¶

events ¶

EventDispatcher ¶

listen ¶

on ¶

listeners_for ¶

dispatch async ¶

EventServiceProvider ¶

emit async ¶

routing ¶

Controller ¶

Router ¶

websocket ¶

include ¶

mount ¶

resource ¶

views ¶

Views ¶

render ¶

exceptions ¶

HTTPException ¶

validation ¶

Rule ¶

validate ¶

providers `property` ¶

dispatch `async` ¶

emit `async` ¶