# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""Session middleware — per-app session management.
Extracts session token from cookie, reconnects or creates session,
injects into scope["session"], and sets Set-Cookie on new sessions.
Runs in per-app middleware chains (after Dispatcher creates the request).
Accesses request.cookies for session token extraction.
Config (session_middleware section in YAML):
cookie_name: "session_id" (default)
secure: false (default)
samesite: "lax" (default)
Enable per-app:
apps:
myapp:
middleware:
session: on
"""
from __future__ import annotations
from collections.abc import MutableMapping
from typing import TYPE_CHECKING, Any
from . import BaseMiddleware
from ..response import make_cookie
if TYPE_CHECKING:
from ..types import ASGIApp, Receive, Scope, Send
__all__ = ["SessionMiddleware"]
class SessionMiddleware(BaseMiddleware):
"""Per-app session middleware. Manages session lifecycle via cookies."""
middleware_name = "session"
middleware_order = 450
middleware_default = False
__slots__ = ("_cookie_name", "_secure", "_samesite")
def __init__(
self,
app: ASGIApp,
*,
cookie_name: str = "session_id",
secure: bool = False,
samesite: str = "lax",
**kwargs: Any,
) -> None:
"""Initialize session middleware.
Args:
app: Next ASGI app in chain.
cookie_name: Name of the session cookie.
secure: Set Secure flag on cookie.
samesite: SameSite policy for cookie.
**kwargs: Additional config (ignored).
"""
super().__init__(app, **kwargs)
self._cookie_name = cookie_name
self._secure = secure
self._samesite = samesite
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Extract/create session and inject into scope.
Reads session token from cookie, reconnects or creates session,
sets scope["session"]. Adds Set-Cookie header for new sessions.
Args:
scope: ASGI scope dictionary.
receive: ASGI receive callable.
send: ASGI send callable.
"""
if scope["type"] not in ("http", "websocket"):
await self.app(scope, receive, send)
return
server = self.server
request = server.request_registry.current
session_id = request.cookies.get(self._cookie_name)
session = None
if session_id:
session = server.session_store.get(session_id)
is_new = session is None
if is_new:
auth = scope.get("auth")
session = server.session_store.create(auth=auth)
assert session is not None
scope["session"] = session
if is_new:
cookie_header = make_cookie(
self._cookie_name,
session.id,
max_age=session.meta["ttl"],
httponly=True,
secure=self._secure,
samesite=self._samesite,
)
async def send_with_cookie(message: MutableMapping[str, Any]) -> None:
if message["type"] == "http.response.start":
headers = list(message.get("headers", []))
headers.append((cookie_header[0].encode(), cookie_header[1].encode()))
message = {**message, "headers": headers}
await send(message)
await self.app(scope, receive, send_with_cookie)
else:
await self.app(scope, receive, send)
if __name__ == "__main__":
pass