API Reference

Notifications module for user notification management.

This module handles the creation, retrieval, and management of user notifications including activity alerts, follower requests, and admin approval notifications.

Exports

• CRUD: get_user_notification_by_id, get_user_notifications, get_user_notifications_count, get_user_notifications_with_pagination, create_notification, mark_notification_as_read
• Schemas: NotificationBase, NotificationCreate, NotificationRead
• Models: Notification (ORM model)
• Constants: NotificationType
• Utils: create_new_activity_notification, create_new_duplicate_start_time_activity_notification, create_new_follower_request_notification, create_accepted_follower_request_notification, create_admin_new_sign_up_approval_request_notification

NotificationBase

Bases: BaseModel

Base schema for notification data.

Attributes:

Name	Type	Description
type	StrictInt | None	Type of the notification.
options	dict | None	Additional metadata for the notification.

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

class NotificationBase(BaseModel):
    """
    Base schema for notification data.

    Attributes:
        type: Type of the notification.
        options: Additional metadata for the notification.
    """

    type: StrictInt | None = Field(
        default=None,
        description="Type of the notification",
    )
    options: dict | None = Field(
        default=None,
        description="Additional notification metadata",
    )

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

NotificationCreate

Bases: NotificationBase

Schema for creating a notification record.

Attributes:

Name	Type	Description
user_id	StrictInt	FK to user.

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

class NotificationCreate(NotificationBase):
    """
    Schema for creating a notification record.

    Attributes:
        user_id: FK to user.
    """

    user_id: StrictInt = Field(
        ...,
        ge=1,
        description="FK to user",
    )

NotificationModel

Bases: Base

User notification data.

Attributes:

Name	Type	Description
id	Mapped[int]	Primary key.
user_id	Mapped[int]	Foreign key to users table.
type	Mapped[int]	Notification type.
options	Mapped[dict[str, Any] | None]	Notification options (JSON).
read	Mapped[bool]	Whether the notification has been read.
created_at	Mapped[datetime]	Notification creation date.
users	Mapped[Users]	Relationship to Users model.

Source code in backend/app/notifications/models.py

class Notification(Base):
    """
    User notification data.

    Attributes:
        id: Primary key.
        user_id: Foreign key to users table.
        type: Notification type.
        options: Notification options (JSON).
        read: Whether the notification has been read.
        created_at: Notification creation date.
        users: Relationship to Users model.
    """

    __tablename__ = "notifications"

    id: Mapped[int] = mapped_column(
        primary_key=True,
        autoincrement=True,
    )
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
        comment="User ID that the notification belongs to",
    )
    type: Mapped[int] = mapped_column(
        nullable=False,
        comment="Notification type",
    )
    options: Mapped[dict[str, Any] | None] = mapped_column(
        JSON,
        nullable=True,
        comment="Notification options (JSON)",
    )
    read: Mapped[bool] = mapped_column(
        default=False,
        nullable=False,
        comment=("Has the notification been read (True) or not (False)"),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        default=func.now(),
        comment=("Notification creation date (DateTime)"),
    )

    # Define a relationship to the Users model
    users: Mapped["Users"] = relationship(back_populates="notifications")

NotificationRead

Bases: NotificationBase

Schema for reading a notification record.

Attributes:

Name	Type	Description
id	StrictInt	Unique notification identifier.
user_id	StrictInt	FK to user.
read	StrictBool	Whether the notification has been read.
created_at	datetime | None	Timestamp of creation.

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

class NotificationRead(NotificationBase):
    """
    Schema for reading a notification record.

    Attributes:
        id: Unique notification identifier.
        user_id: FK to user.
        read: Whether the notification has been read.
        created_at: Timestamp of creation.
    """

    id: StrictInt = Field(..., description="Unique notification ID")
    user_id: StrictInt = Field(
        ...,
        ge=1,
        description="FK to user",
    )
    read: StrictBool = Field(
        default=False,
        description="Whether notification is read",
    )
    created_at: datetime | None = Field(
        default=None,
        description="Timestamp of creation",
    )

    @field_serializer("created_at")
    def serialize_created_at(self, value: datetime | None) -> str | None:
        """
        Serialize created_at as date string.

        Args:
            value: The datetime value to serialize.

        Returns:
            Date string in YYYY-MM-DD format or None.
        """
        if value is None:
            return None
        return value.strftime("%Y-%m-%d")

serialize_created_at

serialize_created_at(value)

Serialize created_at as date string.

Parameters:

Name	Type	Description	Default
value	datetime | None	The datetime value to serialize.	required

Returns:

Type	Description
str | None	Date string in YYYY-MM-DD format or None.

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

@field_serializer("created_at")
def serialize_created_at(self, value: datetime | None) -> str | None:
    """
    Serialize created_at as date string.

    Args:
        value: The datetime value to serialize.

    Returns:
        Date string in YYYY-MM-DD format or None.
    """
    if value is None:
        return None
    return value.strftime("%Y-%m-%d")

NotificationType

Bases: IntEnum

Enumeration of notification types.

Attributes:

Name	Type	Description
NEW_ACTIVITY		New activity notification.
DUPLICATE_ACTIVITY		Duplicate activity notification.
NEW_FOLLOWER_REQUEST		New follower request notification.
NEW_FOLLOWER_REQUEST_ACCEPTED		Follower request accepted notification.
ADMIN_NEW_SIGN_UP_APPROVAL_REQUEST		Admin sign-up approval request notification.

Source code in backend/app/notifications/constants.py

class NotificationType(IntEnum):
    """
    Enumeration of notification types.

    Attributes:
        NEW_ACTIVITY: New activity notification.
        DUPLICATE_ACTIVITY: Duplicate activity notification.
        NEW_FOLLOWER_REQUEST: New follower request notification.
        NEW_FOLLOWER_REQUEST_ACCEPTED: Follower request accepted notification.
        ADMIN_NEW_SIGN_UP_APPROVAL_REQUEST: Admin sign-up approval request
            notification.
    """

    NEW_ACTIVITY = 1
    DUPLICATE_ACTIVITY = 2
    NEW_FOLLOWER_REQUEST = 11
    NEW_FOLLOWER_REQUEST_ACCEPTED = 12
    GARMIN_TOKEN_EXPIRED = 21
    ADMIN_NEW_SIGN_UP_APPROVAL_REQUEST = 101

create_accepted_follower_request_notification async

create_accepted_follower_request_notification(user_id, target_user_id, websocket_manager, db)

Create an accepted follower request notification.

Parameters:

Name	Type	Description	Default
user_id	int	The accepting user ID.	required
target_user_id	int	The user to notify.	required
websocket_manager	WebSocketManager	WebSocket manager instance.	required
db	Session	Database session.	required

Returns:

Type	Description
Notification	The created Notification model.

Raises:

Type	Description
HTTPException	If user not found or error.

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

async def create_accepted_follower_request_notification(
    user_id: int,
    target_user_id: int,
    websocket_manager: websocket_manager.WebSocketManager,
    db: Session,
) -> notifications_models.Notification:
    """
    Create an accepted follower request notification.

    Args:
        user_id: The accepting user ID.
        target_user_id: The user to notify.
        websocket_manager: WebSocket manager instance.
        db: Database session.

    Returns:
        The created Notification model.

    Raises:
        HTTPException: If user not found or error.
    """
    try:
        user = users_crud.get_user_by_id(user_id, db)
        if not user:
            raise HTTPException(
                status_code=(status.HTTP_404_NOT_FOUND),
                detail="User not found",
            )
        return await _create_and_notify(
            notifications_schema.NotificationCreate(
                user_id=target_user_id,
                type=(
                    notifications_constants.NotificationType.NEW_FOLLOWER_REQUEST_ACCEPTED
                ),
                options={
                    "user_id": user_id,
                    "user_name": user.name,
                    "user_username": user.username,
                },
            ),
            "NEW_FOLLOWER_REQUEST_ACCEPTED" "_NOTIFICATION",
            user_id,
            websocket_manager,
            db,
        )
    except HTTPException as http_err:
        raise http_err
    except Exception as err:
        core_logger.print_to_log(
            f"Error in create_accepted_follower_request_notification: {err}",
            "error",
            exc=err,
        )
        raise HTTPException(
            status_code=(status.HTTP_500_INTERNAL_SERVER_ERROR),
            detail="Internal Server Error",
        ) from err

create_admin_new_sign_up_approval_request_notification async

create_admin_new_sign_up_approval_request_notification(user, websocket_manager, db)

Notify all admins of a new sign-up request.

Parameters:

Name	Type	Description	Default
user	Users	The user requesting sign-up.	required
websocket_manager	WebSocketManager	WebSocket manager instance.	required
db	Session	Database session.	required

Returns:

Type	Description
None	None.

Raises:

Type	Description
HTTPException	If creation or notify fails.

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

async def create_admin_new_sign_up_approval_request_notification(
    user: users_models.Users,
    websocket_manager: websocket_manager.WebSocketManager,
    db: Session,
) -> None:
    """
    Notify all admins of a new sign-up request.

    Args:
        user: The user requesting sign-up.
        websocket_manager: WebSocket manager instance.
        db: Database session.

    Returns:
        None.

    Raises:
        HTTPException: If creation or notify fails.
    """
    try:
        admins = users_utils.get_admin_users_or_404(db)
        for admin in admins:
            await _create_and_notify(
                notifications_schema.NotificationCreate(
                    user_id=admin.id,
                    type=(
                        notifications_constants.NotificationType.ADMIN_NEW_SIGN_UP_APPROVAL_REQUEST
                    ),
                    options={
                        "user_id": user.id,
                        "user_name": user.name,
                        "user_username": (user.username),
                    },
                ),
                "ADMIN_NEW_SIGN_UP_APPROVAL" "_REQUEST_NOTIFICATION",
                admin.id,
                websocket_manager,
                db,
            )
    except HTTPException as http_err:
        raise http_err
    except Exception as err:
        core_logger.print_to_log(
            "Error in create_admin_new_sign_up_approval_request"
            f"_notification: {err}",
            "error",
            exc=err,
        )
        raise HTTPException(
            status_code=(status.HTTP_500_INTERNAL_SERVER_ERROR),
            detail="Internal Server Error",
        ) from err

create_new_activity_notification async

create_new_activity_notification(user_id, activity_id, websocket_manager)

Create a new activity notification.

Parameters:

Name	Type	Description	Default
user_id	int	The user ID to notify.	required
activity_id	int	The new activity ID.	required
websocket_manager	WebSocketManager	WebSocket manager instance.	required

Returns:

Type	Description
Notification	The created Notification model.

Raises:

Type	Description
HTTPException	If creation or notify fails.

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

async def create_new_activity_notification(
    user_id: int,
    activity_id: int,
    websocket_manager: websocket_manager.WebSocketManager,
) -> notifications_models.Notification:
    """
    Create a new activity notification.

    Args:
        user_id: The user ID to notify.
        activity_id: The new activity ID.
        websocket_manager: WebSocket manager instance.

    Returns:
        The created Notification model.

    Raises:
        HTTPException: If creation or notify fails.
    """
    with SessionLocal() as db:
        try:
            return await _create_and_notify(
                notifications_schema.NotificationCreate(
                    user_id=user_id,
                    type=(notifications_constants.NotificationType.NEW_ACTIVITY),
                    options={
                        "activity_id": activity_id,
                    },
                ),
                "NEW_ACTIVITY_NOTIFICATION",
                user_id,
                websocket_manager,
                db,
            )
        except HTTPException as http_err:
            raise http_err
        except Exception as err:
            core_logger.print_to_log(
                f"Error in create_new_activity_notification: {err}",
                "error",
                exc=err,
            )
            raise HTTPException(
                status_code=(status.HTTP_500_INTERNAL_SERVER_ERROR),
                detail="Internal Server Error",
            ) from err

create_new_duplicate_start_time_activity_notification async

create_new_duplicate_start_time_activity_notification(user_id, activity_id, websocket_manager)

Create a duplicate start time notification.

Parameters:

Name	Type	Description	Default
user_id	int	The user ID to notify.	required
activity_id	int	The duplicate activity ID.	required
websocket_manager	WebSocketManager	WebSocket manager instance.	required

Returns:

Type	Description
Notification	The created Notification model.

Raises:

Type	Description
HTTPException	If creation or notify fails.

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

async def create_new_duplicate_start_time_activity_notification(
    user_id: int,
    activity_id: int,
    websocket_manager: websocket_manager.WebSocketManager,
) -> notifications_models.Notification:
    """
    Create a duplicate start time notification.

    Args:
        user_id: The user ID to notify.
        activity_id: The duplicate activity ID.
        websocket_manager: WebSocket manager instance.

    Returns:
        The created Notification model.

    Raises:
        HTTPException: If creation or notify fails.
    """
    with SessionLocal() as db:
        try:
            return await _create_and_notify(
                notifications_schema.NotificationCreate(
                    user_id=user_id,
                    type=(notifications_constants.NotificationType.DUPLICATE_ACTIVITY),
                    options={
                        "activity_id": activity_id,
                    },
                ),
                "NEW_DUPLICATE_ACTIVITY_START_TIME" "_NOTIFICATION",
                user_id,
                websocket_manager,
                db,
            )
        except HTTPException as http_err:
            raise http_err
        except Exception as err:
            core_logger.print_to_log(
                "Error in create_new_duplicate_start_time_activity"
                f"_notification: {err}",
                "error",
                exc=err,
            )
            raise HTTPException(
                status_code=(status.HTTP_500_INTERNAL_SERVER_ERROR),
                detail="Internal Server Error",
            ) from err

create_new_follower_request_notification async

create_new_follower_request_notification(user_id, target_user_id, websocket_manager, db)

Create a new follower request notification.

Parameters:

Name	Type	Description	Default
user_id	int	The requesting user ID.	required
target_user_id	int	The user to notify.	required
websocket_manager	WebSocketManager	WebSocket manager instance.	required
db	Session	Database session.	required

Returns:

Type	Description
Notification	The created Notification model.

Raises:

Type	Description
HTTPException	If user not found or error.

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

async def create_new_follower_request_notification(
    user_id: int,
    target_user_id: int,
    websocket_manager: websocket_manager.WebSocketManager,
    db: Session,
) -> notifications_models.Notification:
    """
    Create a new follower request notification.

    Args:
        user_id: The requesting user ID.
        target_user_id: The user to notify.
        websocket_manager: WebSocket manager instance.
        db: Database session.

    Returns:
        The created Notification model.

    Raises:
        HTTPException: If user not found or error.
    """
    try:
        user = users_crud.get_user_by_id(user_id, db)
        if not user:
            raise HTTPException(
                status_code=(status.HTTP_404_NOT_FOUND),
                detail="User not found",
            )
        return await _create_and_notify(
            notifications_schema.NotificationCreate(
                user_id=target_user_id,
                type=(notifications_constants.NotificationType.NEW_FOLLOWER_REQUEST),
                options={
                    "user_id": user_id,
                    "user_name": user.name,
                    "user_username": user.username,
                },
            ),
            "NEW_FOLLOWER_REQUEST_NOTIFICATION",
            target_user_id,
            websocket_manager,
            db,
        )
    except HTTPException as http_err:
        raise http_err
    except Exception as err:
        core_logger.print_to_log(
            f"Error in create_new_follower_request_notification: {err}",
            "error",
            exc=err,
        )
        raise HTTPException(
            status_code=(status.HTTP_500_INTERNAL_SERVER_ERROR),
            detail="Internal Server Error",
        ) from err

create_notification

create_notification(notification, db)

Create a new notification record.

Parameters:

Name	Type	Description	Default
notification	NotificationCreate	The notification data to create.	required
db	Session	Database session.	required

Returns:

Type	Description
Notification	The newly created Notification model.

Raises:

Type	Description
HTTPException	If a database error occurs.

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

@core_decorators.handle_db_errors
def create_notification(
    notification: notifications_schema.NotificationCreate,
    db: Session,
) -> notifications_models.Notification:
    """
    Create a new notification record.

    Args:
        notification: The notification data to create.
        db: Database session.

    Returns:
        The newly created Notification model.

    Raises:
        HTTPException: If a database error occurs.
    """
    new_notification = notifications_models.Notification(
        user_id=notification.user_id,
        type=notification.type,
        options=notification.options,
        read=False,
        created_at=func.now(),
    )
    db.add(new_notification)
    db.commit()
    db.refresh(new_notification)
    return new_notification

get_user_notification_by_id

get_user_notification_by_id(notification_id, user_id, db)

Retrieve a notification by ID for a user.

Parameters:

Name	Type	Description	Default
notification_id	int	The notification ID.	required
user_id	int	The owning user ID.	required
db	Session	Database session.	required

Returns:

Type	Description
Notification | None	Notification model if found, otherwise None.

Raises:

Type	Description
HTTPException	If a database error occurs.

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

@core_decorators.handle_db_errors
def get_user_notification_by_id(
    notification_id: int,
    user_id: int,
    db: Session,
) -> notifications_models.Notification | None:
    """
    Retrieve a notification by ID for a user.

    Args:
        notification_id: The notification ID.
        user_id: The owning user ID.
        db: Database session.

    Returns:
        Notification model if found, otherwise None.

    Raises:
        HTTPException: If a database error occurs.
    """
    stmt = select(notifications_models.Notification).where(
        notifications_models.Notification.user_id == user_id,
        notifications_models.Notification.id == notification_id,
    )
    return db.execute(stmt).scalars().first()

get_user_notifications

get_user_notifications(user_id, db)

Retrieve all notifications for a user.

Parameters:

Name	Type	Description	Default
user_id	int	The owning user ID.	required
db	Session	Database session.	required

Returns:

Type	Description
list[Notification]	List of Notification models for the user.

Raises:

Type	Description
HTTPException	If a database error occurs.

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

@core_decorators.handle_db_errors
def get_user_notifications(
    user_id: int,
    db: Session,
) -> list[notifications_models.Notification]:
    """
    Retrieve all notifications for a user.

    Args:
        user_id: The owning user ID.
        db: Database session.

    Returns:
        List of Notification models for the user.

    Raises:
        HTTPException: If a database error occurs.
    """
    stmt = select(notifications_models.Notification).where(
        notifications_models.Notification.user_id == user_id
    )
    return db.execute(stmt).scalars().all()

get_user_notifications_count

get_user_notifications_count(user_id, db)

Count notifications for a user.

Parameters:

Name	Type	Description	Default
user_id	int	The owning user ID.	required
db	Session	Database session.	required

Returns:

Type	Description
int	Number of notifications for the user.

Raises:

Type	Description
HTTPException	If a database error occurs.

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

@core_decorators.handle_db_errors
def get_user_notifications_count(
    user_id: int,
    db: Session,
) -> int:
    """
    Count notifications for a user.

    Args:
        user_id: The owning user ID.
        db: Database session.

    Returns:
        Number of notifications for the user.

    Raises:
        HTTPException: If a database error occurs.
    """
    stmt = (
        select(func.count())
        .select_from(notifications_models.Notification)
        .where(notifications_models.Notification.user_id == user_id)
    )
    return db.execute(stmt).scalar_one()

get_user_notifications_with_pagination

get_user_notifications_with_pagination(user_id, db, page_number=1, num_records=5)

Retrieve paginated notifications for a user.

Parameters:

Name	Type	Description	Default
user_id	int	The owning user ID.	required
db	Session	Database session.	required
page_number	int	Page number (default 1).	1
num_records	int	Records per page (default 5).	5

Returns:

Type	Description
list[Notification]	List of Notification models for the page.

Raises:

Type	Description
HTTPException	If a database error occurs.

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

@core_decorators.handle_db_errors
def get_user_notifications_with_pagination(
    user_id: int,
    db: Session,
    page_number: int = 1,
    num_records: int = 5,
) -> list[notifications_models.Notification]:
    """
    Retrieve paginated notifications for a user.

    Args:
        user_id: The owning user ID.
        db: Database session.
        page_number: Page number (default 1).
        num_records: Records per page (default 5).

    Returns:
        List of Notification models for the page.

    Raises:
        HTTPException: If a database error occurs.
    """
    stmt = (
        select(notifications_models.Notification)
        .where(notifications_models.Notification.user_id == user_id)
        .order_by(desc(notifications_models.Notification.created_at))
        .offset((page_number - 1) * num_records)
        .limit(num_records)
    )
    return db.execute(stmt).scalars().all()

mark_notification_as_read

mark_notification_as_read(notification_id, user_id, db)

Mark a notification as read for a user.

Parameters:

Name	Type	Description	Default
notification_id	int	The notification ID.	required
user_id	int	The owning user ID.	required
db	Session	Database session.	required

Returns:

Type	Description
Notification | None	Updated Notification model, or None if not found.

Raises:

Type	Description
HTTPException	If a database error occurs.

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

@core_decorators.handle_db_errors
def mark_notification_as_read(
    notification_id: int,
    user_id: int,
    db: Session,
) -> notifications_models.Notification | None:
    """
    Mark a notification as read for a user.

    Args:
        notification_id: The notification ID.
        user_id: The owning user ID.
        db: Database session.

    Returns:
        Updated Notification model, or None if not found.

    Raises:
        HTTPException: If a database error occurs.
    """
    stmt = select(notifications_models.Notification).where(
        notifications_models.Notification.user_id == user_id,
        notifications_models.Notification.id == notification_id,
    )
    notification = db.execute(stmt).scalars().first()
    if notification is None:
        return None
    notification.read = True
    db.commit()
    db.refresh(notification)
    return notification