Stores — Aquilia Documentation
Comprehensive guide and documentation for Stores in the Aquilia framework. View API reference, examples, and implementation patterns.
Sessions / Stores Session Stores Session stores manage persistence and loading of active sessions. Aquilia exposes a typed SessionStore protocol, implemented out-of-the-box by MemoryStore and FileStore. SessionStore Protocol All storage backends must implement the SessionStore protocol. Stores are purely responsible for persistence — they do not enforce expiration policies or idle timeouts. from typing import Protocol from aquilia.sessions import Session, SessionID class SessionStore(Protocol): """Protocol for session storage backends.""" async def load(self, session_id: SessionID) -> Session | None: """Load session by ID. Returns None if not found or corrupted.""" ... async def save(self, session: Session) -> None: """Persist session. Sets session version and resets dirty state.""" ... async def delete(self, session_id: SessionID) -> None: """Delete session by ID.""" ... async def exists(self, session_id: SessionID) -> bool: """Check if session exists in store.""" ... async def list_by_principal(self, principal_id: str) -> list[Session]: """Find active sessions belonging to user principal.""" ... async def count_by_principal(self, principal_id: str) -> int: """Count active sessions belonging to user principal.""" ... async def cleanup_expired(self) -> int: """Remove expired sessions from storage. Returns removed count.""" ... async def shutdown(self) -> None: """Gracefully release storage client resources.""" ... MemoryStore (In-Memory LRU) An in-memory store utilizing an OrderedDict to achieve O(1) LRU eviction. Features locking for asynchronous consistency and secondary index tables for principal queries. MemoryStore is not persistent across server reboots. from aquilia.sessions import MemoryStore # Manual construction (requires max_sessions limit) store = MemoryStore(max_sessions=10000) # Factory presets optimized for specific payloads: # Web: High capacity (25k max sessions) store = MemoryStore.web_optimized() # API: Medium capacity (15k max sessions) store = MemoryStore.api_optimized() # Mobile: Medium capacity (15k max sessions) store = MemoryStore.development_focused() # 1k capacity # High-throughput: Max capacity (50k max sessions) store = MemoryStore.high_throughput() LRU Eviction Mechanism When capacity is reached, new saves trigger LRU eviction, popping the oldest accessed session from the store: store = MemoryStore(max_sessions=3) # 1. Fill the store await store.save(session_a) # OrderedDict: [A] await store.save(session_b) # OrderedDict: [A, B] await store.save(session_c) # OrderedDict: [A, B, C] # 2. Access A to make it recently-used await store.load(session_a.id) # OrderedDict moves A to end: [B, C, A] # 3. Save D triggers LRU eviction of B await store.save(session_d) # OrderedDict evicts B: [C, A, D] print(await store.exists(session_b.id)) # False (evicted!) FileStore (JSON File-based) Saves each session as an individual JSON file. Incorporates atomic write protocols (temp file creation + atomic rename) to ensure file corruption does not occur during system failures. FileStore is suitable for low-traffic development. from aquilia.sessions import FileStore # Constructor takes directory folder path store = FileStore(directory="/var/lib/aquilia/sessions") # Session IDs are strictly validated to prevent path-traversal attacks. # Data is formatted inside sess_*.json files. Building a Custom Store (Redis Example) Implement the SessionStore protocol to define your custom storage (such as Redis, DynamoDB, or PostgreSQL): import json from datetime import datetime, timezone from aquilia.sessions import Session, SessionID class RedisSessionStore: """Pluggable Redis backend implementing SessionStore protocol.""" def __init__(self, client, prefix: str = "aquilia:sess:"): self.client = client self.prefix = prefix def _key(self, sid: SessionID) -> str: return f" " async def load(self, session_id: SessionID) -> Session | None: raw = await self.client.get(self._key(session_id)) if not raw: return None return Session.from_dict(json.loads(raw)) async def save(self, session: Session) -> None: key = self._key(session.id) payload = json.dumps(session.to_dict()) # Calculate remaining TTL seconds dynamically ttl = 3600 if session.expires_at: now = datetime.now(timezone.utc) ttl = max(int((session.expires_at - now).total_seconds()), 1) await self.client.setex(key, ttl, payload) # Keep track of principal query indices if session.principal: p_key = f" principal: " await self.client.sadd(p_key, str(session.id)) session.mark_clean() async def delete(self, session_id: SessionID) -> None: session = await self.load(session_id) key = self._key(session_id) await self.client.delete(key) if session and session.principal: p_key = f" principal: " await self.client.srem(p_key, str(session_id)) async def exists(self, session_id: SessionID) -> bool: return bool(await self.client.exists(self._key(session_id))) async def list_by_principal(self, principal_id: str) -> list[Session]: p_key = f" principal: " sids = await self.client.smembers(p_key) sessions = [] for sid_str in sids: try: sid = SessionID.from_string(sid_str) sess = await self.load(sid) if sess: sessions.append(sess) except Exception: continue return sessions async def count_by_principal(self, principal_id: str) -> int: p_key = f" principal: " return await self.client.scard(p_key) async def cleanup_expired(self) -> int: # No-op: Redis automatically evicts keys using setex TTLs return 0 async def shutdown(self) -> None: await self.client.close() # Register the store in the SessionEngine (engine receives a single store instance) from aquilia.sessions import SessionEngine engine = SessionEngine( policy=policy, store=RedisSessionStore(redis_client), transport=transport ) Store Comparison Feature MemoryStore FileStore Custom (Redis) ))} )
Go to Homepage