Skip to content

API reference#

Generated directly from Hayavega's own docstrings and type hints (the same ones mypy checks in CI) — this page can't drift from the actual public API the way a hand-maintained second copy would.

Routing#

path #

path(pattern: str, view: Callable, methods: tuple[str, ...] = ('GET',)) -> Route

Register a route. The root route is written as an empty string.

Source code in src/hayavega/routing.py
def path(pattern: str, view: Callable, methods: tuple[str, ...] = ("GET",)) -> Route:
    """Register a route. The root route is written as an empty string."""
    return Route(pattern.strip("/"), view, tuple(method.upper() for method in methods))

hayavega.rest.api_route #

api_route(pattern: str, view: Callable, methods: Iterable[str] = ('GET',), *, secured: bool = False, request_schema: dict | None = None, responses: dict[int, dict] | None = None) -> Route

Create a JSON endpoint. Views may return a dict/list or a Response.

secured, request_schema, and responses describe the endpoint for the generated OpenAPI document at /openapi.json. secured marks the endpoint as requiring a Bearer session token. request_schema is a JSON Schema dict for the request body. responses maps a status code to {"description": str, "schema": dict} (schema is optional). None of these change runtime behavior -- omit them and Hayavega documents the endpoint with its previous generic fallback schema.

view may be async def when deployed under hayavega.asgi (see DEPLOYMENT.md): the wrapper stays a coroutine function itself so the ASGI runtime awaits it directly rather than running it in a thread.

Source code in src/hayavega/rest.py
def api_route(
    pattern: str,
    view: Callable,
    methods: Iterable[str] = ("GET",),
    *,
    secured: bool = False,
    request_schema: dict | None = None,
    responses: dict[int, dict] | None = None,
) -> Route:
    """Create a JSON endpoint. Views may return a dict/list or a Response.

    ``secured``, ``request_schema``, and ``responses`` describe the endpoint
    for the generated OpenAPI document at ``/openapi.json``. ``secured``
    marks the endpoint as requiring a Bearer session token. ``request_schema``
    is a JSON Schema dict for the request body. ``responses`` maps a status
    code to ``{"description": str, "schema": dict}`` (``schema`` is optional).
    None of these change runtime behavior -- omit them and Hayavega documents
    the endpoint with its previous generic fallback schema.

    ``view`` may be ``async def`` when deployed under ``hayavega.asgi`` (see
    ``DEPLOYMENT.md``): the wrapper stays a coroutine function itself so the
    ASGI runtime awaits it directly rather than running it in a thread.
    """

    if inspect.iscoroutinefunction(view):

        async def endpoint(request):
            result = await view(request)
            return result if isinstance(result, Response) else json_response(result)
    else:

        def endpoint(request):  # type: ignore[misc]
            result = view(request)
            return result if isinstance(result, Response) else json_response(result)

    route = path(pattern, endpoint, tuple(methods))
    return Route(route.pattern, route.view, route.methods, is_api=True, secured=secured, request_schema=request_schema, responses=responses)

Request & Response#

hayavega.request.Request dataclass #

Request(method: str, path: str, query_params: dict[str, list[str]], body: bytes, headers: dict[str, str], remote_addr: str = '', is_secure: bool = False, raw_query: str = '')

cookies property #

cookies: dict[str, str]

Return safely parsed request cookies.

files #

files() -> dict[str, UploadedFile]

Parse a multipart/form-data body into uploaded files, keyed by field name (the first file, if more than one was submitted under the same field name). Returns {} for any other content type. Requires python-multipart -- see MultipartUnavailableError.

Source code in src/hayavega/request.py
def files(self) -> dict[str, UploadedFile]:
    """Parse a `multipart/form-data` body into uploaded files, keyed
    by field name (the first file, if more than one was submitted
    under the same field name). Returns `{}` for any other content
    type. Requires `python-multipart` -- see `MultipartUnavailableError`.
    """
    content_type = self.headers.get("Content-Type", "")
    if not content_type.startswith("multipart/form-data"):
        return {}
    _, files = _parse_multipart(content_type, self.body)
    return {name: items[0] for name, items in files.items()}

form #

form() -> dict[str, str]

Parse a form body into plain fields -- last value wins per field. Handlesapplication/x-www-form-urlencoded(the default) andmultipart/form-data(needspython-multipart; seefiles()`).

Source code in src/hayavega/request.py
def form(self) -> dict[str, str]:
    """Parse a form body into plain fields -- `last value wins per
    field. Handles `application/x-www-form-urlencoded` (the
    default) and `multipart/form-data` (needs `python-multipart`;
    see `files()`)."""
    content_type = self.headers.get("Content-Type", "")
    if content_type.startswith("multipart/form-data"):
        fields, _ = _parse_multipart(content_type, self.body)
        return fields
    return {name: values[-1] for name, values in parse_qs(self.body.decode("utf-8")).items()}

hayavega.response.Response dataclass #

Response(body: str | bytes, status: int = 200, content_type: str = 'text/html; charset=utf-8', headers: tuple[tuple[str, str], ...] = ())

hayavega.response.json_response #

json_response(data, status: int = 200, headers: tuple[tuple[str, str], ...] = ()) -> Response
Source code in src/hayavega/response.py
def json_response(data, status: int = 200, headers: tuple[tuple[str, str], ...] = ()) -> Response:
    return Response(
        msgspec.json.encode(data),
        status=status,
        content_type="application/json; charset=utf-8",
        headers=headers,
    )

Models#

hayavega.models.Entity #

Entity(**values)
Source code in src/hayavega/models.py
def __init__(self, **values):
    self._values = {}
    for name, column in self._columns.items():
        if name in values:
            setattr(self, name, values[name])
        elif callable(column.default):
            setattr(self, name, column.default())
        elif column.default is not None:
            setattr(self, name, column.default)
        else:
            self._values[name] = None

collection classmethod #

collection(database, *, signals: SignalBus | None = None) -> Collection

signals, when given a SignalBus, publishes "<entity>.created"/"<entity>.updated"/"<entity>.deleted" from append/revise/erase -- opt-in, matching this codebase's "opt-in, not on by default" policy elsewhere (Hardening), so every existing caller's behavior is unchanged when omitted. See docs/topics/signals.md.

Source code in src/hayavega/models.py
@classmethod
def collection(cls, database, *, signals: SignalBus | None = None) -> Collection:
    """`signals`, when given a `SignalBus`, publishes
    `"<entity>.created"`/`"<entity>.updated"`/`"<entity>.deleted"`
    from `append`/`revise`/`erase` -- opt-in, matching this
    codebase's "opt-in, not on by default" policy elsewhere
    (`Hardening`), so every existing caller's behavior is unchanged
    when omitted. See `docs/topics/signals.md`."""
    return Collection(cls, database, signals=signals)

hayavega.models.Collection #

Collection(entity: type[Entity], database, *, signals: SignalBus | None = None)
Source code in src/hayavega/models.py
def __init__(self, entity: type[Entity], database, *, signals: SignalBus | None = None):
    self.entity, self.database, self.signals = entity, database, signals
    _identifier(entity.__table__)

prepare #

prepare() -> None

Ensure the entity's table exists.

Every fetch/total/revise/erase/append call goes through here, so the "already prepared" case -- true for nearly every call after the first -- must not pay for a second connection and a CREATE TABLE IF NOT EXISTS round trip. Cached per (table, database) for the life of the process instead: a table's schema doesn't change without a migration, which is a separate, deliberate operation this cache has no effect on.

Source code in src/hayavega/models.py
def prepare(self) -> None:
    """Ensure the entity's table exists.

    Every ``fetch``/``total``/``revise``/``erase``/``append`` call goes
    through here, so the "already prepared" case -- true for nearly
    every call after the first -- must not pay for a second connection
    and a ``CREATE TABLE IF NOT EXISTS`` round trip. Cached per
    (table, database) for the life of the process instead: a table's
    schema doesn't change without a migration, which is a separate,
    deliberate operation this cache has no effect on.
    """
    key = self._prepared_key()
    if key in _PREPARED_TABLES:
        return
    connection = connect(self.database)
    try:
        definitions = []
        for name, column in self.entity._columns.items():
            definition = f"{_identifier(name)} {column.sql_type(connection.engine)}"
            if not column.primary:
                if not column.nullable:
                    definition += " NOT NULL"
                if column.unique:
                    definition += " UNIQUE"
            definitions.append(definition)
        connection.execute(f"CREATE TABLE IF NOT EXISTS {_identifier(self.entity.__table__)} ({', '.join(definitions)})")
        connection.commit()
    finally:
        connection.close()
    _PREPARED_TABLES.add(key)

hayavega.models.Selection #

Selection(collection: Collection, clauses=(), ordering=(), limit: int | None = None, offset: int = 0, search_clause: tuple[tuple[str, ...], str] | None = None)
Source code in src/hayavega/models.py
def __init__(self, collection: Collection, clauses=(), ordering=(), limit: int | None = None, offset: int = 0, search_clause: tuple[tuple[str, ...], str] | None = None):
    self.collection, self.clauses, self.ordering, self.limit, self.offset, self._search_clause = collection, tuple(clauses), tuple(ordering), limit, offset, search_clause

search #

search(*fields: str, query: str) -> Selection

OR a LIKE match across fields -- a narrow, single-purpose capability for admin's search_fields, not a general Q()/OR composition system. Combines with match()'s clauses via AND (seek(published=True).search("title", "body", query="rust") means both). An empty query is a no-op, so a blank search box doesn't filter anything out.

Source code in src/hayavega/models.py
def search(self, *fields: str, query: str) -> Selection:
    """OR a `LIKE` match across `fields` -- a narrow, single-purpose
    capability for admin's `search_fields`, not a general `Q()`/OR
    composition system. Combines with `match()`'s clauses via AND
    (`seek(published=True).search("title", "body", query="rust")`
    means *both*). An empty `query` is a no-op, so a blank search
    box doesn't filter anything out.
    """
    for field in fields:
        _identifier(field)
        if field not in self.collection.entity._columns:
            raise EntityError(f"Unknown column: {field}")
    if not query:
        return self
    return self._copy(search_clause=(fields, query))

values #

values(*fields: str) -> list[dict]

Fetch rows as plain dicts instead of Entity instances.

fetch() runs every column through its Column.coerce() and builds a real Entity object per row -- the right default when the caller wants entity behavior (further selection, .revise(), type coercion). Read-only rendering (a list page, a JSON API response) often immediately discards that object for a plain dict of a few fields, paying for hydration it never uses. values() skips it: pass specific column names to select only those, or none for every column.

Source code in src/hayavega/models.py
def values(self, *fields: str) -> list[dict]:
    """Fetch rows as plain ``dict``s instead of ``Entity`` instances.

    `fetch()` runs every column through its `Column.coerce()` and
    builds a real `Entity` object per row -- the right default when the
    caller wants entity behavior (further selection, `.revise()`, type
    coercion). Read-only rendering (a list page, a JSON API response)
    often immediately discards that object for a plain dict of a few
    fields, paying for hydration it never uses. `values()` skips it:
    pass specific column names to select only those, or none for every
    column.
    """
    for field in fields:
        _identifier(field)
        if field not in self.collection.entity._columns:
            raise EntityError(f"Unknown column: {field}")
    self.collection.prepare()
    connection = connect_readonly(self.collection.database)
    try:
        columns = ", ".join(_identifier(field) for field in fields) if fields else "*"
        rows = connection.execute(*self._sql(columns)).fetchall()
        return [dict(row) for row in rows]
    finally:
        connection.close()

Database#

hayavega.db.connect #

connect(database: str | Path | dict) -> Connection
Source code in src/hayavega/db.py
def connect(database: str | Path | dict) -> Connection:
    config = normalize_config(database)
    if _pool_enabled:
        pool = _pool_local.__dict__.setdefault("connections", {})
        key = _pool_key(config)
        cached = pool.get(key)
        if cached is not None:
            return cached
        connection = _open(config)
        connection.pooled = True
        pool[key] = connection
        return connection
    return _open(config)

hayavega.db.connect_readonly #

connect_readonly(database: str | Path | dict) -> Connection

Open a connection scoped to reads only.

For SQLite, this opens the file with the mode=ro URI parameter instead of a plain read-write connection. Measured (benchmarks/ sqlite_readonly_reads.py) at a real, consistent win on this project's actual default (journal_mode=DELETE, untouched): ~25-30% faster per call single-threaded and ~18-30% higher aggregate throughput under 16-thread contention, on Windows -- a read-only open skips the reserved/pending-lock bookkeeping SQLite otherwise does on every statement to guard against a writer, which a connection that will only ever SELECT never needs.

Never pooled by default -- opened and closed fresh on every call, which does not reproduce the many-threads-holding-a-connection-open contention that made pooling regress the WSGI path under enable_connection_pooling(). Pass through enable_readonly_connection_pooling() to additionally reuse one read-only connection per thread; see that function's docstring for why pooling read-only connections was found not to carry the write-pooling regression.

Only used internally by Selection.fetch()/values()/total() -- pure-read call sites that never call .commit(). Collection.append()/ Selection.revise()/erase() are unaffected and keep using connect().

Falls back to a normal connect() if the read-only open fails (for example, the database file does not exist yet) rather than raising -- in practice this never happens on the read path, since every read call site calls Collection.prepare() first, which always creates the file via a normal read-write connect() if it is missing.

Non-SQLite engines have no equivalent parameter tested here and fall back to connect() unchanged.

Source code in src/hayavega/db.py
def connect_readonly(database: str | Path | dict) -> Connection:
    """Open a connection scoped to reads only.

    For SQLite, this opens the file with the `mode=ro` URI parameter instead
    of a plain read-write connection. Measured (`benchmarks/
    sqlite_readonly_reads.py`) at a real, consistent win on this project's
    actual default (`journal_mode=DELETE`, untouched): ~25-30% faster per
    call single-threaded and ~18-30% higher aggregate throughput under
    16-thread contention, on Windows -- a read-only open skips the
    reserved/pending-lock bookkeeping SQLite otherwise does on every
    statement to guard against a writer, which a connection that will only
    ever `SELECT` never needs.

    Never pooled by default -- opened and closed fresh on every call, which
    does not reproduce the many-threads-holding-a-connection-open
    contention that made pooling regress the WSGI path under
    `enable_connection_pooling()`. Pass through `enable_readonly_connection_pooling()`
    to additionally reuse one read-only connection per thread; see that
    function's docstring for why pooling *read-only* connections was found
    not to carry the write-pooling regression.

    Only used internally by `Selection.fetch()`/`values()`/`total()` --
    pure-read call sites that never call `.commit()`. `Collection.append()`/
    `Selection.revise()`/`erase()` are unaffected and keep using `connect()`.

    Falls back to a normal `connect()` if the read-only open fails (for
    example, the database file does not exist yet) rather than raising --
    in practice this never happens on the read path, since every read call
    site calls `Collection.prepare()` first, which always creates the file
    via a normal read-write `connect()` if it is missing.

    Non-SQLite engines have no equivalent parameter tested here and fall
    back to `connect()` unchanged.
    """
    config = normalize_config(database)
    engine = str(config.get("ENGINE", "sqlite3")).lower()
    if engine not in {"sqlite", "sqlite3"}:
        return connect(database)
    if _readonly_pool_enabled:
        pool = _readonly_pool_local.__dict__.setdefault("connections", {})
        key = _pool_key(config)
        cached = pool.get(key)
        if cached is not None:
            return cached
        connection = _open_readonly(config)
        if connection is None:
            return connect(database)
        connection.pooled = True
        pool[key] = connection
        return connection
    connection = _open_readonly(config)
    return connection if connection is not None else connect(database)

hayavega.db.enable_connection_pooling #

enable_connection_pooling() -> None

Reuse one connection per (thread, database config) instead of opening a new one on every connect() call.

Off by default; every existing caller of connect() -- models.py, auth.py, admin.py, migrations.py, production.py's readiness check -- needs no change to benefit, since this only changes what connect() and Connection.close() do internally, not their contracts.

Scoped deliberately to match Hayavega's WSGI/ASGI concurrency model: one connection per worker thread, matching the one-request-per-thread (WSGI) or one-coroutine-at-a-time-per-thread (ASGI sync views run on create_asgi_application's dedicated max_sync_workers-sized thread pool) model this framework already uses, rather than a general bounded pool shared across threads with checkout/checkin queuing. That keeps the mechanism simple and avoids the cross-thread connection-sharing hazards DB-API connections generally aren't safe for (SQLite enforces this directly: a connection created with the default check_same_thread=True cannot be used from another thread at all).

A pooled MySQL/PostgreSQL connection that has sat idle long enough for the server to drop it (MySQL's wait_timeout, a network blip) is transparently reconnected: Connection.execute() recognizes the driver-specific "connection is dead" error, reopens the underlying connection, and retries the same statement once -- see _is_stale_connection_error/Connection._reconnect. This does not affect SQLite, which has no server-side idle timeout and is never eligible for this path.

Benchmark this for your own deployment before enabling it -- it is not a universal win. Measured against benchmarks/'s SQLite-backed db_list/template_render scenarios: a real, substantial improvement under hayavega.asgi + uvicorn (db_list +15-30%, and it is what pushed template_render past FastAPI for the first time), but a consistent ~10-15% regression under the WSGI path (waitress, 16 threads) -- confirmed at both the full HTTP level (oha, multiple repeated runs) and in an isolated multi-threaded microbenchmark with no framework code involved at all, so this is a property of many threads holding a persistent SQLite connection open concurrently under contention, not a bug in this pool's own bookkeeping. benchmarks/hayavega_app/app.py (WSGI) deliberately leaves this disabled for exactly that reason; benchmarks/hayavega_asgi_app/app.py (ASGI) deliberately enables it. The mechanism is identical in both -- only which concurrency model ends up holding several connections open at once under load differs.

Source code in src/hayavega/db.py
def enable_connection_pooling() -> None:
    """Reuse one connection per (thread, database config) instead of opening
    a new one on every `connect()` call.

    Off by default; every existing caller of `connect()` -- `models.py`,
    `auth.py`, `admin.py`, `migrations.py`, `production.py`'s readiness
    check -- needs no change to benefit, since this only changes what
    `connect()` and `Connection.close()` do internally, not their contracts.

    Scoped deliberately to match Hayavega's WSGI/ASGI concurrency model: one
    connection per worker thread, matching the one-request-per-thread (WSGI)
    or one-coroutine-at-a-time-per-thread (ASGI sync views run on
    `create_asgi_application`'s dedicated `max_sync_workers`-sized thread
    pool) model this framework already uses, rather than a general bounded
    pool shared across threads with checkout/checkin
    queuing. That keeps the mechanism simple and avoids the cross-thread
    connection-sharing hazards DB-API connections generally aren't safe for
    (SQLite enforces this directly: a connection created with the default
    `check_same_thread=True` cannot be used from another thread at all).

    A pooled MySQL/PostgreSQL connection that has sat idle long enough for
    the server to drop it (MySQL's `wait_timeout`, a network blip) is
    transparently reconnected: `Connection.execute()` recognizes the
    driver-specific "connection is dead" error, reopens the underlying
    connection, and retries the same statement once -- see
    `_is_stale_connection_error`/`Connection._reconnect`. This does not
    affect SQLite, which has no server-side idle timeout and is never
    eligible for this path.

    **Benchmark this for your own deployment before enabling it -- it is
    not a universal win.** Measured against `benchmarks/`'s SQLite-backed
    `db_list`/`template_render` scenarios: a real, substantial improvement
    under `hayavega.asgi` + uvicorn (`db_list` +15-30%, and it is what pushed
    `template_render` past FastAPI for the first time), but a consistent
    ~10-15% *regression* under the WSGI path (`waitress`, 16 threads) --
    confirmed at both the full HTTP level (`oha`, multiple repeated runs)
    and in an isolated multi-threaded microbenchmark with no framework code
    involved at all, so this is a property of many threads holding a
    persistent SQLite connection open concurrently under contention, not a
    bug in this pool's own bookkeeping. `benchmarks/hayavega_app/app.py`
    (WSGI) deliberately leaves this disabled for exactly that reason;
    `benchmarks/hayavega_asgi_app/app.py` (ASGI) deliberately enables it.
    The mechanism is identical in both -- only which concurrency model ends
    up holding several connections open at once under load differs.
    """
    global _pool_enabled
    _pool_enabled = True

hayavega.db.enable_readonly_connection_pooling #

enable_readonly_connection_pooling() -> None

Reuse one read-only connection per (thread, database config) instead of opening a fresh one on every connect_readonly() call.

Independent of enable_connection_pooling() -- that function pools read-write connections and, measured against this project's own benchmarks/, regresses the WSGI path by ~10-15% (many threads holding a read-write connection open contends under SQLite on this environment; see its own docstring for the full result). A read-only connection takes no reserved/pending locks at all, so holding one open per thread was hypothesized not to reproduce that contention -- and an isolated 16-thread microbenchmark confirmed the hypothesis at that layer: pooling read-only connections beat both a fresh read-only connection per call and a pooled read-write connection (which reproduced the known regression, as a same-run sanity check) by ~27-43% across two runs. Only affects connect_readonly() -- the write path (connect(), and therefore Collection.append()/ Selection.revise()/erase()) is completely unaffected by this flag and stays governed by enable_connection_pooling() alone.

That isolated win did not survive to the HTTP level -- measured, not assumed. Four paired oha -z 10s -c 30 runs against benchmarks/hayavega_app/app.py under real waitress, with and without this enabled: db_list averaged 811 req/s pooled vs. 840 req/s unpooled; template_render averaged 708 req/s pooled vs. 735 req/s unpooled -- both within this scenario's established run-to-run noise band (up to ~17%, see INDUSTRY_LEADERSHIP_ROADMAP.md's template_render re-measurement), so neither a real win nor a real loss, not the substantial win the isolated microbenchmark predicted. benchmarks/hayavega_app/app.py (WSGI) deliberately leaves this disabled for exactly that reason: the sub-millisecond connection-layer saving this recovers is proportionally invisible against a request whose own sqlite3.connect()+.execute() already costs ~35-45ms end to end on this environment -- the same "environment ceiling" finding every other connection-layer optimization in this document has run into. Kept available and tested (not reverted) since it is a real, correct capability that may still help a different deployment's workload shape; just not proven to help this one.

Benchmark your own deployment before enabling -- disclosed exactly like every other tunable in this module. A pooled read-only SQLite connection reads a point-in-time snapshot of the schema and any long-lived cached state the driver keeps; this framework's own Collection.prepare() cache already assumes a table's schema never changes without a migration, so this carries no additional risk beyond what enable_connection_pooling() already documents for MySQL/ PostgreSQL connections going stale after a long idle period.

Source code in src/hayavega/db.py
def enable_readonly_connection_pooling() -> None:
    """Reuse one read-only connection per (thread, database config) instead
    of opening a fresh one on every `connect_readonly()` call.

    Independent of `enable_connection_pooling()` -- that function pools
    read-write connections and, measured against this project's own
    `benchmarks/`, regresses the WSGI path by ~10-15% (many threads holding
    a *read-write* connection open contends under SQLite on this
    environment; see its own docstring for the full result). A read-only
    connection takes no reserved/pending locks at all, so holding one open
    per thread was hypothesized not to reproduce that contention -- and
    an isolated 16-thread microbenchmark confirmed the hypothesis at that
    layer: pooling read-only connections beat both a fresh read-only
    connection per call and a pooled *read-write* connection (which
    reproduced the known regression, as a same-run sanity check) by
    ~27-43% across two runs. Only affects `connect_readonly()` -- the write
    path (`connect()`, and therefore `Collection.append()`/
    `Selection.revise()`/`erase()`) is completely unaffected by this flag
    and stays governed by `enable_connection_pooling()` alone.

    **That isolated win did not survive to the HTTP level -- measured, not
    assumed.** Four paired `oha -z 10s -c 30` runs against
    `benchmarks/hayavega_app/app.py` under real `waitress`, with and without
    this enabled: `db_list` averaged 811 req/s pooled vs. 840 req/s
    unpooled; `template_render` averaged 708 req/s pooled vs. 735 req/s
    unpooled -- both within this scenario's established run-to-run noise
    band (up to ~17%, see `INDUSTRY_LEADERSHIP_ROADMAP.md`'s
    `template_render` re-measurement), so neither a real win nor a real
    loss, not the substantial win the isolated microbenchmark predicted.
    `benchmarks/hayavega_app/app.py` (WSGI) deliberately leaves this
    disabled for exactly that reason: the sub-millisecond connection-layer
    saving this recovers is proportionally invisible against a request
    whose own `sqlite3.connect()`+`.execute()` already costs ~35-45ms
    end to end on this environment -- the same "environment ceiling"
    finding every other connection-layer optimization in this document has
    run into. Kept available and tested (not reverted) since it is a real,
    correct capability that may still help a different deployment's
    workload shape; just not proven to help this one.

    **Benchmark your own deployment before enabling -- disclosed exactly
    like every other tunable in this module.** A pooled read-only SQLite
    connection reads a point-in-time snapshot of the schema and any
    long-lived cached state the driver keeps; this framework's own
    `Collection.prepare()` cache already assumes a table's schema never
    changes without a migration, so this carries no additional risk beyond
    what `enable_connection_pooling()` already documents for MySQL/
    PostgreSQL connections going stale after a long idle period.
    """
    global _readonly_pool_enabled
    _readonly_pool_enabled = True

Authentication#

hayavega.auth.create_user #

create_user(database_path: str | Path | dict, username: str, email: str, password: str, *, is_superuser: bool = False, validators: Iterable[Callable[[str], None]] | None = None) -> User

validators, when given, replaces the default [validate_password] entirely -- each is called as validator(password) and must raise ValueError to reject it; the first rejection wins. Omit it for the default policy.

Source code in src/hayavega/auth.py
def create_user(database_path: str | Path | dict, username: str, email: str, password: str, *, is_superuser: bool = False, validators: Iterable[Callable[[str], None]] | None = None) -> User:
    """`validators`, when given, replaces the default `[validate_password]`
    entirely -- each is called as `validator(password)` and must raise
    `ValueError` to reject it; the first rejection wins. Omit it for the
    default policy."""
    username, email = username.strip(), email.strip().lower()
    if not username or len(username) > 150:
        raise ValueError("Username is required and must be 150 characters or fewer.")
    if not _EMAIL_PATTERN.fullmatch(email):
        raise ValueError("A valid email address is required.")
    _run_validators(password, validators)
    initialize_auth_db(database_path)
    connection = _connect(database_path)
    try:
        insert = "INSERT INTO hayavega_users (username, email, password_hash, is_superuser) VALUES (?, ?, ?, ?)"
        cursor = connection.execute(
            insert + (" RETURNING id" if connection.engine == "postgresql" else ""),
            (username, email, hash_password(password), int(is_superuser)),
        )
        connection.commit()
        identifier = cursor.fetchone()["id"] if connection.engine == "postgresql" else cursor.lastrowid
        user = User(identifier, username, email, is_superuser, True)
        record_audit_event(database_path, "auth.user_created", actor_id=user.id, metadata={"is_superuser": is_superuser})
        return user
    except sqlite3.IntegrityError as error:
        raise UserAlreadyExistsError("A user with that username or email already exists.") from error
    finally:
        connection.close()

hayavega.auth.authenticate #

authenticate(database_path: str | Path | dict, username: str, password: str, totp_code: str | None = None, *, ip_address: str | None = None, user_agent: str | None = None, secret_key: str | None = None) -> User | None

Verify credentials and record the attempt, including the caller-supplied IP address and user agent, if given.

secret_key is only needed when the user has TOTP enabled and their secret was stored encrypted (see enable_totp) -- pass the same Settings.secret_key used to enable it, or the TOTP check fails closed rather than silently skipping it.

Source code in src/hayavega/auth.py
def authenticate(database_path: str | Path | dict, username: str, password: str, totp_code: str | None = None, *, ip_address: str | None = None, user_agent: str | None = None, secret_key: str | None = None) -> User | None:
    """Verify credentials and record the attempt, including the caller-supplied IP address and user agent, if given.

    `secret_key` is only needed when the user has TOTP enabled and their
    secret was stored encrypted (see `enable_totp`) -- pass the same
    `Settings.secret_key` used to enable it, or the TOTP check fails
    closed rather than silently skipping it.
    """
    initialize_auth_db(database_path)
    now = int(time.time())
    context = _login_context(ip_address, user_agent)
    connection = _connect(database_path)
    try:
        row = connection.execute("SELECT * FROM hayavega_users WHERE username = ?", (username,)).fetchone()
        if row is None or not row["is_active"]:
            record_audit_event(database_path, "auth.login_failed", metadata={"reason": "invalid_credentials", **context})
            return None
        if row["locked_until"] and row["locked_until"] > now:
            record_audit_event(database_path, "auth.login_failed", actor_id=row["id"], metadata={"reason": "locked", **context})
            return None
        password_valid = verify_password(password, row["password_hash"])
        totp_secret = _resolve_totp_secret(row["totp_secret"], secret_key) if row["totp_enabled"] else None
        totp_valid = not row["totp_enabled"] or (totp_code is not None and totp_secret is not None and verify_totp(totp_secret, totp_code, now))
        if not password_valid or not totp_valid:
            _record_failure(connection, row, now)
            record_audit_event(database_path, "auth.login_failed", actor_id=row["id"], metadata={"reason": "invalid_credentials", **context})
            return None
        connection.execute("UPDATE hayavega_users SET failed_attempts = 0, locked_until = NULL WHERE id = ?", (row["id"],))
        if row["password_hash"].startswith("pbkdf2_sha256$"):
            connection.execute("UPDATE hayavega_users SET password_hash = ? WHERE id = ?", (hash_password(password), row["id"]))
        connection.commit()
        user = _user_from_row(row)
        record_audit_event(database_path, "auth.login_succeeded", actor_id=user.id, metadata=context)
        return user
    finally:
        connection.close()

hayavega.auth.create_session #

create_session(database_path: str | Path | dict, user: User, duration_seconds: int = DEFAULT_SESSION_SECONDS, *, ip_address: str | None = None, user_agent: str | None = None) -> str
Source code in src/hayavega/auth.py
def create_session(database_path: str | Path | dict, user: User, duration_seconds: int = DEFAULT_SESSION_SECONDS, *, ip_address: str | None = None, user_agent: str | None = None) -> str:
    if not user.is_active:
        raise ValueError("Cannot create a session for an inactive user.")
    initialize_auth_db(database_path)
    token = secrets.token_urlsafe(48)
    connection = _connect(database_path)
    try:
        connection.execute(
            "INSERT INTO hayavega_sessions (user_id, token_hash, expires_at, ip_address, user_agent) VALUES (?, ?, ?, ?, ?)",
            (user.id, sha256(token.encode()).hexdigest(), int(time.time()) + duration_seconds, ip_address, user_agent),
        )
        connection.commit()
        record_audit_event(database_path, "auth.session_created", actor_id=user.id, metadata=_login_context(ip_address, user_agent))
        return token
    finally:
        connection.close()

Runtime#

hayavega.runtime.create_application #

create_application(urlpatterns, *, hardening: HardeningOptions | None = None, debug: bool = False, signals: SignalBus | None = None, pulse: SignalBus | None = None, metrics: MetricsRegistry | None = None, api_title: str = 'Hayavega API', api_version: str = '0.1.0', pool_connections: bool = False, middleware: list | None = None)

Build the WSGI application for a route list.

pool_connections=True calls hayavega.db.enable_connection_pooling() -- a convenience, not a scoped setting: pooling is process-global (one connection reused per worker thread per database config), so it affects every hayavega.db.connect() call in the process, not only requests served by this particular application object. See enable_connection_pooling's docstring for what this does and does not solve before enabling it against MySQL/PostgreSQL specifically.

This is separated from serve so application behavior can be tested without opening a network socket. signals, when given a SignalBus, publishes request.started/request.finished asynchronously so a slow subscriber cannot add latency to the response. pulse is a deprecated alias for signals. metrics, when given a MetricsRegistry, records one request-count and duration observation per completed request, labeled by the matched route pattern; pair it with hayavega.metrics.metrics_routes to expose /metrics. api_title and api_version name the generated OpenAPI document at /openapi.json; set api_version to version your API independently of the package. middleware, when given a list of already-resolved middleware factories (hayavega.middleware.load_middleware(Settings.middleware)), wraps route matching and the view call -- it runs after hardening (which always runs first and can't be bypassed by a middleware) and before the matched view. See hayavega.middleware for the factory contract.

Source code in src/hayavega/runtime.py
def create_application(
    urlpatterns,
    *,
    hardening: HardeningOptions | None = None,
    debug: bool = False,
    signals: "SignalBus | None" = None,
    pulse: "SignalBus | None" = None,
    metrics: "MetricsRegistry | None" = None,
    api_title: str = "Hayavega API",
    api_version: str = "0.1.0",
    pool_connections: bool = False,
    middleware: list | None = None,
):
    """Build the WSGI application for a route list.

    ``pool_connections=True`` calls ``hayavega.db.enable_connection_pooling()``
    -- a convenience, not a scoped setting: pooling is process-global (one
    connection reused per worker thread per database config), so it affects
    every ``hayavega.db.connect()`` call in the process, not only requests
    served by this particular application object. See
    ``enable_connection_pooling``'s docstring for what this does and does
    not solve before enabling it against MySQL/PostgreSQL specifically.

    This is separated from ``serve`` so application behavior can be tested
    without opening a network socket. ``signals``, when given a ``SignalBus``,
    publishes ``request.started``/``request.finished`` asynchronously so a
    slow subscriber cannot add latency to the response. ``pulse`` is a
    deprecated alias for ``signals``. ``metrics``, when
    given a ``MetricsRegistry``, records one request-count and duration
    observation per completed request, labeled by the matched route pattern;
    pair it with ``hayavega.metrics.metrics_routes`` to expose ``/metrics``.
    ``api_title`` and ``api_version`` name the generated OpenAPI document at
    ``/openapi.json``; set ``api_version`` to version your API independently
    of the package. ``middleware``, when given a list of already-resolved
    middleware factories (``hayavega.middleware.load_middleware(Settings.middleware)``),
    wraps route matching and the view call -- it runs after hardening
    (which always runs first and can't be bypassed by a middleware) and
    before the matched view. See ``hayavega.middleware`` for the factory
    contract.
    """
    if pulse is not None:
        warnings.warn("`pulse=` is deprecated; use `signals=` instead.", DeprecationWarning, stacklevel=2)
        signals = pulse if signals is None else signals
    if pool_connections:
        db.enable_connection_pooling()
    route_list = [*urlpatterns]
    route_list.extend(documentation_routes(route_list, title=api_title, version=api_version))
    routes = {route.pattern: route for route in route_list}
    wildcard_routes = [route for route in route_list if route.pattern.endswith("/<path>")]
    hardening_layer = Hardening(hardening) if hardening else None
    available_routes_html = _available_routes_html(route_list) if debug else ""

    def _match(requested: str):
        route = routes.get(requested)
        if route is None:
            route = next((item for item in wildcard_routes if requested.startswith(item.pattern.removesuffix("/<path>") + "/")), None)
        return route

    def _dispatch(request) -> Response:
        """Route matching, the view call, and error handling -- the one
        part of the pipeline a middleware chain wraps. Hardening's
        reject/preflight checks happen in the caller, before this is
        reached, and can't be bypassed by a middleware here."""
        requested = request.path.strip("/")
        route = _match(requested)
        if route is None:
            body = f"<h1>404 Not Found</h1><p>No route matches '/{html.escape(requested)}'.</p>{available_routes_html}" if debug else "<h1>404 Not Found</h1>"
            return Response(body, status=404)
        if request.method not in route.methods:
            return Response("<h1>405 Method Not Allowed</h1>", status=405, headers=(("Allow", ", ".join(route.methods)),))
        try:
            result = route.view(request)
            return result if isinstance(result, Response) else Response(str(result))
        except Exception as exc:
            LOGGER.exception("Unhandled request error path=%s", request.path)
            return _error_response(route, exc, debug=debug)

    dispatch = build_chain(middleware or (), _dispatch)

    def application(environ, start_response):
        started_at = perf_counter()
        requested = environ.get("PATH_INFO", "/").strip("/")
        route = _match(requested)
        try:
            content_length = int(environ.get("CONTENT_LENGTH") or 0)
        except ValueError:
            content_length = 0
        size_rejection = hardening_layer.reject_content_length(content_length) if hardening_layer else None
        request = Request.from_wsgi(environ, read_body=size_rejection is None)
        incoming_id = request.headers.get("X-Request-Id", "")
        request_id = incoming_id if _REQUEST_ID.fullmatch(incoming_id) else generate_request_id()
        if signals is not None:
            signals.publish("request.started", data={"method": request.method, "path": request.path})
        rejection = size_rejection or (hardening_layer.reject(request) if hardening_layer else None)
        preflight = hardening_layer.preflight(request) if hardening_layer else None
        if rejection is not None:
            response = rejection
        elif preflight is not None:
            response = preflight
        else:
            response = dispatch(request)
        duration_ms = (perf_counter() - started_at) * 1000
        if signals is not None:
            signals.publish("request.finished", data={"method": request.method, "path": request.path, "status": response.status, "duration_ms": round(duration_ms, 2)})
        extra_headers: tuple[tuple[str, str], ...] = ()
        if hardening_layer:
            response = hardening_layer.finalize(request, response, request_id=request_id, duration_ms=duration_ms)
        elif not any(name.lower() == "x-request-id" for name, _ in response.headers):
            extra_headers = (("X-Request-ID", request_id),)
        if metrics is not None:
            metrics.record(method=request.method, path=route.pattern if route is not None else "unmatched", status=response.status, duration_ms=duration_ms)
        payload = response.body if isinstance(response.body, bytes) else response.body.encode("utf-8")
        start_response(
            f"{response.status} {_STATUS_TEXT.get(response.status, 'OK')}",
            [("Content-Type", response.content_type), ("Content-Length", str(len(payload))), *response.headers, *extra_headers],
        )
        return [payload]

    return application

hayavega.asgi.create_asgi_application #

create_asgi_application(urlpatterns, *, hardening: HardeningOptions | None = None, debug: bool = False, signals: SignalBus | None = None, pulse: SignalBus | None = None, metrics: MetricsRegistry | None = None, api_title: str = 'Hayavega API', api_version: str = '0.1.0', pool_connections: bool = False, max_sync_workers: int = 16, websocket_patterns: list[WebSocketRoute] | None = None, middleware: list | None = None)

Build the ASGI application for a route list.

Accepts the same hardening/debug/signals/metrics/ api_title/api_version/pool_connections arguments as runtime.create_application, with identical behavior (request hardening, Signals events, metrics recording, OpenAPI/Swagger routes) -- only the transport differs. Pooling is thread-local and sync views run via a dedicated worker-thread pool (see max_sync_workers below), so each of those threads gets its own reused connection across the requests it handles, the same benefit WSGI's thread-per-request model gets. pulse is a deprecated alias for signals, kept for parity with the WSGI path.

max_sync_workers (default 16, matching this project's own documented waitress --threads=16 guidance in DEPLOYMENT.md) sizes a dedicated ThreadPoolExecutor that every synchronous view runs on, instead of asyncio.to_thread's event-loop-wide default executor. That default executor is sized min(32, os.cpu_count() + 4) -- tuned for occasional CPU-bound offloading, not for a whole application's worth of concurrent I/O-bound sync views, and on a modest machine (4 cores -> 8 workers) it becomes a real queuing bottleneck under realistic concurrency well before the database or the network does. An in-process microbenchmark for this (benchmarks/asgi_executor_sizing.py) was tried first but proved too noisy on this environment to trust on its own -- repeated runs disagreed on which worker count won. The real evidence is the HTTP-level before/after comparison in benchmarks/README.md#asgi-vs-fastapi: a real uvicorn server under real oha load, run twice before and twice after this change, which was large and consistent in every run (e.g. json_echo roughly doubled). Benchmark your own deployment's concurrency and core count before assuming 16 is right for it, the same standing advice enable_connection_pooling() and DEPLOYMENT.md already give for their own tunables.

Pass websocket_patterns=[websocket_path(...), ...] (from hayavega.websocket) to add WebSocket routes; see that module for the handler contract. WebSocket connections have no WSGI equivalent, so this parameter only ever does anything under this ASGI application. Reuses hardening's allowed_hosts/allowed_origins to reject cross-origin WebSocket connections the same way HTTP requests already are, if either is configured.

middleware, when given a list of already-resolved middleware factories (hayavega.middleware.load_middleware(Settings.middleware)), wraps route matching and the view call for HTTP requests the same way runtime.create_application does -- after hardening, before the matched view. A middleware's returned handler must be async def here (check inspect.iscoroutinefunction if it needs to support both transports); it does not apply to WebSocket connections.

Source code in src/hayavega/asgi.py
def create_asgi_application(
    urlpatterns,
    *,
    hardening: HardeningOptions | None = None,
    debug: bool = False,
    signals: SignalBus | None = None,
    pulse: SignalBus | None = None,
    metrics: MetricsRegistry | None = None,
    api_title: str = "Hayavega API",
    api_version: str = "0.1.0",
    pool_connections: bool = False,
    max_sync_workers: int = 16,
    websocket_patterns: list[WebSocketRoute] | None = None,
    middleware: list | None = None,
):
    """Build the ASGI application for a route list.

    Accepts the same ``hardening``/``debug``/``signals``/``metrics``/
    ``api_title``/``api_version``/``pool_connections`` arguments as
    ``runtime.create_application``, with identical behavior (request
    hardening, Signals events, metrics recording, OpenAPI/Swagger routes)
    -- only the transport differs. Pooling is thread-local and sync views
    run via a dedicated worker-thread pool (see ``max_sync_workers`` below),
    so each of those threads gets its own reused connection across the
    requests it handles, the same benefit WSGI's thread-per-request model
    gets. ``pulse`` is a deprecated alias for ``signals``, kept for parity
    with the WSGI path.

    ``max_sync_workers`` (default 16, matching this project's own documented
    ``waitress --threads=16`` guidance in ``DEPLOYMENT.md``) sizes a
    dedicated ``ThreadPoolExecutor`` that every synchronous view runs on,
    instead of ``asyncio.to_thread``'s event-loop-wide default executor.
    That default executor is sized ``min(32, os.cpu_count() + 4)`` --
    tuned for occasional CPU-bound offloading, not for a whole
    application's worth of concurrent I/O-bound sync views, and on a
    modest machine (4 cores -> 8 workers) it becomes a real queuing
    bottleneck under realistic concurrency well before the database or the
    network does. An in-process microbenchmark for this
    (``benchmarks/asgi_executor_sizing.py``) was tried first but proved too
    noisy on this environment to trust on its own -- repeated runs
    disagreed on which worker count won. The real evidence is the
    HTTP-level before/after comparison in
    ``benchmarks/README.md#asgi-vs-fastapi``: a real ``uvicorn`` server
    under real ``oha`` load, run twice before and twice after this change,
    which was large and consistent in every run (e.g. ``json_echo`` roughly
    doubled). Benchmark your own deployment's concurrency and core count
    before assuming 16 is right for it, the same standing advice
    ``enable_connection_pooling()`` and ``DEPLOYMENT.md`` already give for
    their own tunables.

    Pass ``websocket_patterns=[websocket_path(...), ...]`` (from
    ``hayavega.websocket``) to add WebSocket routes; see that module for the
    handler contract. WebSocket connections have no WSGI equivalent, so this
    parameter only ever does anything under this ASGI application. Reuses
    ``hardening``'s ``allowed_hosts``/``allowed_origins`` to reject
    cross-origin WebSocket connections the same way HTTP requests already
    are, if either is configured.

    ``middleware``, when given a list of already-resolved middleware
    factories (``hayavega.middleware.load_middleware(Settings.middleware)``),
    wraps route matching and the view call for HTTP requests the same way
    ``runtime.create_application`` does -- after hardening, before the
    matched view. A middleware's returned handler must be ``async def``
    here (check ``inspect.iscoroutinefunction`` if it needs to support
    both transports); it does not apply to WebSocket connections.
    """
    if pulse is not None:
        warnings.warn("`pulse=` is deprecated; use `signals=` instead.", DeprecationWarning, stacklevel=2)
        signals = pulse if signals is None else signals
    if pool_connections:
        db.enable_connection_pooling()
    sync_executor = ThreadPoolExecutor(max_workers=max_sync_workers, thread_name_prefix="hayavega-asgi-sync")
    route_list = [*urlpatterns]
    route_list.extend(documentation_routes(route_list, title=api_title, version=api_version))
    routes = {route.pattern: route for route in route_list}
    wildcard_routes = [route for route in route_list if route.pattern.endswith("/<path>")]
    websocket_routes = {route.pattern: route for route in (websocket_patterns or ())}
    hardening_layer = Hardening(hardening) if hardening else None
    available_routes_html = _available_routes_html(route_list) if debug else ""

    def _match(requested: str):
        route = routes.get(requested)
        if route is None:
            route = next((item for item in wildcard_routes if requested.startswith(item.pattern.removesuffix("/<path>") + "/")), None)
        return route

    async def _dispatch(request) -> Response:
        """Route matching, the view call, and error handling -- the one
        part of the pipeline a middleware chain wraps. Hardening's
        reject/preflight checks happen in the caller, before this is
        reached, and can't be bypassed by a middleware here."""
        requested = request.path.strip("/")
        route = _match(requested)
        if route is None:
            not_found_body = f"<h1>404 Not Found</h1><p>No route matches '/{html.escape(requested)}'.</p>{available_routes_html}" if debug else "<h1>404 Not Found</h1>"
            return Response(not_found_body, status=404)
        if request.method not in route.methods:
            return Response("<h1>405 Method Not Allowed</h1>", status=405, headers=(("Allow", ", ".join(route.methods)),))
        try:
            if inspect.iscoroutinefunction(route.view):
                result = await route.view(request)
            else:
                result = await asyncio.get_running_loop().run_in_executor(sync_executor, route.view, request)
            return result if isinstance(result, Response) else Response(str(result))
        except Exception as exc:
            LOGGER.exception("Unhandled request error path=%s", request.path)
            return _error_response(route, exc, debug=debug)

    dispatch = build_chain(middleware or (), _dispatch)

    async def application(scope: dict, receive, send) -> None:
        if scope["type"] == "lifespan":
            await _handle_lifespan(receive, send)
            sync_executor.shutdown(wait=False)
            return
        if scope["type"] == "websocket":
            websocket_route = websocket_routes.get(scope.get("path", "/").strip("/"))
            await _handle_websocket(scope, receive, send, route=websocket_route, hardening_layer=hardening_layer, signals=signals)
            return
        if scope["type"] != "http":
            raise ValueError(f"Hayavega's ASGI application only supports 'http', 'websocket', and 'lifespan' scopes, got {scope['type']!r}.")

        started_at = perf_counter()
        requested = scope.get("path", "/").strip("/")
        route = _match(requested)

        size_rejection = hardening_layer.reject_content_length(_declared_content_length(scope)) if hardening_layer else None
        body = b"" if size_rejection is not None else await _read_body(receive)
        request = _request_from_scope(scope, body)
        incoming_id = request.headers.get("X-Request-Id", "")
        request_id = incoming_id if _REQUEST_ID.fullmatch(incoming_id) else generate_request_id()
        if signals is not None:
            signals.publish("request.started", data={"method": request.method, "path": request.path})
        rejection = size_rejection or (hardening_layer.reject(request) if hardening_layer else None)
        preflight = hardening_layer.preflight(request) if hardening_layer else None
        if rejection is not None:
            response = rejection
        elif preflight is not None:
            response = preflight
        else:
            response = await dispatch(request)
        duration_ms = (perf_counter() - started_at) * 1000
        if signals is not None:
            signals.publish("request.finished", data={"method": request.method, "path": request.path, "status": response.status, "duration_ms": round(duration_ms, 2)})
        extra_headers: tuple[tuple[str, str], ...] = ()
        if hardening_layer:
            response = hardening_layer.finalize(request, response, request_id=request_id, duration_ms=duration_ms)
        elif not any(name.lower() == "x-request-id" for name, _ in response.headers):
            extra_headers = (("X-Request-ID", request_id),)
        if metrics is not None:
            metrics.record(method=request.method, path=route.pattern if route is not None else "unmatched", status=response.status, duration_ms=duration_ms)
        payload = response.body if isinstance(response.body, bytes) else response.body.encode("utf-8")
        # ASGI requires header names to be lowercase bytes (unlike WSGI, which
        # leaves casing to the application); Response objects are built with
        # human-readable casing shared by both transports, so it is lowercased
        # only here, at the ASGI boundary.
        response_headers: list[tuple[bytes, bytes]] = [
            (b"content-type", response.content_type.encode("latin-1")),
            (b"content-length", str(len(payload)).encode("latin-1")),
        ]
        response_headers.extend((name.lower().encode("latin-1"), value.encode("latin-1")) for name, value in (*response.headers, *extra_headers))
        await send({"type": "http.response.start", "status": response.status, "headers": response_headers})
        await send({"type": "http.response.body", "body": payload})

    return application

hayavega.rust_runtime.serve #

serve(urlpatterns, host: str = '127.0.0.1', port: int = 8000, *, hardening: HardeningOptions | None = None, debug: bool = False, workers: int = 8, processes: int = 1, tls_cert: str | None = None, tls_key: str | None = None) -> None

Build the same WSGI application hayavega.runtime.serve would and run it behind the native Rust server instead of wsgiref.

tls_cert/tls_key (PEM file paths) serve the application over HTTPS instead of plain HTTP when both are given; both must be provided together (hayavega_native.serve_wsgi rejects one without the other with a clear error rather than silently serving plaintext). The framework's own Request.is_secure, and therefore Hardening's FORCE_HTTPS/ secure-cookie logic, work correctly under TLS exactly as they do behind a TLS-terminating reverse proxy -- wsgi.url_scheme is set to "https" for every request served this way.

Raises RustRuntimeUnavailableError with an actionable message if hayavega_native was not built for this install (e.g. cargo was not on PATH during pip install, or HAYAVEGA_SKIP_NATIVE_BUILD=1 was set) -- the same situation every other hayavega_native-dependent feature in this codebase (Signals' Rust dispatcher, mango_list) already handles by falling back to a pure-Python equivalent; a WSGI server has no equivalent fallback to fall back to within this same function, so the caller (Settings.server.engine = 'rust' in settings, checked by hayavega launch) is told plainly instead of silently starting the default Python server under a different name.

processes (default 1, i.e. today's single-process behavior) runs processes separate OS processes, each with its own Python interpreter and GIL, all binding the same host:port via SO_REUSEPORT so the kernel load-balances connections across them -- this is the real way to use more than one CPU core: within a single process, CPython's GIL caps Python view-code execution to one thread at a time no matter how large workers is (measured directly: a workers sweep from 1 to 32 plateaus after 2-4 workers on every CPU-bound route). SO_REUSEPORT has no real equivalent on Windows (SO_REUSEADDR there does not load-balance the same way), so processes > 1 raises RustRuntimeUnavailableError there rather than silently falling back to single-process behavior. This is a multi-process launcher, not a supervisor: a child process that exits (crash or otherwise) is not restarted -- that is real, disclosed scope this function does not cover yet (see NEXT_RELEASE_ROADMAP.md's Phase 4 for the remaining supervision/restart work).

Source code in src/hayavega/rust_runtime.py
def serve(urlpatterns, host: str = "127.0.0.1", port: int = 8000, *, hardening: HardeningOptions | None = None, debug: bool = False, workers: int = 8, processes: int = 1, tls_cert: str | None = None, tls_key: str | None = None) -> None:
    """Build the same WSGI application `hayavega.runtime.serve` would and run
    it behind the native Rust server instead of `wsgiref`.

    `tls_cert`/`tls_key` (PEM file paths) serve the application over HTTPS
    instead of plain HTTP when both are given; both must be provided
    together (`hayavega_native.serve_wsgi` rejects one without the other with
    a clear error rather than silently serving plaintext). The framework's
    own `Request.is_secure`, and therefore `Hardening`'s `FORCE_HTTPS`/
    secure-cookie logic, work correctly under TLS exactly as they do behind
    a TLS-terminating reverse proxy -- `wsgi.url_scheme` is set to
    `"https"` for every request served this way.

    Raises `RustRuntimeUnavailableError` with an actionable message if
    `hayavega_native` was not built for this install (e.g. `cargo` was not on
    `PATH` during `pip install`, or `HAYAVEGA_SKIP_NATIVE_BUILD=1` was set) --
    the same situation every other `hayavega_native`-dependent feature in this
    codebase (Signals' Rust dispatcher, `mango_list`) already handles by
    falling back to a pure-Python equivalent; a WSGI *server* has no
    equivalent fallback to fall back to within this same function, so the
    caller (`Settings.server.engine = 'rust'` in settings, checked by
    `hayavega launch`) is told plainly instead of silently starting the
    default Python server under a different name.

    `processes` (default 1, i.e. today's single-process behavior) runs
    `processes` separate OS processes, each with its own Python interpreter
    and GIL, all binding the same `host:port` via `SO_REUSEPORT` so the
    kernel load-balances connections across them -- this is the real way to
    use more than one CPU core: within a single process, CPython's GIL caps
    Python view-code execution to one thread at a time no matter how large
    `workers` is (measured directly: a `workers` sweep from 1 to 32 plateaus
    after 2-4 workers on every CPU-bound route). `SO_REUSEPORT` has no real
    equivalent on Windows (`SO_REUSEADDR` there does not load-balance the
    same way), so `processes > 1` raises `RustRuntimeUnavailableError` there
    rather than silently falling back to single-process behavior. This is a
    multi-process launcher, not a supervisor: a child process that exits
    (crash or otherwise) is not restarted -- that is real, disclosed scope
    this function does not cover yet (see `NEXT_RELEASE_ROADMAP.md`'s
    Phase 4 for the remaining supervision/restart work).
    """
    try:
        import hayavega_native
    except ImportError as error:
        raise RustRuntimeUnavailableError("Settings.server.engine is 'rust' but hayavega_native was not built for this install. Rebuild with `cargo` on PATH (`pip install -e .` re-runs the build hook), or set Settings.server.engine to 'python' to use the standard runtime.") from error

    if processes > 1 and sys.platform == "win32":
        raise RustRuntimeUnavailableError("processes>1 needs SO_REUSEPORT for real multi-core scaling, which has no real equivalent on Windows (SO_REUSEADDR there does not load-balance the same way). Use processes=1 on Windows, or deploy on Linux for multi-process scaling.")

    # Environment-dependent, not a design gap: reached only when
    # hayavega_native is actually built (the ImportError above already
    # returned otherwise), so the pure-Python fallback-mode CI job can
    # never exercise it -- the native-extension job does, for real
    # (tests/test_rust_runtime.py and friends); excluded here so that
    # job's 100% coverage gate stays meaningful.
    application = create_application(urlpatterns, hardening=hardening, debug=debug)  # pragma: no cover
    scheme = "https" if tls_cert else "http"  # pragma: no cover
    reuse_port = processes > 1  # pragma: no cover
    print(f"Hayavega development server (Rust runtime, {processes} process(es) x {workers} workers): {scheme}://{host}:{port}/")  # pragma: no cover

    if processes > 1:  # pragma: no cover
        child_pids = [_fork_worker(application, host, port, workers, tls_cert, tls_key) for _ in range(processes - 1)]
        try:
            hayavega_native.serve_wsgi(application, host, port, workers, tls_cert, tls_key, reuse_port)
        finally:
            for pid in child_pids:
                try:
                    os.kill(pid, 15)  # SIGTERM
                except OSError:
                    pass
    else:  # pragma: no cover
        hayavega_native.serve_wsgi(application, host, port, workers, tls_cert, tls_key, reuse_port)