aiohttp-session
raw JSON → 2.12.1 verified Thu Apr 16 auth: no python
aiohttp-session is a Python library that provides robust session management for aiohttp.web applications. It enables developers to store user-specific data across multiple requests using various storage backends, including secure encrypted cookies, Redis, and Memcached. The library maintains an active development pace, with the current version being 2.12.1, and focuses on stability, security, and compatibility with the latest Python and aiohttp versions.
pip install aiohttp-session Common errors
error RuntimeError: Cannot install aiohttp_session[aioredis] without aioredis library ↓
cause Prior to v2.12.0, the `aioredis` extra explicitly required the `aioredis` package. Post v2.12.0, it shifted to `redis`.
fix
If on
aiohttp-session < 2.12.0, ensure pip install aioredis. If on aiohttp-session >= 2.12.0, ensure pip install redis or pip install aiohttp-session[aioredis]. error cryptography.fernet.InvalidToken: Message too short ↓
cause This error often indicates that the `EncryptedCookieStorage` received a cookie it couldn't decrypt. Common causes include a corrupted session cookie, or the application's encryption key changing between server restarts.
fix
Verify that the Fernet encryption key used by
EncryptedCookieStorage is consistent across all application instances and restarts. Ensure that Fernet.generate_key() is not called repeatedly in production. Users with invalid cookies will simply receive a new session. error TypeError: session_middleware() got an unexpected keyword argument 'storage' ↓
cause The `session_middleware` function is typically called as `session_middleware(storage_instance)` and then passed to `app.middlewares`. If `setup(app, storage_instance)` is used, you generally don't interact with `session_middleware` directly.
fix
Use
setup(app, storage_instance) to configure the session middleware, or pass session_middleware(storage_instance) directly to the middlewares list when creating web.Application. error KeyError: 'AIOHTTP_SESSION' (or other cookie name) or session data not persisting between requests. ↓
cause This usually means the session middleware is not correctly registered or the `aiohttp.web.Application` instance receiving requests is not the one configured with the session middleware.
fix
Ensure
aiohttp_session.setup(app, storage) is called on the web.Application instance that handles your routes, or that session_middleware(storage) is properly included in the middlewares list of web.Application. Warnings
breaking Support for Python 3.7 was dropped in `aiohttp-session` v2.12.1. Applications running on Python 3.7 will need to upgrade their Python version. ↓
fix Upgrade your Python environment to 3.8 or higher.
breaking Starting with `aiohttp-session` v2.12.0, `RedisStorage` migrated its core dependency from `aioredis` to `redis`. If you were manually installing `aioredis` for Redis sessions without using the `[aioredis]` extra, you must now install `redis` instead. ↓
fix If using RedisStorage, ensure `pip install redis` or `pip install aiohttp-session[aioredis]` is used. The `[aioredis]` extra will now install `redis`.
gotcha Using `SimpleCookieStorage` directly is insecure as it stores session data in plain JSON. It is strictly recommended for testing or development purposes only. ↓
fix For production environments, always use `EncryptedCookieStorage` or a server-side storage like `RedisStorage` or `MemcachedStorage`.
gotcha To prevent session fixation attacks, always call `new_session()` in login views to generate a fresh session identity for authenticated users. ↓
fix Replace `await get_session(request)` with `await new_session(request)` in code paths where a user's authentication status changes (e.g., after login).
gotcha The `EncryptedCookieStorage` requires a strong 32-byte secret key (or a `Fernet` object) for encryption. Generating a new key on every application restart will invalidate all existing sessions, logging out users. ↓
fix Generate a `Fernet.generate_key()` once and persist it securely (e.g., in an environment variable or a secret management service) across application restarts. Pass this consistent key to `EncryptedCookieStorage`.
Install
pip install aiohttp-session[secure] pip install aiohttp-session[aioredis] Imports
- get_session
from aiohttp_session import get_session - setup
from aiohttp_session import setup - EncryptedCookieStorage
from aiohttp_session.cookie_storage import EncryptedCookieStorage - RedisStorage
from aiohttp_session.redis_storage import RedisStorage - Fernet
from cryptography.fernet import Fernet
Quickstart
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.")