API Reference

User module for user account management and authentication.

This module provides comprehensive user management including account creation, profile updates, authentication, MFA support, email verification, and admin approval workflows.

Exports

• CRUD: get_all_users, get_users_number, get_users_with_pagination, get_user_by_username, get_user_by_email, get_user_by_id, get_users_admin, create_user, create_signup_user, edit_user, approve_user, verify_user_email, update_user_photo, delete_user
• Schemas: UsersBase, Users, UsersRead, UsersMe, UsersSignup, UsersCreate, UsersEditPassword, UsersListResponse
• Enums: Gender, Language, WeekDay, UserAccessType
• Utils: get_user_by_id_or_404, get_admin_users_or_404, check_user_is_active, create_user_default_data, save_user_image_file, delete_user_photo_filesystem

Gender

Bases: Enum

User gender enumeration.

Attributes:

Name	Type	Description
MALE		Male gender.
FEMALE		Female gender.
UNSPECIFIED		Unspecified or undisclosed gender.

Source code in backend/app/users/users/schema.py

class Gender(Enum):
    """
    User gender enumeration.

    Attributes:
        MALE: Male gender.
        FEMALE: Female gender.
        UNSPECIFIED: Unspecified or undisclosed gender.
    """

    MALE = "male"
    FEMALE = "female"
    UNSPECIFIED = "unspecified"

Language

Bases: Enum

Supported application languages.

Attributes:

Name	Type	Description
CATALAN		Catalan (ca).
CHINESE_SIMPLIFIED		Simplified Chinese (zh-Hans).
CHINESE_TRADITIONAL		Traditional Chinese (zh-Hant).
GERMAN		German (de).
FRENCH		French (fr).
GALICIAN		Galician (gl).
ITALIAN		Italian (it).
DUTCH		Dutch (nl).
PORTUGUESE		Portuguese (pt-PT).
SLOVENIAN		Slovenian (sl).
SWEDISH		Swedish (sv).
SPANISH		Spanish (es).
ENGLISH		English (en).
POLISH		Polish (pl).
TURKISH		Turkish (tr).
UKRAINIAN		Ukrainian (uk).
ROMANIAN		Romanian (ro).
NORWEGIAN		Norwegian Bokmål (nb).
DANISH		Danish (da).
FINNISH		Finnish (fi).
CZECH		Czech (cs).
GREEK		Greek (el).
HUNGARIAN		Hungarian (hu).
BULGARIAN		Bulgarian (bg).
CROATIAN		Croatian (hr).
SERBIAN		Serbian (sr).
SLOVAK		Slovak (sk).
LITHUANIAN		Lithuanian (lt).
LATVIAN		Latvian (lv).
ESTONIAN		Estonian (et).

Source code in backend/app/users/users/schema.py

class Language(Enum):
    """
    Supported application languages.

    Attributes:
        CATALAN: Catalan (ca).
        CHINESE_SIMPLIFIED: Simplified Chinese (zh-Hans).
        CHINESE_TRADITIONAL: Traditional Chinese (zh-Hant).
        GERMAN: German (de).
        FRENCH: French (fr).
        GALICIAN: Galician (gl).
        ITALIAN: Italian (it).
        DUTCH: Dutch (nl).
        PORTUGUESE: Portuguese (pt-PT).
        SLOVENIAN: Slovenian (sl).
        SWEDISH: Swedish (sv).
        SPANISH: Spanish (es).
        ENGLISH: English (en).
        POLISH: Polish (pl).
        TURKISH: Turkish (tr).
        UKRAINIAN: Ukrainian (uk).
        ROMANIAN: Romanian (ro).
        NORWEGIAN: Norwegian Bokmål (nb).
        DANISH: Danish (da).
        FINNISH: Finnish (fi).
        CZECH: Czech (cs).
        GREEK: Greek (el).
        HUNGARIAN: Hungarian (hu).
        BULGARIAN: Bulgarian (bg).
        CROATIAN: Croatian (hr).
        SERBIAN: Serbian (sr).
        SLOVAK: Slovak (sk).
        LITHUANIAN: Lithuanian (lt).
        LATVIAN: Latvian (lv).
        ESTONIAN: Estonian (et).
    """

    CATALAN = "ca"
    CHINESE_SIMPLIFIED = "zh-Hans"
    CHINESE_TRADITIONAL = "zh-Hant"
    GERMAN = "de"
    FRENCH = "fr"
    GALICIAN = "gl"
    ITALIAN = "it"
    DUTCH = "nl"
    PORTUGUESE = "pt-PT"
    SLOVENIAN = "sl"
    SWEDISH = "sv"
    SPANISH = "es"
    ENGLISH = "en"
    POLISH = "pl"
    TURKISH = "tr"
    UKRAINIAN = "uk"
    ROMANIAN = "ro"
    NORWEGIAN = "nb"
    DANISH = "da"
    FINNISH = "fi"
    CZECH = "cs"
    GREEK = "el"
    HUNGARIAN = "hu"
    BULGARIAN = "bg"
    CROATIAN = "hr"
    SERBIAN = "sr"
    SLOVAK = "sk"
    LITHUANIAN = "lt"
    LATVIAN = "lv"
    ESTONIAN = "et"

UserAccessType

Bases: Enum

User access level enumeration.

Attributes:

Name	Type	Description
REGULAR		Standard user access.
ADMIN		Administrative access.

Source code in backend/app/users/users/schema.py

class UserAccessType(Enum):
    """
    User access level enumeration.

    Attributes:
        REGULAR: Standard user access.
        ADMIN: Administrative access.
    """

    REGULAR = "regular"
    ADMIN = "admin"

Users

Bases: UsersBase

Complete users schema with administrative fields.

Note

mfa_secret is intentionally NOT part of any API-facing schema. It lives only on the SQLAlchemy model and is accessed directly by MFA verification utilities. Including it in a Pydantic schema would risk leaking the encrypted seed through responses, exports, logs, or future model_dump callers.

Attributes:

Name	Type	Description
access_type	UserAccessType	User access level.
photo_path	StrictStr | None	Path to user's photo.
active	StrictBool	Whether the user is active.
mfa_enabled	StrictBool	Whether MFA is enabled.
email_verified	StrictBool	Whether email is verified.
pending_admin_approval	StrictBool	Whether pending admin approval.

Source code in backend/app/users/users/schema.py

class Users(UsersBase):
    """
    Complete users schema with administrative fields.

    Note:
        ``mfa_secret`` is intentionally NOT part of any API-facing
        schema. It lives only on the SQLAlchemy model and is
        accessed directly by MFA verification utilities. Including
        it in a Pydantic schema would risk leaking the encrypted
        seed through responses, exports, logs, or future
        ``model_dump`` callers.

    Attributes:
        access_type: User access level.
        photo_path: Path to user's photo.
        active: Whether the user is active.
        mfa_enabled: Whether MFA is enabled.
        email_verified: Whether email is verified.
        pending_admin_approval: Whether pending admin approval.
    """

    access_type: UserAccessType = Field(
        ...,
        description="User access level",
    )
    photo_path: StrictStr | None = Field(
        default=None,
        max_length=250,
        description="Path to user's photo",
    )
    active: StrictBool = Field(
        ...,
        description="Whether the user is active",
    )
    mfa_enabled: StrictBool = Field(
        default=False,
        description="Whether MFA is enabled",
    )
    email_verified: StrictBool = Field(
        default=False,
        description="Whether email is verified",
    )
    pending_admin_approval: StrictBool = Field(
        default=False,
        description="Whether pending admin approval",
    )

    model_config = ConfigDict(
        from_attributes=True,
        extra="forbid",
        validate_assignment=True,
        use_enum_values=True,
    )

UsersBase

Bases: BaseModel

Base users schema with common fields.

Attributes:

Name	Type	Description
name	StrictStr	User's full name (1-250 chars).
username	StrictStr	Unique username (1-250 chars, alphanumeric and dots).
email	EmailStr	User's email address (max 250 chars).
city	StrictStr | None	User's city (max 250 chars).
birthdate	date | None	User's birthdate.
preferred_language	Language	Preferred language.
gender	Gender	User's gender.
units	Units	User units (metric, imperial).
height	StrictInt | None	User's height in centimeters (1-300).
max_heart_rate	StrictInt | None	Maximum heart rate in bpm (30-250).
first_day_of_week	WeekDay	First day of the week.
currency	Currency	User currency (euro, dollar, pound).

Source code in backend/app/users/users/schema.py

class UsersBase(BaseModel):
    """
    Base users schema with common fields.

    Attributes:
        name: User's full name (1-250 chars).
        username: Unique username (1-250 chars, alphanumeric
            and dots).
        email: User's email address (max 250 chars).
        city: User's city (max 250 chars).
        birthdate: User's birthdate.
        preferred_language: Preferred language.
        gender: User's gender.
        units: User units (metric, imperial).
        height: User's height in centimeters (1-300).
        max_heart_rate: Maximum heart rate in bpm (30-250).
        first_day_of_week: First day of the week.
        currency: User currency (euro, dollar, pound).
    """

    name: StrictStr = Field(
        ...,
        min_length=1,
        max_length=250,
        description="User's full name",
    )
    username: StrictStr = Field(
        ...,
        min_length=1,
        max_length=250,
        pattern=r"^[a-zA-Z0-9._-]+$",
        description="Unique username (alphanumeric, dots, hyphen, underscore)",
    )
    email: EmailStr = Field(
        ...,
        max_length=250,
        description="User's email address",
    )
    city: StrictStr | None = Field(
        default=None,
        max_length=250,
        description="User's city",
    )
    birthdate: datetime_date | None = Field(
        default=None,
        description="User's birthdate",
    )
    preferred_language: Language = Field(
        default=Language.ENGLISH,
        description="Preferred language",
    )
    gender: Gender = Field(
        default=Gender.UNSPECIFIED,
        description="User's gender",
    )
    units: server_settings_schema.Units = Field(
        default=server_settings_schema.Units.METRIC,
        description="User units (metric, imperial)",
    )
    height: StrictInt | None = Field(
        default=None,
        ge=1,
        le=300,
        description="Height in centimeters",
    )
    max_heart_rate: StrictInt | None = Field(
        default=None,
        ge=30,
        le=250,
        description="Maximum heart rate in bpm",
    )
    first_day_of_week: WeekDay = Field(
        default=WeekDay.MONDAY,
        description="First day of the week",
    )
    currency: server_settings_schema.Currency = Field(
        default=server_settings_schema.Currency.EURO,
        description="User currency (euro, dollar, pound)",
    )

    model_config = ConfigDict(use_enum_values=True)

    @field_validator("birthdate", mode="before")
    @classmethod
    def validate_birthdate(cls, value: datetime_date | str | None) -> str | None:
        """
        Convert birthdate to ISO format string.

        Args:
            value: Birthdate as date object, string, or None.

        Returns:
            ISO format date string or None.
        """
        if value is None:
            return None
        if isinstance(value, datetime_date):
            return value.isoformat()
        return value

validate_birthdate classmethod

validate_birthdate(value)

Convert birthdate to ISO format string.

Parameters:

Name	Type	Description	Default
value	date | str | None	Birthdate as date object, string, or None.	required

Returns:

Type	Description
str | None	ISO format date string or None.

Source code in backend/app/users/users/schema.py

@field_validator("birthdate", mode="before")
@classmethod
def validate_birthdate(cls, value: datetime_date | str | None) -> str | None:
    """
    Convert birthdate to ISO format string.

    Args:
        value: Birthdate as date object, string, or None.

    Returns:
        ISO format date string or None.
    """
    if value is None:
        return None
    if isinstance(value, datetime_date):
        return value.isoformat()
    return value

UsersCreate

Bases: Users

Users schema for admin user creation.

Attributes:

Name	Type	Description
password	StrictStr	User's password (min 8 chars).

Source code in backend/app/users/users/schema.py

class UsersCreate(Users):
    """
    Users schema for admin user creation.

    Attributes:
        password: User's password (min 8 chars).
    """

    password: StrictStr = Field(
        ...,
        min_length=8,
        max_length=250,
        description="User's password",
    )

UsersEditPassword

Bases: BaseModel

Schema for password update operations (self-service).

Requires the caller to prove possession of the current password — and an MFA code when MFA is enabled — before the new password is accepted. This prevents a stolen in-memory access token from being parlayed into permanent account takeover.

Attributes:

Name	Type	Description
current_password	StrictStr	Caller's existing password.
password	StrictStr	New password (min 8 chars).
mfa_code	StrictStr | None	TOTP or backup code, required when MFA is enabled on the account.

Source code in backend/app/users/users/schema.py

class UsersEditPassword(BaseModel):
    """
    Schema for password update operations (self-service).

    Requires the caller to prove possession of the current
    password — and an MFA code when MFA is enabled — before
    the new password is accepted. This prevents a stolen
    in-memory access token from being parlayed into permanent
    account takeover.

    Attributes:
        current_password: Caller's existing password.
        password: New password (min 8 chars).
        mfa_code: TOTP or backup code, required when MFA is
            enabled on the account.
    """

    current_password: StrictStr = Field(
        ...,
        min_length=1,
        max_length=250,
        description="Current password (step-up verification)",
    )
    password: StrictStr = Field(
        ...,
        min_length=8,
        max_length=250,
        description="New password",
    )
    mfa_code: StrictStr | None = Field(
        default=None,
        max_length=32,
        description="TOTP or backup code, required when MFA is enabled",
    )
    revoke_other_sessions: StrictBool = Field(
        default=False,
        description=(
            "When true, revoke all of the user's other sessions "
            "(keeping the current one) after the password change. "
            "Use this to evict an attacker when changing a password "
            "because of a suspected compromise."
        ),
    )

    model_config = ConfigDict(
        extra="forbid",
        validate_assignment=True,
    )

UsersListResponse

Bases: BaseModel

Response model for paginated user listing.

Attributes:

Name	Type	Description
total	StrictInt	Total number of user records.
num_records	StrictInt | None	Number of records in this response.
page_number	StrictInt | None	Current page number.
records	list[UsersRead]	List of user records.

Source code in backend/app/users/users/schema.py

class UsersListResponse(BaseModel):
    """
    Response model for paginated user listing.

    Attributes:
        total: Total number of user records.
        num_records: Number of records in this response.
        page_number: Current page number.
        records: List of user records.
    """

    total: StrictInt = Field(
        ...,
        ge=0,
        description="Total number of user records",
    )
    num_records: StrictInt | None = Field(
        default=None,
        ge=0,
        description="Number of records in this response",
    )
    page_number: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Current page number",
    )
    records: list[UsersRead] = Field(
        ...,
        description="List of user records",
    )

    model_config = ConfigDict(
        from_attributes=True,
        extra="forbid",
        validate_assignment=True,
    )

UsersMe

Bases: UsersRead

Extended users schema for current user profile.

Includes privacy settings and integration status.

Attributes:

Name	Type	Description
is_strava_linked	StrictInt | None	Strava integration status.
is_garminconnect_linked	StrictInt | None	Garmin Connect status.
default_activity_visibility	StrictStr | None	Default visibility level.
hide_activity_start_time	StrictBool | None	Hide start time setting.
hide_activity_location	StrictBool | None	Hide location setting.
hide_activity_map	StrictBool | None	Hide map setting.
hide_activity_hr	StrictBool | None	Hide heart rate setting.
hide_activity_power	StrictBool | None	Hide power setting.
hide_activity_cadence	StrictBool | None	Hide cadence setting.
hide_activity_elevation	StrictBool | None	Hide elevation setting.
hide_activity_speed	StrictBool | None	Hide speed setting.
hide_activity_pace	StrictBool | None	Hide pace setting.
hide_activity_laps	StrictBool | None	Hide laps setting.
hide_activity_workout_sets_steps	StrictBool | None	Hide workout sets/steps.
hide_activity_gear	StrictBool | None	Hide gear setting.
has_local_password	StrictBool | None	Whether the account has a local password (False for SSO-only accounts).

Source code in backend/app/users/users/schema.py

class UsersMe(UsersRead):
    """
    Extended users schema for current user profile.

    Includes privacy settings and integration status.

    Attributes:
        is_strava_linked: Strava integration status.
        is_garminconnect_linked: Garmin Connect status.
        default_activity_visibility: Default visibility level.
        hide_activity_start_time: Hide start time setting.
        hide_activity_location: Hide location setting.
        hide_activity_map: Hide map setting.
        hide_activity_hr: Hide heart rate setting.
        hide_activity_power: Hide power setting.
        hide_activity_cadence: Hide cadence setting.
        hide_activity_elevation: Hide elevation setting.
        hide_activity_speed: Hide speed setting.
        hide_activity_pace: Hide pace setting.
        hide_activity_laps: Hide laps setting.
        hide_activity_workout_sets_steps: Hide workout
            sets/steps.
        hide_activity_gear: Hide gear setting.
        has_local_password: Whether the account has a local
            password (False for SSO-only accounts).
    """

    is_strava_linked: StrictInt | None = Field(default=None, description="Whether Strava is linked")
    is_garminconnect_linked: StrictInt | None = Field(default=None, description="Whether Garmin Connect is linked")
    default_activity_visibility: StrictStr | None = Field(default=None, description="Default activity visibility")
    hide_activity_start_time: StrictBool | None = Field(default=None, description="Hide activity start time")
    hide_activity_location: StrictBool | None = Field(default=None, description="Hide activity location")
    hide_activity_map: StrictBool | None = Field(default=None, description="Hide activity map")
    hide_activity_hr: StrictBool | None = Field(default=None, description="Hide activity heart rate")
    hide_activity_power: StrictBool | None = Field(default=None, description="Hide activity power")
    hide_activity_cadence: StrictBool | None = Field(default=None, description="Hide activity cadence")
    hide_activity_elevation: StrictBool | None = Field(default=None, description="Hide activity elevation")
    hide_activity_speed: StrictBool | None = Field(default=None, description="Hide activity speed")
    hide_activity_pace: StrictBool | None = Field(default=None, description="Hide activity pace")
    hide_activity_laps: StrictBool | None = Field(default=None, description="Hide activity laps")
    hide_activity_workout_sets_steps: StrictBool | None = Field(
        default=None, description="Hide activity workout sets and steps"
    )
    hide_activity_gear: StrictBool | None = Field(default=None, description="Hide activity gear")
    has_local_password: StrictBool | None = Field(
        default=None,
        description=(
            "Whether the account has a local password set. False"
            " indicates an SSO-only account, in which case"
            " step-up flows must skip the password factor. The"
            " raw password hash is never exposed; only this"
            " derived boolean is returned."
        ),
    )

UsersRead

Bases: Users

Users schema for read operations.

Attributes:

Name	Type	Description
id	StrictInt	User ID.
external_auth_count	StrictInt	Number of external auth providers.

Source code in backend/app/users/users/schema.py

class UsersRead(Users):
    """
    Users schema for read operations.

    Attributes:
        id: User ID.
        external_auth_count: Number of external auth providers.
    """

    id: StrictInt = Field(
        ...,
        ge=1,
        description="User ID",
    )
    external_auth_count: StrictInt = Field(
        default=0,
        ge=0,
        description="Number of external auth providers linked",
    )

UsersSignup

Bases: UsersBase

Users schema for signup operations.

Attributes:

Name	Type	Description
password	StrictStr	User's password (min 8 chars).

Source code in backend/app/users/users/schema.py

class UsersSignup(UsersBase):
    """
    Users schema for signup operations.

    Attributes:
        password: User's password (min 8 chars).
    """

    password: StrictStr = Field(
        ...,
        min_length=8,
        max_length=250,
        description="User's password",
    )

WeekDay

Bases: Enum

Days of the week enumeration.

Attributes:

Name	Type	Description
SUNDAY		Sunday.
MONDAY		Monday.
TUESDAY		Tuesday.
WEDNESDAY		Wednesday.
THURSDAY		Thursday.
FRIDAY		Friday.
SATURDAY		Saturday.

Source code in backend/app/users/users/schema.py

class WeekDay(Enum):
    """
    Days of the week enumeration.

    Attributes:
        SUNDAY: Sunday.
        MONDAY: Monday.
        TUESDAY: Tuesday.
        WEDNESDAY: Wednesday.
        THURSDAY: Thursday.
        FRIDAY: Friday.
        SATURDAY: Saturday.
    """

    SUNDAY = "sunday"
    MONDAY = "monday"
    TUESDAY = "tuesday"
    WEDNESDAY = "wednesday"
    THURSDAY = "thursday"
    FRIDAY = "friday"
    SATURDAY = "saturday"

approve_user

approve_user(user_id, db)

Approve a user by marking them as active.

Parameters:

Name	Type	Description	Default
user_id	int	ID of user to approve.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
None	None

Raises:

Type	Description
HTTPException	404 if user not found.
HTTPException	400 if user email not verified.
HTTPException	500 if database error occurs.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def approve_user(user_id: int, db: Session) -> None:
    """
    Approve a user by marking them as active.

    Args:
        user_id: ID of user to approve.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 400 if user email not verified.
        HTTPException: 500 if database error occurs.
    """
    # Get the user from the database
    db_users = _get_user_model_by_id_or_404(user_id, db)

    if not db_users.email_verified:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="User email is not verified",
        )

    db_users.pending_admin_approval = False
    db_users.active = True

    # Commit the transaction
    db.commit()
    db.refresh(db_users)

check_user_is_active

check_user_is_active(user)

Check if user is active and raise 403 if inactive.

Parameters:

Name	Type	Description	Default
user	UsersRead	User object to check (UsersRead schema).	required

Returns:

Type	Description
None	None

Raises:

Type	Description
HTTPException	403 if user is not active.

Source code in backend/app/users/users/utils.py

def check_user_is_active(
    user: users_schema.UsersRead,
) -> None:
    """
    Check if user is active and raise 403 if inactive.

    Args:
        user: User object to check (UsersRead schema).

    Returns:
        None

    Raises:
        HTTPException: 403 if user is not active.
    """
    if not user.active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Inactive user",
            headers={"WWW-Authenticate": "Bearer"},
        )

create_signup_user

create_signup_user(user, server_settings, identity_service, db, persist_credential=True)

Create a new user during signup process.

Parameters:

Name	Type	Description	Default
user	UsersSignup	User signup data.	required
server_settings	ServerSettingsRead	Server config for signup requirements.	required
identity_service	IdentityService	Identity service dependency.	required
db	Session	SQLAlchemy database session.	required
persist_credential	bool	When True (default), validate the supplied password and store its hash in the auth-owned credential table. SSO-created accounts pass False so they get no local credential row and remain SSO-only (has_local_password then correctly reports False).	True

Returns:

Type	Description
UsersRead	Created user schema.

Raises:

Type	Description
HTTPException	409 if email/username already exists. Abstract message to reduce information leakage.
HTTPException	500 if database error occurs.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def create_signup_user(
    user: users_schema.UsersSignup,
    server_settings: server_settings_schema.ServerSettingsRead,
    identity_service: "auth_identity_service.IdentityService",
    db: Session,
    persist_credential: bool = True,
) -> users_schema.UsersRead:
    """
    Create a new user during signup process.

    Args:
        user: User signup data.
        server_settings: Server config for signup requirements.
        identity_service: Identity service dependency.
        db: SQLAlchemy database session.
        persist_credential: When ``True`` (default), validate the supplied
            password and store its hash in the auth-owned credential table.
            SSO-created accounts pass ``False`` so they get no local
            credential row and remain SSO-only (``has_local_password`` then
            correctly reports ``False``).

    Returns:
        Created user schema.

    Raises:
        HTTPException: 409 if email/username already exists. Abstract message
            to reduce information leakage.
        HTTPException: 500 if database error occurs.
    """
    try:
        # Determine user status based on server settings
        active = True
        email_verified = False
        pending_admin_approval = False

        if server_settings.signup_require_email_verification:
            email_verified = False
            active = False  # Inactive until email verified

        if server_settings.signup_require_admin_approval:
            pending_admin_approval = True
            active = False  # Inactive until approved

        # If both email verification and admin approval are disabled, user is immediately active
        if not server_settings.signup_require_email_verification and not server_settings.signup_require_admin_approval:
            active = True
            email_verified = True

        # Create a new user
        db_users = users_models.Users(
            **user.model_dump(
                exclude={
                    "username",
                    "email",
                    "access_type",
                    "active",
                    "email_verified",
                    "pending_admin_approval",
                    "password",
                }
            ),
            username=user.username.lower(),
            email=user.email.lower(),
            access_type=users_schema.UserAccessType.REGULAR.value,
            active=active,
            email_verified=email_verified,
            pending_admin_approval=pending_admin_approval,
        )

        # Hash the signup password with the configured policy. SSO-created
        # accounts opt out (persist_credential=False) so the supplied
        # placeholder password is never validated or hashed.
        hashed_password: str | None = None
        if persist_credential:
            hashed_password = auth_password_policy.validate_and_hash_for_user(
                identity_service,
                server_settings,
                users_schema.UserAccessType.REGULAR.value,
                user.password,
            )

        # Add the user to the database
        db.add(db_users)
        db.commit()
        db.refresh(db_users)

        # Persist the password hash in the auth-owned credential table. Skipped
        # for SSO-only accounts so that ``has_local_password`` stays a true row
        # existence check.
        if hashed_password is not None:
            identity_service.set_local_password_hash(db_users.id, hashed_password)

        # Return user
        return _transform_users(db_users)
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Conflict status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=("Unable to create user."),
        ) from integrity_error

create_user

create_user(user, identity_service, db)

Create a new user with hashed password.

Parameters:

Name	Type	Description	Default
user	UsersCreate	User creation data with plain text password.	required
identity_service	IdentityService	Identity service dependency.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersRead	Created user schema with hashed password.

Raises:

Type	Description
HTTPException	409 if email/username already exists.
HTTPException	500 if database error occurs.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def create_user(
    user: users_schema.UsersCreate,
    identity_service: "auth_identity_service.IdentityService",
    db: Session,
) -> users_schema.UsersRead:
    """
    Create a new user with hashed password.

    Args:
        user: User creation data with plain text password.
        identity_service: Identity service dependency.
        db: SQLAlchemy database session.

    Returns:
        Created user schema with hashed password.

    Raises:
        HTTPException: 409 if email/username already exists.
        HTTPException: 500 if database error occurs.
    """
    try:
        user.username = user.username.lower()
        user.email = user.email.lower()

        # Get server settings to determine password policy
        server_settings = server_settings_utils.get_server_settings_or_404(db)

        # Normalize access_type to string value
        access_type_value = users_schema.normalize_access_type(user.access_type)

        # Hash the password with configurable policy and length
        hashed_password = auth_password_policy.validate_and_hash_for_user(
            identity_service,
            server_settings,
            access_type_value,
            user.password,
        )

        # Create a new user
        db_users = users_models.Users(
            **user.model_dump(exclude={"password", "access_type", "mfa_enabled"}),
            access_type=access_type_value,
        )

        # Add the user to the database
        db.add(db_users)
        db.commit()
        db.refresh(db_users)

        # Persist the password hash in the auth-owned credential table.
        identity_service.set_local_password_hash(db_users.id, hashed_password)

        # Return user
        return _transform_users(db_users)
    except HTTPException:
        # Rollback the transaction
        db.rollback()
        raise
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Conflict status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=("Duplicate entry error. Check if email and username are unique"),
        ) from integrity_error

create_user_default_data

create_user_default_data(user_id, identity_service, db)

Create default data for newly created user.

Parameters:

Name	Type	Description	Default
user_id	int	ID of user to create default data for.	required
identity_service	IdentityService	Identity service used to initialise the auth-owned MFA row through the auth boundary.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
None	None

Source code in backend/app/users/users/utils.py

def create_user_default_data(
    user_id: int,
    identity_service: "auth_identity_service.IdentityService",
    db: Session,
) -> None:
    """
    Create default data for newly created user.

    Args:
        user_id: ID of user to create default data for.
        identity_service: Identity service used to initialise the
            auth-owned MFA row through the auth boundary.
        db: SQLAlchemy database session.

    Returns:
        None
    """
    # Create the user integrations in the database
    user_integrations_crud.create_user_integrations(user_id, db)

    # Create the user privacy settings
    users_privacy_settings_crud.create_user_privacy_settings(user_id, db)

    # Create the user health targets
    health_targets_crud.create_health_targets(user_id, db)

    # Create the user default gear
    user_default_gear_crud.create_user_default_gear(user_id, db)

    # Create the user's MFA row (disabled by default) via the auth boundary.
    identity_service.initialize_user_mfa(user_id)

delete_user async

delete_user(user_id, db)

Delete a user from the database.

Parameters:

Name	Type	Description	Default
user_id	int	ID of user to delete.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
None	None

Raises:

Type	Description
HTTPException	404 if user not found.
HTTPException	500 if database error occurs.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
async def delete_user(user_id: int, db: Session) -> None:
    """
    Delete a user from the database.

    Args:
        user_id: ID of user to delete.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 500 if database error occurs.
    """
    # Delete the user row off the event loop (blocking sync DB calls),
    # keeping the API event loop responsive.
    await run_in_threadpool(_delete_user_row, user_id, db)

    # Delete the user photo in the filesystem
    await users_utils.delete_user_photo_filesystem(user_id)

delete_user_photo_filesystem async

delete_user_photo_filesystem(user_id)

Delete user photo files from filesystem.

Parameters:

Name	Type	Description	Default
user_id	int	ID of user whose photo files to delete.	required

Returns:

Type	Description
None	None

Source code in backend/app/users/users/utils.py

async def delete_user_photo_filesystem(user_id: int) -> None:
    """
    Delete user photo files from filesystem.

    Args:
        user_id: ID of user whose photo files to delete.

    Returns:
        None
    """
    await core_file_uploads.delete_files_by_pattern(core_config.USER_IMAGES_DIR, f"{user_id}.*")

edit_user async

edit_user(user_id, user, db)

Update an existing user's information.

Note

This dynamic-assignment helper is intended for admin endpoints that legitimately need to set fields like access_type, active, or pending_admin_approval. Self-service profile updates MUST NOT call this — use :func:edit_profile_user instead, which enforces an explicit allow-list and prevents privilege escalation through mass assignment.

Parameters:

Name	Type	Description	Default
user_id	int	ID of user to update.	required
user	UsersRead	User data to update with.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersRead	users_schema.UsersRead

Raises:

Type	Description
HTTPException	404 if user not found.
HTTPException	409 if email/username conflict.
HTTPException	500 if database error occurs.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
async def edit_user(user_id: int, user: users_schema.UsersRead, db: Session) -> users_schema.UsersRead:
    """
    Update an existing user's information.

    Note:
        This dynamic-assignment helper is intended for **admin**
        endpoints that legitimately need to set fields like
        ``access_type``, ``active``, or ``pending_admin_approval``.
        Self-service profile updates MUST NOT call this — use
        :func:`edit_profile_user` instead, which enforces an
        explicit allow-list and prevents privilege escalation
        through mass assignment.

    Args:
        user_id: ID of user to update.
        user: User data to update with.
        db: SQLAlchemy database session.

    Returns:
        users_schema.UsersRead

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 409 if email/username conflict.
        HTTPException: 500 if database error occurs.
    """
    try:
        # Fetch the user off the event loop (blocking DB read).
        db_users = await run_in_threadpool(_get_user_model_by_id_or_404, user_id, db)

        height_before = db_users.height
        max_heart_rate_before = db_users.max_heart_rate
        birthdate_before = db_users.birthdate

        # Check if the photo_path is being updated
        if user.photo_path:
            # Delete the user photo in the filesystem
            await users_utils.delete_user_photo_filesystem(db_users.id)

        user.username = user.username.lower()

        # Dictionary of the fields to update if they are not None
        user_data = user.model_dump(exclude_unset=True, exclude={"password", "external_auth_count", "mfa_enabled"})
        # Iterate over the fields and update the db_users dynamically
        for key, value in user_data.items():
            # Skip attributes that are read-only properties on the ORM model
            # with no setter. Response-only fields (e.g. mfa_enabled,
            # external_auth_count) are already excluded above via
            # model_dump, this is a defensive guard for any future
            # computed property added to Users.
            class_attr = getattr(type(db_users), key, None)
            if isinstance(class_attr, property) and class_attr.fset is None:
                continue
            setattr(db_users, key, value)

        # Persist the changes off the event loop. The commit/refresh and the
        # bulk BMI recalculation are blocking SQLAlchemy calls, so run them
        # in a worker thread to keep the API event loop responsive.
        updated_user = await run_in_threadpool(
            _persist_user_edits,
            db_users,
            height_before,
            max_heart_rate_before,
            birthdate_before,
            True,
            db,
        )

        if db_users.photo_path is None:
            # Delete the user photo in the filesystem
            await users_utils.delete_user_photo_filesystem(db_users.id)

        return updated_user
    except HTTPException:
        raise
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Conflict status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=("Duplicate entry error. Check if email and username are unique"),
        ) from integrity_error

get_admin_users_or_404

get_admin_users_or_404(db)

Retrieve all admin users from database or raise 404 error.

Parameters:

Name	Type	Description	Default
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
list[UsersRead]	List of all admin User schemas.

Raises:

Type	Description
HTTPException	404 if no admin users found.

Source code in backend/app/users/users/utils.py

def get_admin_users_or_404(db: Session) -> list[users_schema.UsersRead]:
    """
    Retrieve all admin users from database or raise 404 error.

    Args:
        db: SQLAlchemy database session.

    Returns:
        List of all admin User schemas.

    Raises:
        HTTPException: 404 if no admin users found.
    """
    admins: list[users_schema.UsersRead] = users_crud.get_users_admin(db)

    if not admins:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No admin users found",
        )

    return admins

get_all_users

get_all_users(db)

Retrieve all users from the database.

Parameters:

Name	Type	Description	Default
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
list[UsersRead]	List of all user schemas.

Raises:

Type	Description
HTTPException	500 error if database query fails.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def get_all_users(db: Session) -> list[users_schema.UsersRead]:
    """
    Retrieve all users from the database.

    Args:
        db: SQLAlchemy database session.

    Returns:
        List of all user schemas.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(users_models.Users)
    users: list[users_models.Users] = list(db.execute(stmt).scalars().all())

    return _transform_users(users)

get_user_by_email

get_user_by_email(email, db)

Retrieve user by email address.

Parameters:

Name	Type	Description	Default
email	str	Email address to search for (case-insensitive).	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersRead | None	User schema if found, None otherwise.

Raises:

Type	Description
HTTPException	500 error if database query fails.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def get_user_by_email(email: str, db: Session) -> users_schema.UsersRead | None:
    """
    Retrieve user by email address.

    Args:
        email: Email address to search for (case-insensitive).
        db: SQLAlchemy database session.

    Returns:
        User schema if found, None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(users_models.Users).where(users_models.Users.email == email.lower())
    user = db.execute(stmt).scalar_one_or_none()
    return _transform_users(user) if user else None

get_user_by_id

get_user_by_id(user_id, db, public_check=False)

Retrieve user by ID.

Parameters:

Name	Type	Description	Default
user_id	int	User ID to search for.	required
db	Session	SQLAlchemy database session.	required
public_check	bool	If True, only returns user when public sharing is enabled in server settings.	False

Returns:

Type	Description
UsersRead | None	User schema if found (and public sharing enabled if
UsersRead | None	public_check=True), None otherwise.

Raises:

Type	Description
HTTPException	500 error if database query fails.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def get_user_by_id(user_id: int, db: Session, public_check: bool = False) -> users_schema.UsersRead | None:
    """
    Retrieve user by ID.

    Args:
        user_id: User ID to search for.
        db: SQLAlchemy database session.
        public_check: If True, only returns user when public sharing
                      is enabled in server settings.

    Returns:
        User schema if found (and public sharing enabled if
        public_check=True), None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    if public_check:
        # Check if public sharable links are enabled in server settings
        server_settings = server_settings_utils.get_server_settings_or_404(db)

        # Return None if public sharable links are disabled
        if not server_settings.public_shareable_links or not server_settings.public_shareable_links_user_info:
            return None

    stmt = select(users_models.Users).where(users_models.Users.id == user_id)
    user = db.execute(stmt).scalar_one_or_none()
    return _transform_users(user) if user else None

get_user_by_id_or_404

get_user_by_id_or_404(user_id, db)

Retrieve user by ID or raise 404 error.

Parameters:

Name	Type	Description	Default
user_id	int	User ID to search for.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersRead	Users schema (guaranteed non-None).

Raises:

Type	Description
HTTPException	404 if user not found.

Source code in backend/app/users/users/utils.py

def get_user_by_id_or_404(user_id: int, db: Session) -> users_schema.UsersRead:
    """
    Retrieve user by ID or raise 404 error.

    Args:
        user_id: User ID to search for.
        db: SQLAlchemy database session.

    Returns:
        Users schema (guaranteed non-None).

    Raises:
        HTTPException: 404 if user not found.
    """
    # Get the user from the database
    db_user: users_schema.UsersRead | None = users_crud.get_user_by_id(user_id, db)

    if db_user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return db_user

get_user_by_username

get_user_by_username(username: str, db: Session, contains: Literal[False] = False) -> users_schema.UsersRead | None

get_user_by_username(username: str, db: Session, contains: Literal[True]) -> list[users_schema.UsersRead]

get_user_by_username(username, db, contains=False)

Retrieve user by username.

Parameters:

Name	Type	Description	Default
username	str	Username to search for.	required
db	Session	SQLAlchemy database session.	required
contains	bool	If True, performs partial match search and returns list of matching users. If False, performs exact match and returns single user or None.	False

Returns:

Type	Description
list[UsersRead] | UsersRead | None	If contains=False: User schema if found, None otherwise.
list[UsersRead] | UsersRead | None	If contains=True: List of user schemas matching the search.

Raises:

Type	Description
HTTPException	500 error if database query fails.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def get_user_by_username(
    username: str, db: Session, contains: bool = False
) -> list[users_schema.UsersRead] | users_schema.UsersRead | None:
    """
    Retrieve user by username.

    Args:
        username: Username to search for.
        db: SQLAlchemy database session.
        contains: If True, performs partial match search and returns
                  list of matching users. If False, performs exact
                  match and returns single user or None.

    Returns:
        If contains=False: User schema if found, None otherwise.
        If contains=True: List of user schemas matching the search.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    # Decode and normalize search term (needed for both exact and partial matches)
    normalized_username = unquote(username).replace("+", " ").lower()

    if contains:
        # Escape LIKE special characters to prevent SQL injection
        escaped_username = normalized_username.replace("\\", "\\\\").replace("%", r"\%").replace("_", r"\_")

        # Query users with username containing the search term
        stmt = select(users_models.Users).where(
            func.lower(users_models.Users.username).like(f"%{escaped_username}%", escape="\\")
        )
        users: list[users_models.Users] = list(db.execute(stmt).scalars().all())
        return _transform_users(users)
    else:
        # Exact match - no LIKE escaping needed
        stmt = select(users_models.Users).where(users_models.Users.username == normalized_username)
        user = db.execute(stmt).scalar_one_or_none()
        return _transform_users(user) if user else None

get_users_admin

get_users_admin(db)

Retrieve all admin users from the database.

Parameters:

Name	Type	Description	Default
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
list[UsersRead]	List of admin user schemas.

Raises:

Type	Description
HTTPException	500 error if database query fails.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def get_users_admin(db: Session) -> list[users_schema.UsersRead]:
    """
    Retrieve all admin users from the database.

    Args:
        db: SQLAlchemy database session.

    Returns:
        List of admin user schemas.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(users_models.Users).where(users_models.Users.access_type == users_schema.UserAccessType.ADMIN.value)
    users: list[users_models.Users] = list(db.execute(stmt).scalars().all())
    return _transform_users(users)

get_users_number

get_users_number(db, show_inactive=True, show_email_unverified=True, show_pending_approval=True)

Count users matching the optional list filters.

Parameters:

Name	Type	Description	Default
db	Session	SQLAlchemy database session.	required
show_inactive	bool | None	If False, excludes inactive users. Defaults to True.	True
show_email_unverified	bool | None	If False, excludes users with unverified emails. Defaults to True.	True
show_pending_approval	bool | None	If False, excludes users pending admin approval. Defaults to True.	True

Returns:

Type	Description
int	Number of users matching the filters.

Raises:

Type	Description
HTTPException	500 error if database query fails.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def get_users_number(
    db: Session,
    show_inactive: bool | None = True,
    show_email_unverified: bool | None = True,
    show_pending_approval: bool | None = True,
) -> int:
    """
    Count users matching the optional list filters.

    Args:
        db: SQLAlchemy database session.
        show_inactive: If False, excludes inactive users.
            Defaults to True.
        show_email_unverified: If False, excludes users with
            unverified emails. Defaults to True.
        show_pending_approval: If False, excludes users pending
            admin approval. Defaults to True.

    Returns:
        Number of users matching the filters.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(func.count(users_models.Users.id))

    if show_inactive is False:
        stmt = stmt.where(users_models.Users.active.is_(True))
    if show_email_unverified is False:
        stmt = stmt.where(users_models.Users.email_verified.is_(True))
    if show_pending_approval is False:
        stmt = stmt.where(users_models.Users.pending_admin_approval.is_(False))

    return db.execute(stmt).scalar_one()

get_users_with_pagination

get_users_with_pagination(db, page_number=None, num_records=None, show_inactive=True, show_email_unverified=True, show_pending_approval=True)

Retrieve a paginated list of users with optional filtering.

Parameters:

Name	Type	Description	Default
db	Session	Database session for executing queries.	required
page_number	int | None	The page number for pagination (1-indexed). If None, pagination is not applied. Defaults to None.	None
num_records	int | None	The number of records per page. If None, pagination is not applied. Defaults to None.	None
show_inactive	bool | None	If False, excludes inactive users. Defaults to True (includes inactive users).	True
show_email_unverified	bool | None	If False, excludes users with unverified emails. Defaults to True (includes email unverified users).	True
show_pending_approval	bool | None	If False, excludes users pending admin approval. Defaults to True (includes pending approval users).	True

Returns:

Type	Description
list[UsersRead]	list[users_schema.UsersRead]: A list of User schemas matching the specified criteria, ordered by username. Returns an empty list if no users match the filters.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def get_users_with_pagination(
    db: Session,
    page_number: int | None = None,
    num_records: int | None = None,
    show_inactive: bool | None = True,
    show_email_unverified: bool | None = True,
    show_pending_approval: bool | None = True,
) -> list[users_schema.UsersRead]:
    """
    Retrieve a paginated list of users with optional filtering.

    Args:
        db (Session): Database session for executing queries.
        page_number (int | None): The page number for pagination (1-indexed).
            If None, pagination is not applied. Defaults to None.
        num_records (int | None): The number of records per page.
            If None, pagination is not applied. Defaults to None.
        show_inactive (bool | None): If False, excludes inactive users.
            Defaults to True (includes inactive users).
        show_email_unverified (bool | None): If False, excludes users with
            unverified emails. Defaults to True (includes email unverified
            users).
        show_pending_approval (bool | None): If False, excludes users pending
            admin approval. Defaults to True (includes pending approval users).

    Returns:
        list[users_schema.UsersRead]: A list of User schemas matching the specified
            criteria, ordered by username. Returns an empty list if no users
            match the filters.
    """
    stmt = select(users_models.Users)

    if show_inactive is False:
        stmt = stmt.where(users_models.Users.active.is_(True))
    if show_email_unverified is False:
        stmt = stmt.where(users_models.Users.email_verified.is_(True))
    if show_pending_approval is False:
        stmt = stmt.where(users_models.Users.pending_admin_approval.is_(False))

    stmt = stmt.order_by(users_models.Users.username)

    if page_number is not None and num_records is not None:
        stmt = stmt.offset((page_number - 1) * num_records).limit(num_records)

    users: list[users_models.Users] = list(db.execute(stmt).scalars().all())
    return _transform_users(users)

save_user_image_file async

save_user_image_file(user_id, file, db)

Save user image file with security validation and update DB.

Uses centralized file upload handler for validation and async I/O, then updates user photo path in database.

Parameters:

Name	Type	Description	Default
user_id	int	ID of user whose image is being saved.	required
file	UploadFile	Uploaded image file (UploadFile).	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
str	Path to saved image file.

Raises:

Type	Description
HTTPException	400 if filename or extension is invalid, 413 if too large, 500 if upload fails.

Source code in backend/app/users/users/utils.py

async def save_user_image_file(user_id: int, file: UploadFile, db: Session) -> str:
    """
    Save user image file with security validation and update DB.

    Uses centralized file upload handler for validation and async
    I/O, then updates user photo path in database.

    Args:
        user_id: ID of user whose image is being saved.
        file: Uploaded image file (UploadFile).
        db: SQLAlchemy database session.

    Returns:
        Path to saved image file.

    Raises:
        HTTPException: 400 if filename or extension is invalid,
            413 if too large, 500 if upload fails.
    """
    # Validate the user exists off the event loop (blocking sync DB read).
    # A missing user raises HTTPException 404 inside the thread, which
    # propagates through the await unchanged.
    await run_in_threadpool(get_user_by_id_or_404, user_id, db)

    if not file.filename:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Filename is required",
        )

    # Defense-in-depth allow-list on the user-supplied extension.
    # SafeUploads still validates the magic number afterwards, so a
    # mismatched signature is rejected even if the extension passes.
    _, file_extension = os.path.splitext(file.filename)
    file_extension = file_extension.lower()
    if file_extension not in _ALLOWED_USER_IMAGE_EXTENSIONS:
        raise HTTPException(
            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
            detail="Unsupported user image file type",
        )

    filename: str = f"{user_id}{file_extension}"

    # Save file using centralized file upload handler
    await core_file_uploads.save_validated_upload(
        file,
        kind=core_file_uploads.UploadKind.IMAGE,
        upload_dir=core_config.USER_IMAGES_DIR,
        filename=filename,
    )

    # Update user photo path in database
    return str(await users_crud.update_user_photo(user_id, db, os.path.join(core_config.USER_IMAGES_DIR, filename)))

update_user_photo async

update_user_photo(user_id, db, photo_path=None)

Update a user's photo path.

Parameters:

Name	Type	Description	Default
user_id	int	ID of user to update photo for.	required
db	Session	SQLAlchemy database session.	required
photo_path	str | None	New photo path. If None, removes photo.	None

Returns:

Type	Description
str | None	The updated photo path, or None if removed.

Raises:

Type	Description
HTTPException	404 if user not found.
HTTPException	500 if database error occurs.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
async def update_user_photo(user_id: int, db: Session, photo_path: str | None = None) -> str | None:
    """
    Update a user's photo path.

    Args:
        user_id: ID of user to update photo for.
        db: SQLAlchemy database session.
        photo_path: New photo path. If None, removes photo.

    Returns:
        The updated photo path, or None if removed.

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 500 if database error occurs.
    """
    # Persist the photo path change off the event loop (blocking sync
    # DB calls), keeping the API event loop responsive.
    await run_in_threadpool(_apply_user_photo_update, user_id, photo_path, db)

    if photo_path:
        # Return the photo path
        return photo_path

    # Delete the user photo in the filesystem
    await users_utils.delete_user_photo_filesystem(user_id)

    return None

verify_user_email

verify_user_email(user_id, server_settings, db)

Verify user email and conditionally activate account.

Parameters:

Name	Type	Description	Default
user_id	int	ID of user to verify.	required
server_settings	ServerSettingsRead	Server config determining activation policy.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
None	None

Raises:

Type	Description
HTTPException	404 if user not found.
HTTPException	500 if database error occurs.

Source code in backend/app/users/users/crud.py

@core_decorators.handle_db_errors
def verify_user_email(
    user_id: int,
    server_settings: server_settings_schema.ServerSettingsRead,
    db: Session,
) -> None:
    """
    Verify user email and conditionally activate account.

    Args:
        user_id: ID of user to verify.
        server_settings: Server config determining activation policy.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 500 if database error occurs.
    """
    # Get the user from the database
    db_users = _get_user_model_by_id_or_404(user_id, db)

    db_users.email_verified = True
    if not server_settings.signup_require_admin_approval:
        db_users.pending_admin_approval = False
        db_users.active = True

    # Commit the transaction
    db.commit()
    db.refresh(db_users)