Database Adapters

AllSafe Fast uses a protocol-based storage layer that abstracts persistence behind four storage protocols. This design allows you to swap storage backends without changing application code — use the in-memory adapter during development and testing, and the SQLAlchemy adapter in production.

Storage Protocols

The storage layer is defined by four protocols that together form the StorageAdapter:

from typing import Protocol, Awaitable

# Simplified protocol definitions
class UserStore(Protocol):
    async def create(self, user: User) -> User: ...
    async def get_by_id(self, user_id: UUID) -> User | None: ...
    async def get_by_email(self, email: str) -> User | None: ...
    async def update(self, user: User) -> User: ...
    async def delete(self, user_id: UUID) -> None: ...

class StorageAdapter:
    # Bundles all four stores into a single adapter
    users: UserStore
    accounts: AccountStore
    sessions: SessionStore
    verifications: VerificationStore

In-Memory Adapter (Development)

The in-memory adapter stores all data in Python dictionaries guarded by asyncio.Lock for safe concurrent access. It is designed for development and testing — it does not persist data across restarts.

Features

• asyncio.Lock-guarded — All operations are protected by async locks to prevent race conditions in concurrent test scenarios.
• No persistence — Data is lost when the process exits. Perfect for testing.
• Zero configuration — No database or external dependencies required.
• Fast — No network overhead, no serialization cost.

Usage

from allsafe_fast import Auth, AuthConfig

# The in-memory adapter is used by default when no database URL is configured
auth = Auth(AuthConfig(
    secret_key="dev-secret-key-change-in-production",
    # No ALLSAFE_DATABASE_URL set — in-memory adapter is used
))

# Or explicitly for testing
auth = Auth(AuthConfig(
    secret_key="test-secret-key",
    env="test",  # Use test environment
))

How It Works

import asyncio

class InMemoryUserStore:
    def __init__(self):
        self._users: dict[UUID, User] = {}
        self._email_index: dict[str, UUID] = {}
        self._lock = asyncio.Lock()

    async def create(self, user: User) -> User:
        async with self._lock:
            self._users[user.id] = user
            self._email_index[user.email] = user.id
            return user

    async def get_by_email(self, email: str) -> User | None:
        async with self._lock:
            user_id = self._email_index.get(email)
            return self._users.get(user_id) if user_id else None

SQLAlchemy Adapter (Production)

The SQLAlchemy adapter provides async PostgreSQL persistence using asyncpg. It is the recommended adapter for production deployments.

Features

• Async PostgreSQL — Uses asyncpg driver via postgresql+asyncpg:// connection string.
• ARRAY(String) types — Roles and permissions are stored as PostgreSQL arrays for efficient querying.
• UUID primary keys — All tables use UUID primary keys for globally unique identifiers.
• CASCADE deletes — Foreign keys with CASCADE ensure clean deletion of related records.
• Proper indexing — Unique and regular indexes on frequently queried columns.

Configuration

from allsafe_fast import Auth, AuthConfig

# Set the database URL via environment variable
# ALLSAFE_DATABASE_URL=postgresql+asyncpg://user:pass@host:5432/dbname

auth = Auth(AuthConfig(
    secret_key="your-production-secret-key",
    database_url="postgresql+asyncpg://user:pass@localhost:5432/myapp",
    env="production",
))

# Or via environment variables only
# export ALLSAFE_DATABASE_URL="postgresql+asyncpg://..."
# export ALLSAFE_SECRET_KEY="your-secret-key"
# export ALLSAFE_ENV="production"
auth = Auth(AuthConfig())  # Reads from environment

Table Schema

The SQLAlchemy adapter creates four tables with the following key characteristics:

Database URL Format

# PostgreSQL with asyncpg (recommended)
postgresql+asyncpg://username:password@host:port/database

# Example with all components
postgresql+asyncpg://myuser:mypassword@localhost:5432/myapp

# Example with connection pool parameters
postgresql+asyncpg://myuser:mypassword@localhost:5432/myapp?pool_size=20&max_overflow=10

Configuration Summary

Production Considerations

When using the SQLAlchemy adapter in production, consider the following:

• Connection pooling — Configure pool size and max overflow based on your expected load. The async engine manages a pool of connections.
• Database migrations — Run allsafe db migrate as part of your deployment pipeline to keep the schema up to date.
• Session cleanup — Set up a periodic task to delete expired sessions and verification tokens. The expires_at indexes make this efficient.
• Backup strategy — Ensure your PostgreSQL database is backed up regularly. Auth tables contain user data that cannot be reconstructed.
• SSL/TLS — Use SSL for database connections in production. Add ?ssl=require to the connection URL.

# Production database URL with SSL
# ALLSAFE_DATABASE_URL=postgresql+asyncpg://user:pass@host:5432/db?ssl=require

# Periodic session cleanup (run as a background task)
async def cleanup_expired_sessions(auth: Auth):
    await auth.storage.sessions.delete_expired()
    await auth.storage.verifications.delete_expired()

Planned Adapters

AllSafe Fast's protocol-based storage design makes it straightforward to add new adapters. The following adapters are planned for future releases:

Implementing a Custom Adapter

If you need a storage backend that AllSafe Fast doesn't support yet, you can implement the four storage protocols yourself:

from allsafe_fast import UserStore, AccountStore, SessionStore, VerificationStore, StorageAdapter

# Implement the protocols for your storage backend
class MyUserStore:
    async def create(self, user): ...
    async def get_by_id(self, user_id): ...
    # ... implement all protocol methods

# Bundle them into a StorageAdapter
adapter = StorageAdapter(
    users=MyUserStore(),
    accounts=MyAccountStore(),
    sessions=MySessionStore(),
    verifications=MyVerificationStore(),
)

# Pass to Auth
auth = Auth(config, storage_adapter=adapter)