Skip to content

UserRepository

The adapter between CRUDAuth's logical field names and your user model's actual columns. Configure the mapping with column_map= on CRUDAuth.

crudauth.repository.UserRepository 

UserRepository(
    model: type[Any],
    column_map: dict[str, str] | None = None,
    register_extra_fields: Iterable[str] | None = None,
    login_fields: Iterable[str] | None = None,
    recovery: str | None = "email",
)

Logical-field data access over the app's user model.

The single boundary to the user table: other layers read and write through these methods (get_by_email, resolve_login, create, update, token_version, ...), addressing logical fields (email, hashed_password, ...) that the column_map resolves to real columns. Reachable as auth.repo.

Example 
user = await auth.repo.get_by_email(db, "alice@example.com")
if user is not None and auth.repo.is_active(user):
    ...

col 

col(logical: str) -> str

Resolve a logical field name to the actual model attribute name.

has 

has(logical: str) -> bool

Whether the model actually has the column for logical.

is_unique_column 

is_unique_column(logical: str) -> bool

Whether the resolved column for logical is single-field unique.

Detects column-level unique=True, a single-column UniqueConstraint, and a single-column unique Index - all through column_map (the resolved column name). Composite uniqueness does NOT count: a multi-column unique key doesn't make one field a safe first-match-wins login key, so a composite-only field is treated as non-unique (and the construction check raises for it).

get 

get(user: Any, logical: str, default: Any = None) -> Any

Read a logical field off user, or default if the column is absent.

set_field 

set_field(user: Any, logical: str, value: Any) -> None

Set a logical field on user by its resolved column name.

get_by_id async 

get_by_id(db: AsyncSession, user_id: Any) -> Any | None

Fetch the user by primary key, or None.

Parameters:

Name Type Description Default
db AsyncSession

Active async session.

 required
user_id Any

PK value; coerced to the column's Python type first (so a string "42" from a token matches an int PK on Postgres).

 required

Returns:

Type Description
Any | None

The user row, or None if absent or user_id can't be coerced.

get_by_field async 

get_by_field(
    db: AsyncSession, logical: str, value: Any
) -> Any | None

Fetch the user whose logical field equals value, or None.

Email is canonicalized before matching (the column is stored canonical); every other field matches as-is.

get_by_email async 

get_by_email(db: AsyncSession, email: str) -> Any | None

Fetch the user by (canonicalized) email, or None.

get_by_username async 

get_by_username(
    db: AsyncSession, username: str
) -> Any | None

Fetch the user by username, or None.

resolve_login async 

resolve_login(
    db: AsyncSession, identifier: str
) -> Any | None

Resolve a login identifier against login_fields, in order; first match wins.

Replaces the old @-heuristic: the contract decides which fields a login identifier may match. Safe because every login field is asserted unique at construction, so a match is unambiguous.

get_by_oauth async 

get_by_oauth(
    db: AsyncSession, provider: str, provider_user_id: str
) -> Any | None

Look up by {provider}_id.

Note

Assumes provider is a validated/registered provider name (the OAuth router checks it before this is reached) - it builds an attribute name from the argument, so an unvalidated value would probe arbitrary columns (it returns None for any column the model lacks, so the blast radius is a missed lookup, not a leak).

gated_register_fields 

gated_register_fields(
    schema_fields: Iterable[str],
) -> set[str]

Which of schema_fields are crudauth privileged fields (always dropped).

droppable_register_fields 

droppable_register_fields(
    schema_fields: Iterable[str],
) -> set[str]

Which of schema_fields map to a real model column but are not privileged and not opted in, so registration silently drops them.

These are the fields a developer most likely expects to persist (e.g. a full_name column they added to register_schema) and won't, until they add the name to register_extra_fields.

filter_registration_data 

filter_registration_data(
    data: dict[str, Any],
) -> dict[str, Any]

Keep only the allowlisted registration fields; drop everything else.

A field survives only if it is in :data:REGISTRATION_ALLOWED_FIELDS or was opted in via register_extra_fields (matched by logical or mapped column name) - and is never one of crudauth's privileged logical fields (is_superuser, email_verified, hashed_password, the oauth linkage, the PK), which stay gated even if mistakenly opted in. Unknown app columns (role, credits, ...) are dropped unless explicitly opted in, so a custom register_schema can't turn /register into a privilege-escalation or mass-assignment endpoint.

filter_provisioning_data 

filter_provisioning_data(
    data: dict[str, Any],
) -> dict[str, Any]

Keep only app columns from a new_user_fields callback.

Drops any key that is a crudauth logical field (by logical or mapped column name) and logs a warning, so the callback can fill the app's own columns at signup but can never override identity, privilege, or state crudauth owns (email, hashed_password, is_superuser, email_verified, the oauth linkage, the PK). The dropped key keeps crudauth's authoritative value.

Note

The callback runs per signup, so a misconfigured one would drop the same key on every registration. Each distinct dropped field is warned only ONCE per process (deduped across the constant new_user_defaults and the runtime callback) so a standing misconfiguration can't flood the logs.

create async 

create(db: AsyncSession, data: dict[str, Any]) -> Any

Insert a user from logical-field data and return the row.

Note

Owns the transaction boundary - commits and refreshes on the passed-in session. Apps using a request-scoped "commit at the end" pattern should know auth writes commit eagerly.

Note

Email is canonicalized off the resolved column: kwargs is keyed by actual column names, so a column_map that renames email would otherwise be stored un-normalized.

update async 

update(
    db: AsyncSession, user: Any, data: dict[str, Any]
) -> Any

Apply logical-field data to user and return it.

Note

Commits and refreshes eagerly on the passed-in session (see create).

recovery_verified 

recovery_verified(user: Any) -> bool

Whether the contract's recovery factor is proven controlled.

Email recovery reads email_verified; another factor reads {factor}_verified; recovery=None is never verified. This is the general meaning of "verified" - email is the special case, not the concept.

mark_recovery_verified async 

mark_recovery_verified(db: AsyncSession, user: Any) -> None

Set the contract's recovery-factor verified flag (the verify-flow write).

token_version 

token_version(user: Any) -> int

The user's credential epoch (0 if the model has no such column).

increment_token_version async 

increment_token_version(
    db: AsyncSession, user: Any
) -> None

Bump the credential epoch, revoking outstanding bearer tokens.

A no-op when the model has no token_version column (bearer tokens then simply aren't epoch-revocable; the limitation is documented).

to_dict 

to_dict(user: Any) -> dict[str, Any]

Project a user row onto the logical contract (for hooks).

Note

Contract-only by design - the dict holds the crudauth logical fields (id, email, ...), not your app's own columns. A hook that needs full_name should re-load the row via db using the id, not expect it in this dict.