Module

server

Server — orchestrates the full pounce lifecycle.

Manages the state machine:

CONFIG → DETECT → BIND → LIFESPAN → SERVE → SHUTDOWN

Whenworkers == 1the server runs a single-worker fast path with no supervisor overhead. Whenworkers > 1the supervisor spawns and monitors worker threads (nogil) or processes (GIL).

Signal handling: SIGINT/SIGTERM trigger graceful shutdown.

1Class1Function

Classes

Server 10 ▼
Top-level server that orchestrates the full lifecycle. Creates the socket(s), runs lifespan events…

Top-level server that orchestrates the full lifecycle.

Creates the socket(s), runs lifespan events, starts workers (via the supervisor when multi-worker), and handles shutdown signals.

Methods

run 0 ▼
Start the server (blocking). Lifecycle: 1. Configure logging 2. Resolve effect…
def run(self) -> None

Start the server (blocking).

Lifecycle:

  1. Configure logging
  2. Resolve effective worker count and detect worker mode
  3. Create TLS context (if configured)
  4. Print startup banner
  5. Bind socket(s)
  6. Run ASGI lifespan startup (once, in the main thread)
  7. Start worker(s) — single-worker fast path or supervisor
  8. Wait for shutdown signal
  9. Run ASGI lifespan shutdown
  10. Close socket(s)
shutdown 0 ▼
Request graceful shutdown. Thread-safe and idempotent. Can be called from any …
def shutdown(self) -> None

Request graceful shutdown. Thread-safe and idempotent.

Can be called from any thread to stop a running server. In single-worker mode this wakes the asyncio event loop via call_soon_threadsafe. In multi-worker mode this delegates to the supervisor's shutdown.

Safe to call beforerun()— the server will exit immediately on startup. Safe to call multiple times.

Internal Methods 8 ▼
__init__ 4 ▼
def __init__(self, config: ServerConfig, app: ASGIApp, *, app_path: str | None = None, lifecycle_collector: LifecycleCollector | None = None) -> None
Parameters
Name Type Description
config —
app —
app_path — Default:None
lifecycle_collector — Default:None
_run_single 0 ▼
Run with a single worker — no supervisor, minimal overhead.
def _run_single(self) -> None
_run_single_with_reload 0 ▼
Single-worker mode with auto-reload on source changes. Runs the worker in a lo…
def _run_single_with_reload(self) -> None

Single-worker mode with auto-reload on source changes.

Runs the worker in a loop. When the file watcher detects changes it sets_reload_requestedand signals shutdown, causing the asyncio loop to exit. The outer loop then restarts the worker with a fresh socket and event loop.

_run_single_async 1 ▼
Async entry point for single-worker mode.
async
async def _run_single_async(self, sock: socket.socket) -> None
Parameters
Name Type Description
sock —
_run_multi 2 ▼
Run with multiple workers managed by the supervisor. Lifespan runs once in the…
def _run_multi(self, effective_workers: int, mode: WorkerMode) -> None

Run with multiple workers managed by the supervisor.

Lifespan runs once in the main process/thread before workers are spawned. Workers do not run lifespan.

When--reloadis active, a file watcher thread runs alongside the supervisor and triggersrestart_workers()on changes.

Parameters
Name Type Description
effective_workers —
mode —
_run_lifespan_then_supervise 2 ▼
Run lifespan in the main thread, then hand off to supervisor. Installs asyncio…
async
async def _run_lifespan_then_supervise(self, supervisor: Supervisor, sockets: list[socket.socket]) -> None

Run lifespan in the main thread, then hand off to supervisor.

Installs asyncio signal handlers so SIGINT/SIGTERM trigger a coordinated graceful shutdown through the supervisor instead of a rawKeyboardInterrupt. On the first signal the supervisor drains workers; a second signal removes the handler so the defaultKeyboardInterruptforces an immediate exit.

Parameters
Name Type Description
supervisor —
sockets —
_print_banner 2 ▼
Print the startup banner to stderr.
def _print_banner(self, effective_workers: int, mode: WorkerMode) -> None
Parameters
Name Type Description
effective_workers —
mode —
_close_sockets 1 ▼
Close all sockets, deduplicating shared-fd sockets. On macOS (no SO_REUSEPORT)…
staticmethod
def _close_sockets(sockets: list[socket.socket]) -> None

Close all sockets, deduplicating shared-fd sockets.

On macOS (no SO_REUSEPORT) all workers share the same socket fd. Deduplicate by fd and guard against already-closed fds to avoid ValueError: Invalid file descriptor: -1on shutdown.

Parameters
Name Type Description
sockets —

Functions

_get_version 0 str ▼
Get the pounce version string.
def _get_version() -> str
Returns
str