A Lightweight Real-Time Notification System with FastAPI, Redis Pub/Sub, SSE, and React Query
A Lightweight Real-Time Notification System with FastAPI, Redis Pub/Sub, SSE, and React Query
A useful notification feature needs to do two things well:
- Keep a durable history users can return to.
- Show new activity quickly without asking users to refresh the page.
Those are related problems, but they are not the same problem. A database is a good source of truth for notification history. A live connection is useful for timely updates. For a small application, PostgreSQL, Redis Pub/Sub, Server-Sent Events (SSE), and TanStack Query can cover both without introducing a separate broker and worker fleet.
This post walks through that design using a FastAPI backend and a React/Next.js frontend.
Scope: this is a lightweight, best-effort real-time design. It does not promise exactly-once delivery or replay of every event. The database is durable; Redis Pub/Sub is not.
Architecture
Business operation │ ▼ Notification service ├── Insert notification into PostgreSQL └── Publish a live event to Redis Pub/Sub │ ▼ FastAPI SSE endpoint │ ▼ Browser EventSource │ ▼ TanStack Query cache / UI Notification history and unread count ──► PostgreSQL The distinction is important:
- PostgreSQL stores notification history and read/archive state.
- Redis Pub/Sub broadcasts updates to clients listening right now.
- SSE keeps a server-to-browser connection open over HTTP.
- TanStack Query handles history, unread counts, and cache updates in the UI.
If a browser is offline or disconnects, it can fetch the durable notification list when it reconnects. It cannot replay missed Redis Pub/Sub messages.
Backend: persist notifications as domain data
1. Model and schema
Each notification belongs to a user and an organization. That scope should be applied consistently to reads, updates, deletes, and live channels.
class Notification(Base, UUIDPrimaryKeyMixin, TenantMixin, TimestampMixin): __tablename__ = "notifications" user_id: Mapped[UUID] = mapped_column( ForeignKey("auth_user.id", ondelete="CASCADE"), nullable=False, index=True, ) title: Mapped[str] = mapped_column(String(255), nullable=False) body: Mapped[str | None] = mapped_column(Text) category: Mapped[str] = mapped_column(String(30), nullable=False) action_url: Mapped[str | None] = mapped_column(String(500)) meta: Mapped[dict | None] = mapped_column(JSONB) is_read: Mapped[bool] = mapped_column( Boolean, nullable=False, server_default=text("false") ) is_archived: Mapped[bool] = mapped_column( Boolean, nullable=False, server_default=text("false") ) meta is for optional, structured context such as a trip_id or expense_id. Keep it as JSON data; do not treat it as trusted HTML.
The Pydantic response schema should expose the same identifier and fields in both ordinary API responses and live events:
class NotificationResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: UUID organization_id: UUID user_id: UUID title: str body: str | None category: str action_url: str | None meta: dict[str, Any] | None is_read: bool is_archived: bool created_at: datetime In the deployed model, the primary key is a UUID named id. The frontend must use that identifier; it should not assume a numeric pkid.
An Alembic migration creates the table, foreign keys, and indexes. Indexes on (user_id, is_read) and (user_id, is_archived) help the common unread and archive filters.
2. Keep database access in the repository
The repository owns SQLAlchemy queries. Every user-facing query includes both user_id and organization_id; this prevents an ID from another tenant from being used to read or mutate a notification.
def list_query( self, user_id: UUID, organization_id: UUID, tab: str, category: str | None, ): stmt = select(Notification).where( Notification.user_id == user_id, Notification.organization_id == organization_id, ) if tab == "unread": stmt = stmt.where( Notification.is_read.is_(False), Notification.is_archived.is_(False), ) elif tab == "archive": stmt = stmt.where(Notification.is_archived.is_(True)) else: stmt = stmt.where(Notification.is_archived.is_(False)) if category: stmt = stmt.where(Notification.category == category) return stmt.order_by(Notification.created_at.desc()) Unread counts should use the same definition as the unread list:
stmt = select(func.count()).where( Notification.user_id == user_id, Notification.organization_id == organization_id, Notification.is_read.is_(False), Notification.is_archived.is_(False), ) Cursor pagination keeps list requests small as notification history grows. The FastAPI route uses CursorPage; the frontend should read its next_page field to request the next page.
3. Create notifications from domain services
Business services know when an event matters. They call the notification service rather than coupling domain logic to Redis channel names or serialization.
await notification_service.create_notification( NotificationCreate( user_id=trip.created_by_id, organization_id=organization_id, title="Trip allocated", body=f"Trip {trip.transaction_number} was allocated.", category=NotificationCategory.TRIPS, meta={"trip_id": str(trip.id)}, ), redis=redis, ) The notification service persists via the repository and then publishes a JSON-safe representation:
notification = await self.repo.create(payload) event_data = NotificationResponse.model_validate( notification, from_attributes=True, ).model_dump(mode="json") await publish_notification( redis, payload.organization_id, payload.user_id, {"event_type": "notification", **event_data}, ) The attached implementation publishes during the request while the request-scoped database session is still open. The repository flushes the insert, and the request dependency commits later. This means Redis can deliver an event before the database transaction commits. An immediate history refetch can race that commit, too. For a small system, refetch with a small delay/retry or accept eventual UI refresh; do not claim that the event and database commit are atomic.
For stronger consistency, write an outbox row in the same transaction as the notification. A worker can publish committed outbox records and retry failures. That is a reliability upgrade, not a feature Redis Pub/Sub provides by itself.
4. Publish and subscribe on user-scoped channels
Use organization and user IDs in channel names:
channel = f"notifications:{organization_id}:{user_id}" await redis.publish(channel, json.dumps(event_data)) Unread-count changes can use a separate channel:
channel = f"notifications:count:{organization_id}:{user_id}" await redis.publish(channel, json.dumps({"unread_count": count})) Do not let clients choose these IDs. Derive the user and active organization from the authenticated request, and only subscribe the server-side stream to those values.
5. Encode real SSE events correctly
SSE has a wire format with named events, data, and optional comments. With sse-starlette, yield ServerSentEvent objects (or supported event dictionaries), not strings that already contain event: and data: lines. A plain string is encoded as an SSE data: value; wrapping preformatted SSE text as a string does not preserve it as event framing.
import asyncio import json from redis.asyncio import Redis from sse_starlette.sse import ServerSentEvent async def notification_stream( redis: Redis, user_id: UUID, organization_id: UUID, initial_count: int, ): channel = f"notifications:{organization_id}:{user_id}" pubsub = redis.pubsub() try: await pubsub.subscribe(channel) yield ServerSentEvent( event="connected", data=json.dumps({"unread_count": initial_count}), ) while True: try: message = await asyncio.wait_for( pubsub.get_message(ignore_subscribe_messages=True), timeout=15, ) except TimeoutError: yield ServerSentEvent(comment="heartbeat") continue if message and message["type"] == "message": payload = json.loads(message["data"]) event_name = payload.pop("event_type", "notification") yield ServerSentEvent( event=event_name, data=json.dumps(payload), ) finally: await pubsub.unsubscribe(channel) await pubsub.aclose() The timeout is handled inside the loop so heartbeats continue while the stream is idle. Also retain cancellation handling and cleanup; an SSE connection can live much longer than an ordinary request.
The route returns the stream using EventSourceResponse:
ActiveMemberDep = Annotated[ OrganizationMember, Depends(resolve_active_member), ] CurrentUserDep = Annotated[ UserResponseSchema, Depends(get_current_active_user), ] DbSessionDep = Annotated[ AsyncSession, Depends(get_db_session, scope="function"), ] RedisDep = Annotated[Redis, Depends(get_redis_from_request)] @router.get("/stream") async def stream_notifications( member: ActiveMemberDep, current_user: CurrentUserDep, db: DbSessionDep, redis: RedisDep, ): initial_count = await notification_service(db).get_unread_count( current_user.id, member.organization_id, ) events = notification_stream( redis, current_user.id, member.organization_id, initial_count, ) return EventSourceResponse(events) Authentication and active-membership resolution happen before returning the stream. The stream’s Redis subscription is then scoped to that authenticated user and organization.
6. Expose history and state-changing endpoints
A small API usually needs:
| Method | Path | Purpose |
|---|---|---|
GET | /v1/notifications?tab=all&size=10 | Cursor-paginated history |
GET | /v1/notifications/unread-count | Current unread count |
GET | /v1/notifications/stream | Named SSE events |
PATCH | /v1/notifications/{id}/read | Mark one as read |
PATCH | /v1/notifications/mark-all-read | Mark visible unread notifications as read |
PATCH | /v1/notifications/{id}/archive | Archive one notification |
DELETE | /v1/notifications/{id} | Delete one notification |
Use UUID path parameters in the backend. In the example application, the notification router is mounted under /query/api/v1/notifications; the frontend reaches the upstream service through its /api Next.js proxy. Confirm your deployment’s proxy/base URL instead of copying a path without checking it.
Frontend: make the API contract explicit
The attached frontend types and component use pkid: number, next_cursor, and category names such as BOOKING and SCHEDULING. The backend returns id: UUID, cursor-page fields including next_page, and categories such as TRIPS, EXPENSES, and PAYMENTS. These contracts need to match or be translated at one explicit boundary.
For the API shown above, define the frontend types from the backend contract:
export type NotificationCategory = | "SYSTEM" | "COMMUNICATION" | "ADMINISTRATIVE" | "PAYMENTS" | "EXPENSES" | "TRIPS"; export type NotificationTab = "all" | "unread" | "archive"; export interface Notification { id: string; // UUID serialized as a string organization_id: string; user_id: string; title: string; body: string | null; category: NotificationCategory; action_url: string | null; meta: Record<string, unknown> | null; is_read: boolean; is_archived: boolean; created_at: string; } export interface NotificationPage { items: Notification[]; next_page: string | null; previous_page: string | null; total: number; } Use id consistently for React keys, mutation variables, cache matching, and API routes. If an older frontend needs a numeric pkid, build and test an adapter; do not silently assume the backend UUID can be used as a number.
1. Query history and unread count
The frontend API functions call the durable endpoints. getNextPageParam must match the backend cursor response:
export const useInfiniteNotifications = ({ tab, category, }: { tab: NotificationTab; category?: NotificationCategory; }) => useInfiniteQuery({ queryKey: ["notifications", { tab, category }], initialPageParam: null as string | null, queryFn: ({ pageParam }) => get<NotificationPage>("/v1/notifications", { params: { tab, category, size: 10, cursor: pageParam, }, }), getNextPageParam: (lastPage) => lastPage.next_page ?? undefined, }); The unread count is a separate query so the badge can be updated independently:
export const useNotificationUnreadCount = () => useQuery({ queryKey: ["notifications", "unread-count"], queryFn: () => get<{ data: { unread_count: number }; status: number }>( "/v1/notifications/unread-count" ), staleTime: 30_000, retry: false, }); The current Next.js fetcher sends requests to a same-origin /api/... route, which forwards cookies upstream. Use that same origin for SSE when possible:
const source = new EventSource("/api/v1/notifications/stream", { withCredentials: true, }); This avoids hard-coding a backend host in a client component. For direct cross-origin SSE, configure CORS and cookie attributes deliberately, and use the correct upstream base path (the backend router is mounted beneath /query/api in this application).
2. Listen for named events—not only onmessage
The backend sends named SSE events: connected, notification, and unread_count. Browser EventSource.onmessage handles only unnamed message events. Register listeners for each custom event:
useEffect(() => { const source = new EventSource("/api/v1/notifications/stream", { withCredentials: true, }); const onConnected = (event: Event) => { const data = JSON.parse((event as MessageEvent<string>).data) as { unread_count: number; }; queryClient.setQueryData(UNREAD_COUNT_QUERY_KEY, (old: UnreadCount | undefined) => old ? { ...old, data: { unread_count: data.unread_count } } : { data: { unread_count: data.unread_count }, status: 200 } ); }; const onNotification = (_event: Event) => { // History is durable; refetch it instead of trusting a transient event // to maintain cursor pages and every active tab/category filter. void queryClient.invalidateQueries({ queryKey: ["notifications"] }); void queryClient.invalidateQueries({ queryKey: UNREAD_COUNT_QUERY_KEY }); }; const onUnreadCount = (event: Event) => { const data = JSON.parse((event as MessageEvent<string>).data) as { unread_count: number; }; queryClient.setQueryData(UNREAD_COUNT_QUERY_KEY, (old: UnreadCount | undefined) => old ? { ...old, data: { unread_count: data.unread_count } } : { data: