# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""Session middleware — session management driven by the request cookie.
Extracts the session token from the request cookie (read from the ASGI scope,
not from a request object), reconnects or creates the session, injects it into
scope["session"], and sets Set-Cookie on new sessions.
Like the other server middleware (authentication, cache), it works on the
scope: it reads ``scope["_headers"]`` (populated by ``@headers_dict``) and
parses the Cookie header itself, so it does not depend on the request object
having been created yet.
Config:
cookie_name: "session_id" (default)
secure: false (default)
samesite: "lax" (default)
"""
from __future__ import annotations
from collections.abc import MutableMapping
from http.cookies import SimpleCookie
from typing import TYPE_CHECKING, Any
from . import BaseMiddleware, headers_dict
from ..response import make_cookie
if TYPE_CHECKING:
from ..types import ASGIApp, Receive, Scope, Send
__all__ = ["SessionMiddleware"]
[docs]
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")
[docs]
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
def _cookie_value(self, scope: Scope) -> str | None:
"""Return the session cookie value from the scope, or None if absent."""
cookie_header = scope["_headers"].get("cookie")
if not cookie_header:
return None
jar: SimpleCookie = SimpleCookie()
jar.load(cookie_header)
morsel = jar.get(self._cookie_name)
return morsel.value if morsel is not None else None
@headers_dict
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Extract/create session and inject into scope.
Reads the session token from the request cookie (via ``scope["_headers"]``),
reconnects or creates the session, sets scope["session"]. Adds a
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
session_id = self._cookie_value(scope)
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