#Server API Reference

The server module manages the Granian ASGI server lifecycle: PID file management, port availability checks, single-instance enforcement, background and foreground serving, hot reload, and graceful shutdown.

Server symbols are lazily imported from the top-level fastware package to avoid the ~60ms cost of importing Granian when only the routing/response layer is needed.

#src.fastware.server

Granian ASGI server lifecycle management with PID file tracking, port availability checks, foreground and background serve modes, and graceful stop.

Instance registry ----------------- Alongside the .pid/.port files, each running instance drops a JSON descriptor (PID, port, name) into a per-directory registry so peers can enumerate running instances -- something a single PID file cannot express.

Layout: for a PID file at <dir>/<name>.pid the registry lives in <dir>/.fastware-registry/ with one file per instance named <pid>.json. The PID is globally unique, so instances that share a directory never collide, and a crashed instance leaves at most one stale file.

Concurrency model (design note for future shared-counter use): the base design needs no locking. Each instance owns exactly one file which it writes once (at startup) and deletes once (at exit); readers prune files whose PID is dead (kill -0). Because writes and deletes target disjoint paths, concurrent instances never contend and there is no shared mutable file to corrupt.

Should a shared, atomically-updated value ever be required (e.g. a global "active instances" counter or a leader election token), it must NOT be layered on top of these per-instance files by read-modify-writing a shared JSON blob -- that reintroduces the lost-update race this design avoids. The correct extension is either (a) derive the aggregate by enumerating the per-instance files on read (no shared state at all -- the counter is len(list_instances)), or (b) if a durable token is truly needed, use an atomic filesystem primitive (O_CREAT|O_EXCL create for claim, atomic rename for compare-and-swap) rather than a lock around a mutable file. This note feeds a later decision in the desktop layer; the per-instance-file design is intentionally the floor.

#PortInUseError

Raised when the requested port is already in use.

#AlreadyRunningError

Raised when a single-instance server is already running.

#_probe_health

def _probe_health(url: str, timeout: float) -> int | None

Probe url over HTTP within timeout seconds.

Returns the HTTP status code if the server answered at the HTTP level -- including error statuses like 404, which still prove the server is up and listening. Returns None if the connection failed outright (refused, reset, timed out, DNS failure): the server is not reachable.

Callers pick their own health criterion from the result: "answered at all" (code is not None) proves the process is up; "answered with a success status" (200 <= code < 400) proves the endpoint is healthy.

#_write_pid

def _write_pid(pid_path: Path, *, exclusive: bool=False) -> None

Write the current PID to disk, become process group leader, and register cleanup.

Becoming a process group leader (via os.setpgid(0, 0)) lets stop() signal our entire process group with os.killpg, so granian worker subprocesses die with us instead of being orphaned.

When exclusive is True, the PID file is created atomically with O_CREAT|O_EXCL: if it already exists, another instance won the race between the single-instance check and PID creation, and AlreadyRunningError is raised instead of overwriting its PID file.

No signal handlers are installed here: Granian installs its own SIGTERM/SIGINT handlers during startup (which trigger a graceful shutdown, after which the atexit hook removes the PID file), and group-wide termination is stop()'s responsibility.

#_remove_pid

def _remove_pid(pid_path: Path) -> None

Remove the PID file if it exists.

#_reap_if_dead_child

def _reap_if_dead_child(pid: int) -> None

Clear pid from the process table if it is our own already-exited child.

Every liveness probe in this module is os.kill(pid, 0), which SUCCEEDS for a zombie -- a process that has exited but whose parent has not waited on it yet. serve_background spawns the server as a direct child of the caller and discards the handle, so in the very common case where the same process later calls stop/status/list_instances, a dead server reads as alive: stop polls out its entire 10s grace window and escalates to SIGKILL every time, status reports it running, list_instances keeps serving a stale descriptor, and check_already_running refuses to start a replacement.

A non-blocking wait clears the zombie so the probe that follows tells the truth. A pid that is not our child raises ChildProcessError and one that is still running returns immediately with (0, 0) -- both are no-ops.

Consequence worth stating, because the wait is unconditional: if the calling program also holds a subprocess.Popen for this PID, this reap consumes the exit status that Popen was waiting for. CPython's Popen._try_wait swallows the resulting ChildProcessError and records a returncode of 0, so the consumer's later poll()/wait() reports a clean exit it never observed -- including for a process that crashed or was killed with SIGKILL here.

That is acceptable at these call sites because the PIDs reaching them are fastware's own: they come from a PID file or an instance descriptor that only serve/serve_background write, for the supervisor process fastware itself spawned and whose Popen handle it deliberately discards (serve_background detaches; nothing in this module keeps one to wait on). The rule that keeps it true: never point a fastware PID file or registry descriptor at a process whose exit status someone else needs.

#check_already_running

def check_already_running(pid_path: Path) -> int | None

Check if another instance is running. Returns the PID if running, None otherwise.

Stale PID files (process dead) are cleaned up automatically.

#ensure_port_available

def ensure_port_available(host: str, port: int, name: str='server', pid_path: Path | None=None) -> int

Ensure port is available. Returns port on success.

If the port is occupied and pid_path names a previous instance of this server that verifiably holds the port (the PID in the PID file matches a PID reported as holding the port), that stale instance is stopped and the port reclaimed. Any other occupant raises PortInUseError -- ownership is never guessed from response contents.

#_find_port_holder_pids

def _find_port_holder_pids(port: int) -> list[int]

Return the PIDs of processes holding a TCP port, via lsof or fuser.

Returns an empty list when no holder can be identified (including when neither tool is available) -- callers must treat that as "ownership not proven" and refuse to kill anything.

#_resolve_target

def _resolve_target(target: str | Callable) -> str

Convert a target to a Granian-compatible string.

If target is already a string (e.g. "myapp:app"), return as-is. If target is a callable, register it on a synthetic module so Granian can import it via its string-based loader.

#_materialize_target

def _materialize_target(target: str | Callable) -> tuple[str, Path | None]

Resolve a target for use in a separate process.

Strings pass through unchanged (with no shim directory). Callables are materialized as a real shim module file in a fresh temp directory: the child process adds that directory to sys.path and imports the shim, which re-imports the callable from its defining module. Returns (target_str, shim_dir); the caller must clean up shim_dir once the child has imported it.

Only module-level callables of importable modules can be materialized. Anything else (locals, lambdas, objects defined in __main__) cannot be re-imported by another process -- raise a clear error immediately instead of letting the child crash.

#_remove_shim_dir

def _remove_shim_dir(shim_dir: Path | None) -> None

Remove a temp shim directory created by _materialize_target.

#_find_free_port

def _find_free_port(host: str='127.0.0.1') -> int

Find an ephemeral port that is currently free on host.

#_port_file_path

def _port_file_path(pid_path: Path) -> Path

Derive the port file path from a PID file path.

E.g. .pixelweaver.pid -> .pixelweaver.port.

#_write_port_file

def _write_port_file(pid_path: Path, port: int) -> None

Write the bound port alongside the PID file.

#_remove_port_file

def _remove_port_file(port_path: Path) -> None

Remove a port file if it exists.

#read_port_file

def read_port_file(pid_path: Path) -> int | None

Read the port stored alongside a PID file. Returns None if missing or unreadable.

#RegistryEntry

A registered running server instance: process id, bound port, and name.

#_registry_dir

def _registry_dir(pid_path: Path) -> Path

Registry directory derived from the PID file's directory.

<dir>/<name>.pid -> <dir>/.fastware-registry/.

#_registry_entry_path

def _registry_entry_path(pid_path: Path, pid: int) -> Path

Path of the JSON descriptor for pid in the registry.

#_pid_alive

def _pid_alive(pid: int) -> bool

Return True if pid is alive (mirrors check_already_running's kill-0).

#register_instance

def register_instance(pid_path: Path, port: int, name: str) -> None

Register the current process as a running instance.

Writes a <pid>.json descriptor (PID, port, name) into the registry directory derived from pid_path and schedules its removal at process exit. Called wherever the PID/port files are written -- for serve_background that is the spawned child, keeping ownership consistent with the PID/port files it also writes.

#deregister_instance

def deregister_instance(pid_path: Path, pid: int | None=None) -> None

Remove an instance's registry descriptor.

Defaults to the current process's PID. Missing files are ignored so the atexit hook is safe even if the file was already pruned by a reader.

#list_instances

def list_instances(pid_path: Path) -> list[RegistryEntry]

Enumerate live registered instances, pruning stale descriptors.

Reads every *.json in the registry directory. Descriptors whose PID is no longer alive (kill -0), or that are unreadable/corrupt, are deleted and excluded. Returns the live entries sorted by PID for stable ordering.

#_marker_path

def _marker_path(pid_path: Path, kind: str, pid: int, marker_id: str) -> Path

Path of the marker file for (kind, pid, marker_id).

#write_marker

def write_marker(pid_path: Path, kind: str, marker_id: str, *, fields: dict | None=None) -> Path

Write a presence marker owned by the current process. Returns its path.

kind groups markers (e.g. "window", "focus-request"); marker_id identifies the marker within this process. The payload always carries pid, marker_id and kind; fields merges in extra values (e.g. a window_id). kind and marker_id must be filesystem-safe (no / or -- separators).

#list_markers

def list_markers(pid_path: Path, kind: str, *, prune_dead: bool=True) -> list[dict]

Enumerate markers of kind, returning each payload as a dict.

When prune_dead is True (default), markers whose owning PID is no longer alive (kill -0) -- and any unreadable/corrupt marker -- are deleted and excluded, mirroring :func:list_instances. Pass prune_dead=False when the caller consumes markers explicitly (e.g. focus requests) and a dead owner's request should still be honoured. Sorted by (pid, marker_id).

#remove_marker

def remove_marker(pid_path: Path, kind: str, marker_id: str, *, pid: int | None=None) -> None

Remove a marker. Defaults to the current process's PID.

Pass pid explicitly to remove a marker owned by another process (e.g. a consumer clearing a focus request written by a joining process).

#_resolve_host_port

def _resolve_host_port(host: str | None, port: int | None, name: str) -> tuple[str, int]

Resolve host and port from explicit args or env vars.

Explicit args override env vars. If neither is provided, raise ValueError.

#_make_server

def _make_server(target: str, host: str, port: int, *, loop: LoopChoice='asyncio', workers: int=1) -> Granian

Create a Granian instance bound to host on port.

target is an ASGI module path, e.g. "myapp:app".

loop pins the event-loop implementation (see :func:serve). workers is the number of worker processes (see :func:serve).

#_make_embed_server

def _make_embed_server(target: Callable, host: str, port: int) -> object

Create a Granian embed server for in-process background serving.

target is the ASGI callable (not a string). Returns the embed server instance (with serve() and stop() methods).

#_run_server

def _run_server(target: str, host: str, port: int, extra_sys_path: str | None=None, loop: LoopChoice='asyncio', workers: int=1) -> None

Create and run a Granian server. Used as the reload subprocess target.

extra_sys_path lets the reload child import the temp shim module created for callable targets (see _materialize_target).

#_serve_subprocess

def _serve_subprocess(target: str, host: str, port: int, pid_path_str: str, loop: LoopChoice='asyncio', workers: int=1, name: str='FASTWARE') -> None

Entry point for the server subprocess. Runs granian in foreground mode.

The child owns the PID/port files and the instance registry entry, keeping all instance-tracking artifacts written by the same process.

#serve_background

def serve_background(target: str | Callable, *, host: str, port: int, pid_path: Path, name: str='FASTWARE', loop: LoopChoice='asyncio', workers: int=1) -> str

Start the server as an independent background process. Returns the URL.

Unlike serve(foreground=False) which uses a daemon thread (dies with the parent), this spawns a fully detached subprocess that survives the parent exiting. The subprocess writes its own PID and port files.

Parameters ---------- target: ASGI application -- either a module path string (e.g. "myapp:app") or a callable ASGI application object. host: Bind address. port: Bind port. Pass 0 to pick a random free port. pid_path: Path for the PID file. The subprocess writes this, not the caller. name: Application name for log messages. loop: Event-loop implementation. See :func:serve for the rationale behind the pinned "asyncio" default. workers: Number of worker processes. See :func:serve for the duplication hazard when workers > 1 with lifespan-started background tasks.

Returns ------- str The URL the server is listening on (e.g. "http://127.0.0.1:8000").

Raises ------ RuntimeError If the server process exits prematurely or fails to start within 10s.

#serve

def serve(target: str | Callable, *, foreground: bool, host: str | None=None, port: int | None=None, pid_path: Path | None=None, name: str='FASTWARE', pre_serve: Callable[[], None] | None=None, reload: bool=False, single_instance: bool=True, loop: LoopChoice='asyncio', workers: int=1) -> str | None

Start Granian serving the ASGI app.

Parameters ---------- target: ASGI application -- either a module path string (e.g. "myapp:app") or a callable ASGI application object. foreground: Required. When True, blocks the calling thread. When False, spawns a daemon thread and returns the URL string. host: Bind address. Falls back to {NAME}_HOST env var. ValueError if neither. port: Bind port. Falls back to {NAME}_PORT env var. ValueError if neither. pid_path: If given, enables PID file management (detect existing instances, write PID, register cleanup). name: Application name for env var prefix (uppercased) and log messages. Default "FASTWARE". pre_serve: Optional callable invoked synchronously after PID/port checks but before Granian starts. reload: When True, watches .py files in the current working directory and restarts the server on changes. Requires foreground=True. single_instance: When True (default), checks for an existing running instance via the PID file and exits with an error if one is found. When False, skips the PID check (but still writes the PID file if pid_path is given). loop: Event-loop implementation: "asyncio" (default), "uvloop", or "rloop". fastware deliberately does NOT expose Granian's "auto" mode, which silently resolves rloop -> uvloop -> asyncio based on which packages are installed -- meaning the same code would pick a different loop just because a transitive dependency pulled in rloop or uvloop. The pinned "asyncio" default keeps behaviour environment-independent and preserves stdlib asyncio-subprocess semantics (rloop, which would win "auto" resolution, breaks asyncio.create_subprocess_* workloads such as Playwright). Choose "uvloop"/"rloop" explicitly if you want their throughput and understand the trade-offs. workers: Number of Granian worker processes (default 1). WARNING: workers > 1 forks the ENTIRE app -- including the lifespan handler -- once per worker. Any singleton or background task started in the lifespan (a scheduler, a connection pool, an asyncio worker loop) is DUPLICATED per worker, which is almost never what you want. For apps with lifespan-started singletons, workers=1 is the only correct value.

Returns ------- str | None When foreground=False, returns the URL string (e.g. "http://127.0.0.1:8000"). When foreground=True, returns None (blocks until server stops).

#stop_background

def stop_background(url: str) -> None

Stop an in-process background server started by serve(foreground=False).

Parameters ---------- url: The URL returned by serve(foreground=False).

Raises ------ KeyError If no background server is tracked for url.

#ServerStatus

Result of a status check on a server process.

#_cleanup_pid_and_port

def _cleanup_pid_and_port(pid_path: Path) -> None

Remove both the PID file and its companion port file.

#stop

def stop(pid_path: Path) -> None

Stop a server by reading its PID file.

Sends SIGTERM to the server's process group when the server is its own group leader (as arranged by _write_pid), so granian worker subprocesses die with it. If the server does not lead its group (setpgid failed at startup), only the single PID is signalled -- a group the server does not lead may contain unrelated processes such as the launching shell.

Waits up to 10s polling with os.kill(pid, 0), then escalates to SIGKILL and waits (briefly, bounded) for that kill to land before returning, so a caller that probes immediately afterwards is never told the corpse is still running. Cleans up the PID file and port file.

Raises FileNotFoundError if PID file does not exist. If the process is already gone (stale PID file), removes the PID file and returns normally.

#_confirm_dead

def _confirm_dead(pid: int) -> bool

Wait (briefly, bounded) until pid is gone from the process table.

os.kill only queues SIGKILL: it returns before the kernel has torn the target down, and once it has, a process of ours lingers as a zombie until it is waited on. Both states answer os.kill(pid, 0), so a stop() that returned straight after signalling would leave every liveness probe -- status, list_instances, check_already_running, or a caller's own kill-0 -- reporting a corpse as a running server.

Returns True once the PID is confirmed gone. Returns False if the bounded window expires or the PID is not ours to probe: a PID that survives SIGKILL is one we neither own nor can reap (a foreign zombie held by another parent, or a process we may signal but not wait on), and waiting longer cannot change that.

#status

def status(pid_path: Path, health_url: str | None=None) -> ServerStatus

Check the status of a server process.

Parameters ---------- pid_path: Path to the PID file. health_url: Optional URL to probe for health (e.g. "http://127.0.0.1:8000/health"). If provided and the process is running, an HTTP GET is attempted with a short timeout.

Returns ------- ServerStatus Dataclass with running, pid, and healthy fields.

#ServerStatus

The ServerStatus enum represents the 3 possible states of a fastware server instance: running, stopped, or unknown. It is returned by status() after checking the PID file and probing the process.

#Error Types

#PortInUseError

Raised by ensure_port_available when the requested port is already in use and cannot be reclaimed. The error message includes the host and port, and suggests using a different port or stopping the process holding it.

Before raising, ensure_port_available attempts to detect whether the port holder is a stale instance of the same server (by probing GET /health). If it is, the stale process is killed automatically and the port is reclaimed without error.

#AlreadyRunningError

Raised by serve (when single_instance=True) if a PID file exists and the corresponding process is still alive. The error message includes the PID of the running instance. This prevents accidentally starting duplicate servers on the same port, which would cause bind failures or silent request splitting.

Stale PID files (where the process has died) are cleaned up automatically and do not trigger this error.

#serve() vs serve_background()

serve() is the primary entry point for starting the Granian ASGI server. It supports both foreground (blocking) and background (daemon thread) modes, with optional PID file management, single-instance enforcement, and hot reload for development workflows:

from fastware.server import serve

# Foreground mode -- blocks until the server stops
serve(
    "myapp:app",
    foreground=True,
    host="127.0.0.1",
    port=8000,
    pid_path=Path(".myapp.pid"),
)

# Background mode -- returns the URL, server runs in a daemon thread
url = serve(
    "myapp:app",
    foreground=False,
    host="127.0.0.1",
    port=8000,
)
# url == "http://127.0.0.1:8000"

serve_background() spawns a fully detached subprocess that survives the parent process exiting. Use this when you need the server to outlive the caller (e.g., starting a server from a CLI command):

from fastware.server import serve_background

# Detached subprocess -- survives parent exit
url = serve_background(
    "myapp:app",
    host="127.0.0.1",
    port=8000,
    pid_path=Path(".myapp.pid"),
)

#Hot Reload

Pass reload=True to serve() for development. This uses watchfiles to monitor all .py files in the project directory and automatically restart the Granian server on changes, providing sub-second feedback during development. Requires foreground=True:

serve(
    "myapp:app",
    foreground=True,
    host="127.0.0.1",
    port=8000,
    reload=True,
)