API Reference

User identity provider module for SSO/OAuth account linking.

This module provides CRUD operations and data models for managing user-to-identity-provider associations, including token storage and last login tracking.

Exports

• CRUD: check_user_identity_providers_by_idp_id, get_user_identity_providers_by_user_id, get_user_identity_provider_by_user_id_and_idp_id, get_user_identity_provider_by_subject_and_idp_id, create_user_identity_provider, update_user_identity_provider_last_login, store_user_identity_provider_tokens, clear_user_identity_provider_refresh_token_by_user_id_and_idp_id, delete_user_identity_provider
• Schemas: UsersIdentityProviderBase, UsersIdentityProviderCreate, UsersIdentityProviderRead, UsersIdentityProviderResponse, UsersIdentityProviderTokenUpdate
• Models: UsersIdentityProvider (ORM model)
• Utils: get_user_identity_provider_refresh_token_by_user_id_and_idp_id

UserIdentityProviderModel

Bases: Base

User-to-identity-provider association for SSO/OAuth linking.

Attributes:

Name	Type	Description
id	Mapped[int]	Primary key.
user_id	Mapped[int]	Foreign key to users table.
idp_id	Mapped[int]	Foreign key to identity_providers table.
idp_subject	Mapped[str]	Subject/ID from the identity provider.
linked_at	Mapped[datetime]	Timestamp when the IdP was linked.
last_login	Mapped[datetime | None]	Last login timestamp using this IdP.
idp_refresh_token	Mapped[str | None]	Encrypted refresh token.
idp_access_token_expires_at	Mapped[datetime | None]	Access token expiry time.
idp_refresh_token_updated_at	Mapped[datetime | None]	Refresh token update time.
user	Mapped[datetime | None]	Relationship to Users model.
identity_providers		Relationship to IdentityProvider model.

Source code in backend/app/users/users_identity_providers/models.py

class UsersIdentityProvider(Base):
    """
    User-to-identity-provider association for SSO/OAuth linking.

    Attributes:
        id: Primary key.
        user_id: Foreign key to users table.
        idp_id: Foreign key to identity_providers table.
        idp_subject: Subject/ID from the identity provider.
        linked_at: Timestamp when the IdP was linked.
        last_login: Last login timestamp using this IdP.
        idp_refresh_token: Encrypted refresh token.
        idp_access_token_expires_at: Access token expiry time.
        idp_refresh_token_updated_at: Refresh token update time.
        user: Relationship to Users model.
        identity_providers: Relationship to IdentityProvider model.
    """

    __tablename__ = "users_identity_providers"

    id: Mapped[int] = mapped_column(
        primary_key=True,
        autoincrement=True,
        index=True,
    )
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
        comment="User ID",
    )
    idp_id: Mapped[int] = mapped_column(
        ForeignKey("identity_providers.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
        comment="Identity Provider ID",
    )
    idp_subject: Mapped[str] = mapped_column(
        String(length=500),
        nullable=False,
        comment="Subject/ID from the identity provider",
    )
    linked_at: Mapped[datetime] = mapped_column(
        nullable=False,
        server_default=func.now(),
        comment="When this IdP was linked to the user",
    )
    last_login: Mapped[datetime | None] = mapped_column(
        nullable=True,
        comment="Last login using this IdP",
    )
    idp_refresh_token: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
        comment="Encrypted refresh token",
    )
    idp_access_token_expires_at: Mapped[datetime | None] = mapped_column(
        nullable=True,
        comment="Access token expiry time",
    )
    idp_refresh_token_updated_at: Mapped[datetime | None] = mapped_column(
        nullable=True,
        comment="Last refresh token update",
    )

    # Relationships
    # TODO: Change to Mapped["User"] when all modules use mapped
    users = relationship("Users", back_populates="user_identity_providers")
    # TODO: Change to Mapped["IdentityProvider"] when all modules use mapped
    identity_providers = relationship(
        "IdentityProvider",
        back_populates="user_identity_providers",
    )

UsersIdentityProviderBase

Bases: BaseModel

Base schema for users identity provider association.

Attributes:

Name	Type	Description
user_id	StrictInt	User ID reference.
idp_id	StrictInt	Identity Provider ID reference.
idp_subject	StrictStr	Subject/ID from the identity provider.

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

class UsersIdentityProviderBase(BaseModel):
    """
    Base schema for users identity provider association.

    Attributes:
        user_id: User ID reference.
        idp_id: Identity Provider ID reference.
        idp_subject: Subject/ID from the identity provider.
    """

    user_id: StrictInt = Field(..., ge=1, description="User ID")
    idp_id: StrictInt = Field(..., ge=1, description="Identity Provider ID")
    idp_subject: StrictStr = Field(
        ...,
        max_length=500,
        min_length=1,
        description="Subject/ID from the identity provider",
    )

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

UsersIdentityProviderCreate

Bases: UsersIdentityProviderBase

Schema for creating user identity provider links.

Inherits all attributes from UsersIdentityProviderBase. Links are created during SSO authentication flow.

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

class UsersIdentityProviderCreate(UsersIdentityProviderBase):
    """
    Schema for creating user identity provider links.

    Inherits all attributes from UsersIdentityProviderBase.
    Links are created during SSO authentication flow.
    """

UsersIdentityProviderRead

Bases: UsersIdentityProviderBase

Schema for reading user identity provider links.

Extends base with identifier and timestamp fields. Refresh token is excluded for security.

Attributes:

Name	Type	Description
id	StrictInt	Unique identifier for the link.
linked_at	datetime	When this IdP was linked.
last_login	datetime | None	Last login using this IdP.
idp_access_token_expires_at	datetime | None	Access token expiry.
idp_refresh_token_updated_at	datetime | None	Last refresh token update.

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

class UsersIdentityProviderRead(UsersIdentityProviderBase):
    """
    Schema for reading user identity provider links.

    Extends base with identifier and timestamp fields.
    Refresh token is excluded for security.

    Attributes:
        id: Unique identifier for the link.
        linked_at: When this IdP was linked.
        last_login: Last login using this IdP.
        idp_access_token_expires_at: Access token expiry.
        idp_refresh_token_updated_at: Last refresh token update.
    """

    id: StrictInt = Field(..., description="Link ID")
    linked_at: datetime = Field(..., description="When linked")
    last_login: datetime | None = Field(
        default=None,
        description="Last login using this IdP",
    )
    idp_access_token_expires_at: datetime | None = Field(
        default=None,
        description="Access token expiry",
    )
    idp_refresh_token_updated_at: datetime | None = Field(
        default=None,
        description="Last refresh token update",
    )

UsersIdentityProviderResponse

Bases: UsersIdentityProviderRead

Response schema with enriched identity provider details.

Extends Read schema with display fields added by API layer for frontend convenience.

Attributes:

Name	Type	Description
idp_name	StrictStr | None	Identity provider name (enriched).
idp_slug	StrictStr | None	Identity provider slug (enriched).
idp_icon	StrictStr | None	Identity provider icon (enriched).
idp_provider_type	StrictStr | None	Provider type (enriched).

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

class UsersIdentityProviderResponse(UsersIdentityProviderRead):
    """
    Response schema with enriched identity provider details.

    Extends Read schema with display fields added by API layer
    for frontend convenience.

    Attributes:
        idp_name: Identity provider name (enriched).
        idp_slug: Identity provider slug (enriched).
        idp_icon: Identity provider icon (enriched).
        idp_provider_type: Provider type (enriched).
    """

    idp_name: StrictStr | None = Field(
        default=None,
        description="Identity provider name",
    )
    idp_slug: StrictStr | None = Field(
        default=None,
        description="Identity provider slug",
    )
    idp_icon: StrictStr | None = Field(
        default=None,
        description="Identity provider icon",
    )
    idp_provider_type: StrictStr | None = Field(
        default=None,
        description="Provider type",
    )

UsersIdentityProviderTokenUpdate

Bases: BaseModel

Internal schema for updating IdP token data.

Used only by service layer, never exposed via API. Refresh token must be encrypted before storage.

Attributes:

Name	Type	Description
idp_refresh_token	StrictStr | None	Encrypted refresh token.
idp_access_token_expires_at	datetime | None	Access token expiry.
idp_refresh_token_updated_at	datetime | None	Last refresh token update.

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

class UsersIdentityProviderTokenUpdate(BaseModel):
    """
    Internal schema for updating IdP token data.

    Used only by service layer, never exposed via API.
    Refresh token must be encrypted before storage.

    Attributes:
        idp_refresh_token: Encrypted refresh token.
        idp_access_token_expires_at: Access token expiry.
        idp_refresh_token_updated_at: Last refresh token update.
    """

    idp_refresh_token: StrictStr | None = Field(
        default=None,
        description="Encrypted refresh token",
    )
    idp_access_token_expires_at: datetime | None = Field(
        default=None,
        description="Access token expiry",
    )
    idp_refresh_token_updated_at: datetime | None = Field(
        default=None,
        description="Last refresh token update",
    )

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

check_user_identity_providers_by_idp_id

check_user_identity_providers_by_idp_id(idp_id, db)

Check if any user links exist for an identity provider.

Parameters:

Name	Type	Description	Default
idp_id	int	The ID of the identity provider.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
bool	True if at least one user is linked, False otherwise.

Raises:

Type	Description
HTTPException	500 error if database query fails.

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

@core_decorators.handle_db_errors
def check_user_identity_providers_by_idp_id(
    idp_id: int,
    db: Session,
) -> bool:
    """
    Check if any user links exist for an identity provider.

    Args:
        idp_id: The ID of the identity provider.
        db: SQLAlchemy database session.

    Returns:
        True if at least one user is linked, False otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(
        exists().where(user_idp_models.UsersIdentityProvider.idp_id == idp_id)
    )
    return db.execute(stmt).scalar() or False

clear_user_identity_provider_refresh_token_by_user_id_and_idp_id

clear_user_identity_provider_refresh_token_by_user_id_and_idp_id(user_id, idp_id, db)

Clear IdP refresh token and metadata.

Called when user logs out, token refresh fails, user unlinks IdP, or security requires token invalidation.

Parameters:

Name	Type	Description	Default
user_id	int	The ID of the user.	required
idp_id	int	The ID of the identity provider.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
bool	True if token was cleared, False if link not found.

Raises:

Type	Description
HTTPException	500 error if database operation fails.

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

@core_decorators.handle_db_errors
def clear_user_identity_provider_refresh_token_by_user_id_and_idp_id(
    user_id: int,
    idp_id: int,
    db: Session,
) -> bool:
    """
    Clear IdP refresh token and metadata.

    Called when user logs out, token refresh fails, user unlinks
    IdP, or security requires token invalidation.

    Args:
        user_id: The ID of the user.
        idp_id: The ID of the identity provider.
        db: SQLAlchemy database session.

    Returns:
        True if token was cleared, False if link not found.

    Raises:
        HTTPException: 500 error if database operation fails.
    """
    db_link = get_user_identity_provider_by_user_id_and_idp_id(
        user_id,
        idp_id,
        db,
    )
    if db_link:
        db_link.idp_refresh_token = None
        db_link.idp_access_token_expires_at = None
        db_link.idp_refresh_token_updated_at = None
        db.commit()
        return True
    return False

create_user_identity_provider

create_user_identity_provider(user_id, idp_id, idp_subject, db)

Create a link between a user and an identity provider.

Parameters:

Name	Type	Description	Default
user_id	int	The ID of the user to link.	required
idp_id	int	The ID of the identity provider.	required
idp_subject	str	The subject identifier from the IdP.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersIdentityProvider	The newly created UsersIdentityProvider link object.

Raises:

Type	Description
HTTPException	409 error if link already exists.
HTTPException	500 error if database operation fails.

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

@core_decorators.handle_db_errors
def create_user_identity_provider(
    user_id: int,
    idp_id: int,
    idp_subject: str,
    db: Session,
) -> user_idp_models.UsersIdentityProvider:
    """
    Create a link between a user and an identity provider.

    Args:
        user_id: The ID of the user to link.
        idp_id: The ID of the identity provider.
        idp_subject: The subject identifier from the IdP.
        db: SQLAlchemy database session.

    Returns:
        The newly created UsersIdentityProvider link object.

    Raises:
        HTTPException: 409 error if link already exists.
        HTTPException: 500 error if database operation fails.
    """
    db_link = user_idp_models.UsersIdentityProvider(
        user_id=user_id,
        idp_id=idp_id,
        idp_subject=idp_subject,
        last_login=func.now(),
    )
    db.add(db_link)
    db.commit()
    db.refresh(db_link)
    return db_link

delete_user_identity_provider

delete_user_identity_provider(user_id, idp_id, db)

Delete link between user and identity provider.

Implements defense-in-depth by clearing sensitive token data before deletion.

Parameters:

Name	Type	Description	Default
user_id	int	The ID of the user.	required
idp_id	int	The ID of the identity provider to unlink.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
bool	True if link was found and deleted, False otherwise.

Raises:

Type	Description
HTTPException	500 error if database operation fails.

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

@core_decorators.handle_db_errors
def delete_user_identity_provider(
    user_id: int,
    idp_id: int,
    db: Session,
) -> bool:
    """
    Delete link between user and identity provider.

    Implements defense-in-depth by clearing sensitive token
    data before deletion.

    Args:
        user_id: The ID of the user.
        idp_id: The ID of the identity provider to unlink.
        db: SQLAlchemy database session.

    Returns:
        True if link was found and deleted, False otherwise.

    Raises:
        HTTPException: 500 error if database operation fails.
    """
    db_link = get_user_identity_provider_by_user_id_and_idp_id(
        user_id,
        idp_id,
        db,
    )
    if db_link:
        # Clear sensitive data first (defense in depth)
        db_link.idp_refresh_token = None
        db_link.idp_access_token_expires_at = None
        db_link.idp_refresh_token_updated_at = None
        db.commit()

        # Then delete the link
        db.delete(db_link)
        db.commit()
        return True
    return False

get_user_identity_provider_by_subject_and_idp_id

get_user_identity_provider_by_subject_and_idp_id(idp_id, idp_subject, db)

Retrieve identity provider link by IdP and subject.

Parameters:

Name	Type	Description	Default
idp_id	int	The ID of the identity provider.	required
idp_subject	str	The subject identifier from the IdP.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersIdentityProvider | None	The matching UsersIdentityProvider record if found, None
UsersIdentityProvider | None	otherwise.

Raises:

Type	Description
HTTPException	500 error if database query fails.

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

@core_decorators.handle_db_errors
def get_user_identity_provider_by_subject_and_idp_id(
    idp_id: int,
    idp_subject: str,
    db: Session,
) -> user_idp_models.UsersIdentityProvider | None:
    """
    Retrieve identity provider link by IdP and subject.

    Args:
        idp_id: The ID of the identity provider.
        idp_subject: The subject identifier from the IdP.
        db: SQLAlchemy database session.

    Returns:
        The matching UsersIdentityProvider record if found, None
        otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(user_idp_models.UsersIdentityProvider).where(
        user_idp_models.UsersIdentityProvider.idp_id == idp_id,
        user_idp_models.UsersIdentityProvider.idp_subject == idp_subject,
    )
    return db.execute(stmt).scalar_one_or_none()

get_user_identity_provider_by_user_id_and_idp_id

get_user_identity_provider_by_user_id_and_idp_id(user_id, idp_id, db)

Retrieve identity provider link for a user.

Parameters:

Name	Type	Description	Default
user_id	int	The ID of the user.	required
idp_id	int	The ID of the identity provider.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersIdentityProvider | None	The UsersIdentityProvider instance if found, None otherwise.

Raises:

Type	Description
HTTPException	500 error if database query fails.

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

@core_decorators.handle_db_errors
def get_user_identity_provider_by_user_id_and_idp_id(
    user_id: int,
    idp_id: int,
    db: Session,
) -> user_idp_models.UsersIdentityProvider | None:
    """
    Retrieve identity provider link for a user.

    Args:
        user_id: The ID of the user.
        idp_id: The ID of the identity provider.
        db: SQLAlchemy database session.

    Returns:
        The UsersIdentityProvider instance if found, None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(user_idp_models.UsersIdentityProvider).where(
        user_idp_models.UsersIdentityProvider.user_id == user_id,
        user_idp_models.UsersIdentityProvider.idp_id == idp_id,
    )
    return db.execute(stmt).scalar_one_or_none()

get_user_identity_provider_refresh_token_by_user_id_and_idp_id

get_user_identity_provider_refresh_token_by_user_id_and_idp_id(user_id, idp_id, db)

Get encrypted refresh token for a user-IdP link.

Convenience wrapper that retrieves the link and extracts the refresh token field. Caller must decrypt using Fernet.

Parameters:

Name	Type	Description	Default
user_id	int	The ID of the user.	required
idp_id	int	The ID of the identity provider.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
str | None	The encrypted refresh token if found, None otherwise.

Raises:

Type	Description
HTTPException	500 error if database query fails.

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

def get_user_identity_provider_refresh_token_by_user_id_and_idp_id(
    user_id: int,
    idp_id: int,
    db: Session,
) -> str | None:
    """
    Get encrypted refresh token for a user-IdP link.

    Convenience wrapper that retrieves the link and extracts
    the refresh token field. Caller must decrypt using Fernet.

    Args:
        user_id: The ID of the user.
        idp_id: The ID of the identity provider.
        db: SQLAlchemy database session.

    Returns:
        The encrypted refresh token if found, None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    db_link = user_idp_crud.get_user_identity_provider_by_user_id_and_idp_id(
        user_id,
        idp_id,
        db,
    )
    if db_link:
        return db_link.idp_refresh_token
    return None

get_user_identity_providers_by_user_id

get_user_identity_providers_by_user_id(user_id, db)

Retrieve all identity provider links for a user.

Parameters:

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

Returns:

Type	Description
list[UsersIdentityProvider]	List of UsersIdentityProvider objects linked to the user.

Raises:

Type	Description
HTTPException	500 error if database query fails.

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

@core_decorators.handle_db_errors
def get_user_identity_providers_by_user_id(
    user_id: int,
    db: Session,
) -> list[user_idp_models.UsersIdentityProvider]:
    """
    Retrieve all identity provider links for a user.

    Args:
        user_id: The ID of the user.
        db: SQLAlchemy database session.

    Returns:
        List of UsersIdentityProvider objects linked to the user.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(user_idp_models.UsersIdentityProvider).where(
        user_idp_models.UsersIdentityProvider.user_id == user_id
    )
    return list(db.execute(stmt).scalars().all())

store_user_identity_provider_tokens

store_user_identity_provider_tokens(user_id, idp_id, encrypted_refresh_token, access_token_expires_at, db)

Store encrypted IdP tokens for a user-IdP link.

Token must be pre-encrypted with Fernet before calling.

Parameters:

Name	Type	Description	Default
user_id	int	The ID of the user.	required
idp_id	int	The ID of the identity provider.	required
encrypted_refresh_token	str	Fernet-encrypted refresh token.	required
access_token_expires_at	datetime	Access token expiry time.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersIdentityProvider | None	The updated UsersIdentityProvider link if found, None
UsersIdentityProvider | None	otherwise.

Raises:

Type	Description
HTTPException	500 error if database operation fails.

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

@core_decorators.handle_db_errors
def store_user_identity_provider_tokens(
    user_id: int,
    idp_id: int,
    encrypted_refresh_token: str,
    access_token_expires_at: datetime,
    db: Session,
) -> user_idp_models.UsersIdentityProvider | None:
    """
    Store encrypted IdP tokens for a user-IdP link.

    Token must be pre-encrypted with Fernet before calling.

    Args:
        user_id: The ID of the user.
        idp_id: The ID of the identity provider.
        encrypted_refresh_token: Fernet-encrypted refresh token.
        access_token_expires_at: Access token expiry time.
        db: SQLAlchemy database session.

    Returns:
        The updated UsersIdentityProvider link if found, None
        otherwise.

    Raises:
        HTTPException: 500 error if database operation fails.
    """
    db_link = get_user_identity_provider_by_user_id_and_idp_id(
        user_id,
        idp_id,
        db,
    )
    if db_link:
        db_link.idp_refresh_token = encrypted_refresh_token
        db_link.idp_access_token_expires_at = access_token_expires_at
        db_link.idp_refresh_token_updated_at = datetime.now(timezone.utc)
        db.commit()
        db.refresh(db_link)
    return db_link

update_user_identity_provider_last_login

update_user_identity_provider_last_login(user_id, idp_id, db)

Update last login timestamp for a user-IdP link.

Parameters:

Name	Type	Description	Default
user_id	int	The ID of the user.	required
idp_id	int	The ID of the identity provider.	required
db	Session	SQLAlchemy database session.	required

Returns:

Type	Description
UsersIdentityProvider | None	The updated UsersIdentityProvider link if found, None
UsersIdentityProvider | None	otherwise.

Raises:

Type	Description
HTTPException	500 error if database operation fails.

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

@core_decorators.handle_db_errors
def update_user_identity_provider_last_login(
    user_id: int,
    idp_id: int,
    db: Session,
) -> user_idp_models.UsersIdentityProvider | None:
    """
    Update last login timestamp for a user-IdP link.

    Args:
        user_id: The ID of the user.
        idp_id: The ID of the identity provider.
        db: SQLAlchemy database session.

    Returns:
        The updated UsersIdentityProvider link if found, None
        otherwise.

    Raises:
        HTTPException: 500 error if database operation fails.
    """
    db_link = get_user_identity_provider_by_user_id_and_idp_id(
        user_id,
        idp_id,
        db,
    )
    if db_link:
        db_link.last_login = datetime.now(timezone.utc)
        db.commit()
        db.refresh(db_link)
    return db_link