Python client

Python client

warden ships the client it uses itself. Install warden-ports and import warden — the distribution and the package have different names.

from warden import register

port = register("shop-api", kind="backend", project="shop")

Asking for a port at startup

The shortest thing that works. register returns the port and nothing else:

import os

import uvicorn
from fastapi import FastAPI
from warden import register

app = FastAPI()

if __name__ == "__main__":
    port = register("shop-api", kind="backend", project="shop", pid=os.getpid())
    uvicorn.run(app, port=port)

Restart it and the same name gets the same port back. Passing pid is optional and only makes warden ls more useful to whoever is reading it.

Finding a neighbour

The other half of the point: not hardcoding where the backend is.

from warden import WardenClient

with WardenClient() as client:
    backend = client.lookup("shop-api")

BASE_URL = f"http://{backend.address}"

lookup raises UnknownServiceError if nothing is registered under that name, which is usually what you want — a frontend that starts against a backend that is not there is worse than one that refuses to start.

Handing the port back

For a test fixture or anything short-lived, reserve releases on the way out, including when the block raises:

from warden import reserve

def test_the_thing():
    with reserve("test-fixture", kind="worker") as port:
        with serve_on(port):
            ...
    # released here

WardenClient.session is the same thing when you already have a client:

with WardenClient() as client, client.session("fixture", kind="worker") as service:
    print(service.port, service.name)

A lease, for things that will not clean up

CI runners and containers do not always get to run their exit code. A ttl makes the registration expire on its own:

with WardenClient() as client:
    service = client.register("ci-runner", kind="worker", ttl=600)

    while working:
        do_some_work()
        client.heartbeat("ci-runner")     # ten more minutes

A heartbeat without a ttl renews the lease the service registered with, so it can never turn a lease into a permanent registration by accident.

Every method

from warden import WardenClient

client = WardenClient("http://hub:7010", token="...", timeout=5.0)

Without arguments it reads WARDEN_URL and WARDEN_TOKEN, falling back to http://127.0.0.1:7010. Use it as a context manager, or call close().

Services

Method	Returns
register(name, kind=..., project=..., host=..., preferred_port=..., require_port=..., pid=..., ttl=..., meta=..., node=...)	Registration
lookup(name)	Registration
services(project=..., kind=...)	list[Registration]
heartbeat(name, pid=..., ttl=..., node=...)	Registration
release(name, node=...)	None
pool()	PoolStatus
session(name, **kwargs)	context manager yielding Registration

This machine, or the one the warden runs on

Method	Returns
listeners(udp=True)	list[Listener]
stop(pid, force=False, node=...)	None

The fleet

Method	Returns
nodes()	list[Node]
announce(name, url=..., pool_start=..., pool_end=..., version=...)	Node
forget(name)	None
fleet_services(project=..., kind=...)	FleetServices
fleet_lookup(node, name)	FleetRegistration
fleet_pool()	FleetPool
fleet_listeners(udp=True)	FleetListeners

node= on the write methods puts the request through the warden you are talking to and onto that one. See Cluster.

Updates

Method	Returns
update_status()	UpdateStatus
update_self()	str, what the update command printed
update_fleet()	FleetUpdate

What comes back

Everything is a pydantic model, so it has real attributes and model_dump(mode="json") when you need plain data.

service.name        # "shop-api"
service.port        # 8000
service.host        # "127.0.0.1"
service.address     # "127.0.0.1:8000"
service.kind        # "backend"
service.project     # "shop" or None
service.pid         # 14204 or None
service.ttl         # 600 or None
service.expires_at  # datetime or None
service.meta        # {"branch": "main"}

A FleetServices keeps what answered and what did not in separate fields, on purpose — a shorter list because a machine was down reads exactly like a shorter list because nothing is registered there:

fleet = client.fleet_services()
for service in fleet.services:
    print(service.node, service.name, service.port)
for missing in fleet.unreachable:
    print(missing.node, missing.reason)
for clash in fleet.duplicates:
    print(clash.name, "is held by", clash.nodes)

When it goes wrong

Every failure is an exception under WardenError, so one except catches the lot and the specific ones are there when you want them:

from warden.errors import (
    WardenError,              # everything below, and an unreachable warden
    UnknownServiceError,      # 404, no such service
    UnknownNodeError,         # 404, no such node
    PortUnavailableError,     # 409, a required port is held
    PoolExhaustedError,       # 503, nothing left in the pool
    NotPermittedError,        # 403, allowed to ask, not allowed to do
    ProtectedProcessError,    # the operating system's own, or warden itself
    StillRunningError,        # a process ignored the request to stop
    UnknownProcessError,      # no process with that id
    UpdateFailedError,        # the update command failed
    RelayedError,             # a node refused, in its own words
)

Each carries .message, which is written to be shown to a person as it is:

try:
    port = register("shop-api", kind="backend")
except PoolExhaustedError as exc:
    sys.exit(f"cannot start: {exc.message}")
except WardenError as exc:
    sys.exit(f"warden: {exc.message}")

A warden that is not running raises the base WardenError with an explanation rather than a bare connection error, so a script does not have to know about httpx to say something useful.

Falling back when there is no warden

register needs one. If you would rather your service start anyway:

from warden import register
from warden.errors import WardenError

try:
    port = register("shop-api", kind="backend")
except WardenError:
    port = 8000     # what you would have hardcoded anyway

Worth deciding deliberately: starting on a guess is what warden exists to stop, but a development machine where nobody has started the registry is a fair reason to carry on.

Reading the machine without a warden at all

warden.listeners needs no server, because the sockets belong to the machine:

from warden.listeners import listeners, holder_of, stop

for socket in listeners(udp=False):
    print(socket.port, socket.process, socket.pid)

held = holder_of(3000)
if held and held.pid:
    stop(held.pid, force=False)

stop refuses the operating system's own processes and warden itself, whatever is asked of it.