app.routes.notifications

Notification routing module.

Provides authenticated endpoints for retrieving notifications, checking the unread badge count, and marking a notification as read.

View Source

 1"""
 2Notification routing module.
 3
 4Provides authenticated endpoints for retrieving notifications, checking the
 5unread badge count, and marking a notification as read.
 6"""
 7
 8from __future__ import annotations
 9
10from datetime import datetime
11from uuid import UUID
12
13from fastapi import APIRouter, Depends, HTTPException, status
14from pydantic import BaseModel, ConfigDict
15from sqlmodel.ext.asyncio.session import AsyncSession
16
17from app.database import get_session
18from app.models import User
19from app.services.auth import current_active_user
20from app.services.notification_service import (
21    NotificationNotFoundError,
22    get_notification_for_user,
23    get_notifications,
24    get_unread_count,
25    mark_read,
26)
27
28router = APIRouter(prefix="/api/notifications", tags=["notifications"])
29
30
31class NotificationRead(BaseModel):
32    model_config = ConfigDict(from_attributes=True)
33
34    id: int
35    bookingID: int
36    message: str
37    type: str
38    isRead: bool
39    createdAt: datetime
40
41
42class UnreadCountRead(BaseModel):
43    count: int
44
45
46
47@router.get("", response_model=list[NotificationRead])
48async def list_notifications(
49    user: User = Depends(current_active_user),
50    session: AsyncSession = Depends(get_session),
51):
52    return await session.run_sync(lambda sync_session: get_notifications(user.id, sync_session))
53
54
55@router.get("/unread-count", response_model=UnreadCountRead)
56async def unread_count(
57    user: User = Depends(current_active_user),
58    session: AsyncSession = Depends(get_session),
59):
60    count = await session.run_sync(lambda sync_session: get_unread_count(user.id, sync_session))
61    return UnreadCountRead(count=count)
62
63
64@router.patch("/{notification_id}/read", response_model=NotificationRead)
65async def read_notification(
66    notification_id: int,
67    user: User = Depends(current_active_user),
68    session: AsyncSession = Depends(get_session),
69):
70    try:
71        await session.run_sync(
72            lambda sync_session: get_notification_for_user(
73                notification_id, user.id, sync_session
74            )
75        )
76        return await session.run_sync(
77            lambda sync_session: mark_read(notification_id, sync_session)
78        )
79    except NotificationNotFoundError as exc:
80        raise HTTPException(
81            status_code=status.HTTP_404_NOT_FOUND,
82            detail=str(exc),
83        ) from exc

router = <fastapi.routing.APIRouter object>

class NotificationRead(pydantic.main.BaseModel): View Source

32class NotificationRead(BaseModel):
33    model_config = ConfigDict(from_attributes=True)
34
35    id: int
36    bookingID: int
37    message: str
38    type: str
39    isRead: bool
40    createdAt: datetime

!!! abstract "Usage Documentation" Models

A base class for creating Pydantic models.

Attributes: __class_vars__: The names of the class variables defined on the model. __private_attributes__: Metadata about the private attributes of the model. __signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom `__init__` function.
__pydantic_decorators__: Metadata containing the decorators defined on the model.
    This replaces `Model.__validators__` and `Model.__root_validators__` from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to
    __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [`RootModel`][pydantic.root_model.RootModel].
__pydantic_serializer__: The `pydantic-core` `SchemaSerializer` used to dump instances of the model.
__pydantic_validator__: The `pydantic-core` `SchemaValidator` used to validate instances of the model.

__pydantic_fields__: A dictionary of field names and their corresponding [`FieldInfo`][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__: A dictionary containing extra values, if [`extra`][pydantic.config.ConfigDict.extra]
    is set to `'allow'`.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

id: int = PydanticUndefined

bookingID: int = PydanticUndefined

message: str = PydanticUndefined

type: str = PydanticUndefined

isRead: bool = PydanticUndefined

createdAt: datetime.datetime = PydanticUndefined

class UnreadCountRead(pydantic.main.BaseModel): View Source

43class UnreadCountRead(BaseModel):
44    count: int

!!! abstract "Usage Documentation" Models

A base class for creating Pydantic models.

__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom `__init__` function.
__pydantic_decorators__: Metadata containing the decorators defined on the model.
    This replaces `Model.__validators__` and `Model.__root_validators__` from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to
    __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [`RootModel`][pydantic.root_model.RootModel].
__pydantic_serializer__: The `pydantic-core` `SchemaSerializer` used to dump instances of the model.
__pydantic_validator__: The `pydantic-core` `SchemaValidator` used to validate instances of the model.

__pydantic_fields__: A dictionary of field names and their corresponding [`FieldInfo`][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__: A dictionary containing extra values, if [`extra`][pydantic.config.ConfigDict.extra]
    is set to `'allow'`.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

count: int = PydanticUndefined

@router.get('', response_model=list[NotificationRead])

async def list_notifications( user: app.models.User = Depends(dependency=<function Authenticator.current_user.<locals>.current_user_dependency>, use_cache=True, scope=None), session: sqlmodel.ext.asyncio.session.AsyncSession = Depends(dependency=<function get_session>, use_cache=True, scope=None)): View Source

48@router.get("", response_model=list[NotificationRead])
49async def list_notifications(
50    user: User = Depends(current_active_user),
51    session: AsyncSession = Depends(get_session),
52):
53    return await session.run_sync(lambda sync_session: get_notifications(user.id, sync_session))

@router.get('/unread-count', response_model=UnreadCountRead)

async def unread_count( user: app.models.User = Depends(dependency=<function Authenticator.current_user.<locals>.current_user_dependency>, use_cache=True, scope=None), session: sqlmodel.ext.asyncio.session.AsyncSession = Depends(dependency=<function get_session>, use_cache=True, scope=None)): View Source

56@router.get("/unread-count", response_model=UnreadCountRead)
57async def unread_count(
58    user: User = Depends(current_active_user),
59    session: AsyncSession = Depends(get_session),
60):
61    count = await session.run_sync(lambda sync_session: get_unread_count(user.id, sync_session))
62    return UnreadCountRead(count=count)

@router.patch('/{notification_id}/read', response_model=NotificationRead)

async def read_notification( notification_id: int, user: app.models.User = Depends(dependency=<function Authenticator.current_user.<locals>.current_user_dependency>, use_cache=True, scope=None), session: sqlmodel.ext.asyncio.session.AsyncSession = Depends(dependency=<function get_session>, use_cache=True, scope=None)): View Source

65@router.patch("/{notification_id}/read", response_model=NotificationRead)
66async def read_notification(
67    notification_id: int,
68    user: User = Depends(current_active_user),
69    session: AsyncSession = Depends(get_session),
70):
71    try:
72        await session.run_sync(
73            lambda sync_session: get_notification_for_user(
74                notification_id, user.id, sync_session
75            )
76        )
77        return await session.run_sync(
78            lambda sync_session: mark_read(notification_id, sync_session)
79        )
80    except NotificationNotFoundError as exc:
81        raise HTTPException(
82            status_code=status.HTTP_404_NOT_FOUND,
83            detail=str(exc),
84        ) from exc