aiohttp-session

import asyncio import os from cryptography import fernet from aiohttp import web from aiohttp_session import get_session, setup from aiohttp_session.cookie_storage import EncryptedCookieStorage async def handler(request): session = await get_session(request) last_visit = session.get('last_visit', 'Never') session['last_visit'] = str(request.app['current_time']) text = f"Last visited: {last_visit}\nHello, current time is {session['last_visit']}" return web.Response(text=text) async def make_app(): app = web.Application() # Generate a Fernet key. In production, this should be stored securely # and loaded from environment variables or a secret management system. # Using a dummy key for demonstration purposes. fernet_key_str = os.environ.get('AIOHTTP_SESSION_KEY', None) if fernet_key_str is None: # WARNING: DO NOT generate a new key on every app startup in production! # A new key invalidates all existing sessions. Load from a persistent source. fernet_key = fernet.Fernet.generate_key() print(f"Generated new Fernet key (for demo only): {fernet_key.decode()}\nSet AIOHTTP_SESSION_KEY env var in production.") else: fernet_key = fernet_key_str.encode() f = fernet.Fernet(fernet_key) setup(app, EncryptedCookieStorage(f)) app['current_time'] = 'Not set yet' app.router.add_get('/', handler) return app async def main(): app = await make_app() runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, 'localhost', 8080) await site.start() print("Server started at http://localhost:8080") try: while True: app['current_time'] = asyncio.get_event_loop().time() await asyncio.sleep(1) except asyncio.CancelledError: pass finally: await runner.cleanup() if __name__ == '__main__': try: asyncio.run(main()) except KeyboardInterrupt: print("Server stopped.")