Custom User Models

AllSafe Fast's domain User model is a plain dataclass — not an ORM model. This design decision allows AllSafe Fast to work with any storage backend and, crucially, lets you use your existing application user models by implementing the storage protocol interface. This page explains how the domain model works and how to integrate your own user table.

How the AllSafe User Model Works

AllSafe Fast's internal User model is a plain Python dataclass. It is a domain model — it represents the concept of a user in the authentication system, independent of how that user is stored:

from dataclasses import dataclass
from datetime import datetime
from uuid import UUID

@dataclass
class User:
    id: UUID
    email: str
    email_verified: bool
    name: str | None
    image: str | None
    is_active: bool
    roles: list[str]
    permissions: list[str]
    created_at: datetime
    updated_at: datetime

SQLAlchemy ORM Models vs. Domain Models

When using the SQLAlchemy adapter, AllSafe Fast maintains two representations of a user:

Representation	Type	Purpose
ORM Model	AllSafeUser (SQLAlchemy)	Maps to the allsafe_users table, handles database operations
Domain Model	User (dataclass)	Used throughout the auth system, provider-agnostic

The SQLAlchemy adapter converts between these two representations using a _to_domain method:

# Inside the SQLAlchemy UserStore
class SQLAlchemyUserStore:
    def _to_domain(self, orm_user: AllSafeUser) -> User:
        # Convert ORM model to domain model
        return User(
            id=orm_user.id,
            email=orm_user.email,
            email_verified=orm_user.email_verified,
            name=orm_user.name,
            image=orm_user.image,
            is_active=orm_user.is_active,
            roles=orm_user.roles,          # ARRAY(String) -> list[str]
            permissions=orm_user.permissions,  # ARRAY(String) -> list[str]
            created_at=orm_user.created_at,
            updated_at=orm_user.updated_at,
        )

    def _to_orm(self, domain_user: User) -> AllSafeUser:
        # Convert domain model to ORM model
        return AllSafeUser(
            id=domain_user.id,
            email=domain_user.email,
            # ... map all fields
        )

Using Existing Application User Models

If your application already has a user table (e.g., app_users with its own schema), you can integrate it with AllSafe Fast by implementing the UserStoreProtocol. This lets AllSafe Fast use your existing table instead of creating the allsafe_users table.

The Storage Protocol Interface

from typing import Protocol
from uuid import UUID
from allsafe_fast import User

class UserStoreProtocol(Protocol):
    # Protocol that any custom user store must implement
    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: ...

Example: Implementing a Custom UserStore

Here is a complete example of implementing a custom UserStore that uses your application's existing user table:

from uuid import UUID, uuid4
from datetime import datetime, timezone
from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
from allsafe_fast import User, NotFoundError, ConflictError

# Your existing application's ORM model
class AppUser(Base):
    __tablename__ = "app_users"
    id = Column(UUID, primary_key=True, default=uuid4)
    email = Column(String, unique=True, index=True)
    email_verified = Column(Boolean, default=False)
    display_name = Column(String, nullable=True)
    avatar_url = Column(String, nullable=True)
    is_active = Column(Boolean, default=True)
    role = Column(String, default="user")  # Single role column
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    updated_at = Column(DateTime, onupdate=lambda: datetime.now(timezone.utc))

# Your custom UserStore implementation
class CustomUserStore:
    def __init__(self, session_factory):
        self.session_factory = session_factory

    def _to_domain(self, orm: AppUser) -> User:
        # Map your ORM model to AllSafe's domain model
        return User(
            id=orm.id,
            email=orm.email,
            email_verified=orm.email_verified,
            name=orm.display_name,       # display_name -> name
            image=orm.avatar_url,        # avatar_url -> image
            is_active=orm.is_active,
            roles=[orm.role],            # single role -> list
            permissions=[],              # no permissions in your model
            created_at=orm.created_at,
            updated_at=orm.updated_at or orm.created_at,
        )

    def _to_orm_fields(self, domain: User) -> dict:
        # Map domain model fields to your ORM model's columns
        return {
            "email": domain.email,
            "email_verified": domain.email_verified,
            "display_name": domain.name,
            "avatar_url": domain.image,
            "is_active": domain.is_active,
            "role": domain.roles[0] if domain.roles else "user",
        }

    async def create(self, user: User) -> User:
        async with self.session_factory() as session:
            existing = await session.execute(
                select(AppUser).where(AppUser.email == user.email)
            )
            if existing.scalar_one_or_none():
                raise ConflictError("Email already registered")

            orm = AppUser(id=user.id, **self._to_orm_fields(user))
            session.add(orm)
            await session.commit()
            return self._to_domain(orm)

    async def get_by_id(self, user_id: UUID) -> User | None:
        async with self.session_factory() as session:
            result = await session.execute(
                select(AppUser).where(AppUser.id == user_id)
            )
            orm = result.scalar_one_or_none()
            return self._to_domain(orm) if orm else None

    async def get_by_email(self, email: str) -> User | None:
        async with self.session_factory() as session:
            result = await session.execute(
                select(AppUser).where(AppUser.email == email)
            )
            orm = result.scalar_one_or_none()
            return self._to_domain(orm) if orm else None

    async def update(self, user: User) -> User:
        async with self.session_factory() as session:
            await session.execute(
                update(AppUser)
                .where(AppUser.id == user.id)
                .values(**self._to_orm_fields(user))
            )
            await session.commit()
            return await self.get_by_id(user.id)

    async def delete(self, user_id: UUID) -> None:
        async with self.session_factory() as session:
            await session.execute(
                delete(AppUser).where(AppUser.id == user_id)
            )
            await session.commit()

Wiring It Into Auth

from allsafe_fast import Auth, AuthConfig, StorageAdapter
from allsafe_fast.storage.sqlalchemy import SQLAlchemyAccountStore, SQLAlchemySessionStore, SQLAlchemyVerificationStore

# Create your custom user store
user_store = CustomUserStore(session_factory)

# Use built-in stores for accounts, sessions, verifications
# (or implement those too if needed)
account_store = SQLAlchemyAccountStore(session_factory)
session_store = SQLAlchemySessionStore(session_factory)
verification_store = SQLAlchemyVerificationStore(session_factory)

# Bundle into a StorageAdapter
adapter = StorageAdapter(
    users=user_store,
    accounts=account_store,
    sessions=session_store,
    verifications=verification_store,
)

# Create Auth with your custom adapter
auth = Auth(
    config=AuthConfig(secret_key="your-secret-key"),
    storage_adapter=adapter,
)

Field Mapping Considerations

When mapping between your model and AllSafe's domain model, keep these points in mind:

• Required fields — The User domain model requires id, email, email_verified, is_active, roles, permissions, created_at, and updated_at. If your model doesn't have one of these, provide a default.
• Roles and permissions — The domain model uses list[str]. If your model uses a single role string, wrap it in a list. If you have no permissions concept, return an empty list.
• Nullable fields — name and image can be None. Map them to your model's nullable columns.
• Timestamps — Ensure created_at and updated_at are timezone-aware datetimes.