Skip to content

Dependencies, sessions, logging, testing

Depends 

Depends(dependency: Any, *, use_cache: bool = True)

Marks a handler argument as supplied by a dependency.

Source code in python/aether/_depends.py 
35
36
37
38
39
40
41
def __init__(self, dependency: Any, *, use_cache: bool = True) -> None:
    if not callable(dependency):
        raise TypeError(
            f"Depends() needs a callable, got {type(dependency).__name__}"
        )
    self.dependency = dependency
    self.use_cache = use_cache

dependency instance-attribute 

dependency = dependency

use_cache instance-attribute 

use_cache = use_cache

Sessions 

Sessions(
    secret: str | bytes,
    *,
    cookie: str = "aether_session",
    max_age: int = DEFAULT_MAX_AGE,
    secure: bool = True,
    same_site: str = "Lax",
    path: str = "/",
)

Reads and writes a signed session cookie.

Source code in python/aether/_sessions.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    secret: str | bytes,
    *,
    cookie: str = "aether_session",
    max_age: int = DEFAULT_MAX_AGE,
    secure: bool = True,
    same_site: str = "Lax",
    path: str = "/",
) -> None:
    if not secret:
        raise ValueError("sessions need a secret")
    self.secret = secret.encode() if isinstance(secret, str) else secret
    self.cookie = cookie
    self.max_age = max_age
    self.secure = secure
    self.same_site = same_site
    self.path = path

secret instance-attribute 

secret = (
    secret.encode() if isinstance(secret, str) else secret
)

cookie instance-attribute 

cookie = cookie

max_age instance-attribute 

max_age = max_age

secure instance-attribute 

secure = secure

same_site instance-attribute 

same_site = same_site

path instance-attribute 

path = path

encode 

encode(data: dict) -> str
Source code in python/aether/_sessions.py 
106
107
108
109
110
def encode(self, data: dict) -> str:
    payload = _b64encode(
        json.dumps({"d": data, "t": int(time.time())}, separators=(",", ":")).encode()
    )
    return f"{payload}.{self._sign(payload)}"

decode 

decode(raw: str) -> dict

Return the data, or an empty dict if the cookie is not trustworthy.

Source code in python/aether/_sessions.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def decode(self, raw: str) -> dict:
    """Return the data, or an empty dict if the cookie is not trustworthy."""
    payload, _, signature = raw.partition(".")
    if not payload or not signature:
        return {}
    # Constant time: a fast reject on the first wrong byte would leak the
    # signature one byte at a time.
    if not hmac.compare_digest(signature, self._sign(payload)):
        return {}
    try:
        body = json.loads(_b64decode(payload))
    except (ValueError, TypeError):
        return {}
    issued = body.get("t", 0)
    if self.max_age and (time.time() - issued) > self.max_age:
        return {}
    data = body.get("d")
    return data if isinstance(data, dict) else {}

load 

load(request: Any) -> Session

Dependency: the session for this request.

Source code in python/aether/_sessions.py 
133
134
135
136
137
138
def load(self, request: Any) -> Session:
    """Dependency: the session for this request."""
    raw = request.cookies.get(self.cookie)
    session = Session(self.decode(raw) if raw else {})
    _current.set(session)
    return session

cookie_header 

cookie_header(session: Session) -> str
Source code in python/aether/_sessions.py 
140
141
142
143
144
145
146
147
148
149
150
def cookie_header(self, session: Session) -> str:
    parts = [
        f"{self.cookie}={self.encode(dict(session))}",
        f"Path={self.path}",
        f"Max-Age={self.max_age}",
        "HttpOnly",
        f"SameSite={self.same_site}",
    ]
    if self.secure:
        parts.append("Secure")
    return "; ".join(parts)

middleware async 

middleware(request: Any, call_next: Any) -> Any
Source code in python/aether/_sessions.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
async def middleware(self, request: Any, call_next: Any) -> Any:
    token = _current.set(None)
    try:
        reply = await call_next(request)
    finally:
        session = _current.get(None)
        _current.reset(token)

    # Only when the handler actually touched it: rewriting an unchanged
    # session on every response is wasted bytes and a needless refresh.
    if isinstance(session, Session) and session.modified:
        reply.headers["set-cookie"] = self.cookie_header(session)
    return reply

Session 

Session(*args: Any, **kwargs: Any)

Bases: dict

A dict that remembers whether anything changed.

Source code in python/aether/_sessions.py 
43
44
45
def __init__(self, *args: Any, **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)
    self.modified = False

modified instance-attribute 

modified = False

clear 

clear() -> None
Source code in python/aether/_sessions.py 
55
56
57
def clear(self) -> None:
    super().clear()
    self.modified = True

pop 

pop(*args: Any) -> Any
Source code in python/aether/_sessions.py 
59
60
61
def pop(self, *args: Any) -> Any:
    self.modified = True
    return super().pop(*args)

update 

update(*args: Any, **kwargs: Any) -> None
Source code in python/aether/_sessions.py 
63
64
65
def update(self, *args: Any, **kwargs: Any) -> None:
    super().update(*args, **kwargs)
    self.modified = True

json_logging 

json_logging(level: int = INFO) -> None

Send Aether's logs to stderr as JSON lines.

Source code in python/aether/_logging.py 
55
56
57
58
59
60
61
62
def json_logging(level: int = logging.INFO) -> None:
    """Send Aether's logs to stderr as JSON lines."""
    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger("aether")
    root.handlers[:] = [handler]
    root.setLevel(level)
    root.propagate = False

JsonFormatter

Bases: Formatter

One JSON object per line, with any extra fields included.

format 

format(record: LogRecord) -> str
Source code in python/aether/_logging.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
def format(self, record: logging.LogRecord) -> str:
    payload: dict[str, Any] = {
        "time": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
        "level": record.levelname.lower(),
        "logger": record.name,
        "message": record.getMessage(),
    }
    for key, value in record.__dict__.items():
        if key not in _STANDARD and not key.startswith("_"):
            payload[key] = value
    if record.exc_info:
        payload["exception"] = self.formatException(record.exc_info)
    return json.dumps(payload, default=str)

TestClient 

TestClient(
    app: Any,
    *,
    host: str = "127.0.0.1",
    port: int | None = None,
    timeout: float = 10.0,
    **server_options: Any,
)

Runs an app for the duration of a with block.

Source code in python/aether/testing.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    app: Any,
    *,
    host: str = "127.0.0.1",
    port: int | None = None,
    timeout: float = 10.0,
    **server_options: Any,
) -> None:
    self.app = app
    self.host = host
    self.port = port or free_port()
    self.base_url = f"http://{host}:{self.port}"
    self.ws_url = f"ws://{host}:{self.port}"
    self.timeout = timeout
    self._options = server_options
    self._server: Any = None
    self._thread: threading.Thread | None = None
    self._client: httpx.Client | None = None

app instance-attribute 

app = app

host instance-attribute 

host = host

port instance-attribute 

port = port or free_port()

base_url instance-attribute 

base_url = f'http://{host}:{self.port}'

ws_url instance-attribute 

ws_url = f'ws://{host}:{self.port}'

timeout instance-attribute 

timeout = timeout

http property 

http: Client

start 

start() -> TestClient
Source code in python/aether/testing.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def start(self) -> "TestClient":
    self._server = self.app.build_server(
        self.host, self.port, **self._options
    )
    self._thread = threading.Thread(target=self._server.serve, daemon=True)
    self._thread.start()

    deadline = threading.Event()
    for _ in range(200):
        try:
            with socket.create_connection((self.host, self.port), timeout=0.2):
                break
        except OSError:
            deadline.wait(0.05)
    else:
        raise RuntimeError(f"server did not start on {self.base_url}")

    self._client = httpx.Client(base_url=self.base_url, timeout=self.timeout)
    return self

stop 

stop() -> None

Drain, then disconnect.

Order matters: closing the HTTP pool first would yank connections out from under requests the server is still finishing, which is exactly what graceful shutdown exists to avoid.

Source code in python/aether/testing.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def stop(self) -> None:
    """Drain, then disconnect.

    Order matters: closing the HTTP pool first would yank connections out
    from under requests the server is still finishing, which is exactly
    what graceful shutdown exists to avoid.
    """
    if self._server is not None:
        self._server.shutdown()
        self._server = None
    if self._thread is not None:
        self._thread.join(timeout=30)
        self._thread = None
    if self._client is not None:
        self._client.close()
        self._client = None

request 

request(method: str, path: str, **kwargs: Any) -> Response
Source code in python/aether/testing.py 
117
118
def request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
    return self.http.request(method, path, **kwargs)

get 

get(path: str, **kwargs: Any) -> Response
Source code in python/aether/testing.py 
120
121
def get(self, path: str, **kwargs: Any) -> httpx.Response:
    return self.http.get(path, **kwargs)

post 

post(path: str, **kwargs: Any) -> Response
Source code in python/aether/testing.py 
123
124
def post(self, path: str, **kwargs: Any) -> httpx.Response:
    return self.http.post(path, **kwargs)

put 

put(path: str, **kwargs: Any) -> Response
Source code in python/aether/testing.py 
126
127
def put(self, path: str, **kwargs: Any) -> httpx.Response:
    return self.http.put(path, **kwargs)

delete 

delete(path: str, **kwargs: Any) -> Response
Source code in python/aether/testing.py 
129
130
def delete(self, path: str, **kwargs: Any) -> httpx.Response:
    return self.http.delete(path, **kwargs)

head 

head(path: str, **kwargs: Any) -> Response
Source code in python/aether/testing.py 
132
133
def head(self, path: str, **kwargs: Any) -> httpx.Response:
    return self.http.head(path, **kwargs)

stream 

stream(method: str, path: str, **kwargs: Any)

A streaming response, for reading an SSE feed.

Source code in python/aether/testing.py 
137
138
139
140
141
@contextlib.contextmanager
def stream(self, method: str, path: str, **kwargs: Any):
    """A streaming response, for reading an SSE feed."""
    with self.http.stream(method, path, **kwargs) as response:
        yield response

websocket 

websocket(path: str)

An open WebSocket, as an async context manager.

async with client.websocket("/ws") as ws: await ws.send("hi") assert await ws.recv() == "hi"

Source code in python/aether/testing.py 
143
144
145
146
147
148
149
150
151
152
def websocket(self, path: str):
    """An open WebSocket, as an async context manager.

        async with client.websocket("/ws") as ws:
            await ws.send("hi")
            assert await ws.recv() == "hi"
    """
    import websockets

    return websockets.connect(f"{self.ws_url}{path}")

mcp 

mcp(
    method: str,
    params: dict | None = None,
    request_id: int = 1,
) -> Any

One JSON-RPC call against the app's MCP endpoint.

Returns the result, or raises with the JSON-RPC error message.

Source code in python/aether/testing.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
def mcp(self, method: str, params: dict | None = None, request_id: int = 1) -> Any:
    """One JSON-RPC call against the app's MCP endpoint.

    Returns the `result`, or raises with the JSON-RPC error message.
    """
    url = self.app.mcp_url or "/mcp"
    payload: dict[str, Any] = {"jsonrpc": "2.0", "id": request_id, "method": method}
    if params is not None:
        payload["params"] = params
    body = self.http.post(url, json=payload).json()
    if "error" in body:
        raise RuntimeError(f"MCP {method} failed: {body['error']}")
    return body.get("result")

call_tool 

call_tool(name: str, arguments: dict | None = None) -> Any

Invoke a capability the way an agent would.

Source code in python/aether/testing.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
def call_tool(self, name: str, arguments: dict | None = None) -> Any:
    """Invoke a capability the way an agent would."""
    result = self.mcp("tools/call", {"name": name, "arguments": arguments or {}})
    if result.get("isError"):
        text = "".join(part.get("text", "") for part in result.get("content", []))
        raise RuntimeError(f"tool {name} failed: {text}")
    if "structuredContent" in result:
        return result["structuredContent"]
    text = "".join(part.get("text", "") for part in result.get("content", []))
    try:
        return json.loads(text)
    except ValueError:
        return text