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:

1. Keep a durable history users can return to.
2. 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: { unread_count: data.unread_count }, status: 200 }
    );
  };

  source.addEventListener("connected", onConnected);
  source.addEventListener("notification", onNotification);
  source.addEventListener("unread_count", onUnreadCount);
  source.onerror = () => {
    // EventSource retries automatically unless the connection is closed.
    // Keep the durable list endpoint as the recovery path.
  };

  return () => {
    source.removeEventListener("connected", onConnected);
    source.removeEventListener("notification", onNotification);
    source.removeEventListener("unread_count", onUnreadCount);
    source.close();
  };
}, [queryClient]);

The important correction is addEventListener. If the server sets event: notification, that event will not arrive through source.onmessage.

Invalidating the list on a notification event is intentionally conservative. Prepending into every cached cursor page can put items into the wrong tab, category, or page. If you choose to update the cache directly for a snappier UI, deduplicate by id and update only query keys whose filters match the event.

3. Make read/archive/delete actions optimistic

TanStack Query mutations can update the UI before the server responds. The safe pattern is:

1. Cancel queries that could overwrite the optimistic change.
2. Snapshot the affected caches.
3. Apply the change.
4. Restore the snapshots if the mutation fails.
5. Invalidate the queries after settlement.

For example, mark-read mutations should accept the backend UUID:

import {
  InfiniteData,
  QueryKey,
  useMutation,
  useQueryClient,
} from "@tanstack/react-query";
import { ApiError, patch } from "@/lib/fetcher";

type UnreadCount = {
  data: { unread_count: number };
  status: number;
};

const UNREAD_COUNT_QUERY_KEY = ["notifications", "unread-count"] as const;

type NotificationMutationContext = {
  lists: Array<[QueryKey, InfiniteData<NotificationPage> | undefined]>;
  unreadCount: UnreadCount | undefined;
};

const markRead = (id: string) =>
  patch<void>(`/v1/notifications/${id}/read`, {});

const useMarkNotificationRead = () => {
  const queryClient = useQueryClient();

  return useMutation<void, ApiError, string, NotificationMutationContext>({
    mutationFn: markRead,
    onMutate: async (id) => {
      await queryClient.cancelQueries({ queryKey: ["notifications"] });
      await queryClient.cancelQueries({ queryKey: UNREAD_COUNT_QUERY_KEY });

      const lists = queryClient.getQueriesData<InfiniteData<NotificationPage>>({
        queryKey: ["notifications"],
      });
      const unreadCount =
        queryClient.getQueryData<UnreadCount>(UNREAD_COUNT_QUERY_KEY);
      const target = lists
        .flatMap(([, data]) => data?.pages.flatMap((page) => page.items) ?? [])
        .find((item) => item.id === id);

      // Snapshot first. Then update matching items by their UUID.
      for (const [queryKey, data] of lists) {
        if (!data) continue;
        const tab = (queryKey[1] as { tab?: NotificationTab } | undefined)?.tab;
        queryClient.setQueryData<InfiniteData<NotificationPage>>(
          queryKey,
          {
            ...data,
            pages: data.pages.map((page) => ({
              ...page,
              items:
                tab === "unread"
                  ? page.items.filter((item) => item.id !== id)
                  : page.items.map((item) =>
                      item.id === id ? { ...item, is_read: true } : item
                    ),
            })),
          }
        );
      }

      if (target && !target.is_read) {
        queryClient.setQueryData<UnreadCount>(
          UNREAD_COUNT_QUERY_KEY,
          (old) =>
            old
              ? {
                  ...old,
                  data: {
                    unread_count: Math.max(0, old.data.unread_count - 1),
                  },
                }
              : old
        );
      }

      return { lists, unreadCount };
    },
    onError: (_error, _id, context) => {
      context?.lists.forEach(([queryKey, data]) =>
        queryClient.setQueryData(queryKey, data)
      );
      if (context?.unreadCount) {
        queryClient.setQueryData(
          UNREAD_COUNT_QUERY_KEY,
          context.unreadCount
        );
      }
    },
    onSettled: async () => {
      await queryClient.invalidateQueries({ queryKey: ["notifications"] });
      await queryClient.invalidateQueries({
        queryKey: UNREAD_COUNT_QUERY_KEY,
      });
    },
  });
};

For a real implementation, decrement the cached unread count only when the target item was unread, and remove it from an unread-only list. The attached mutation hooks already use this snapshot/rollback pattern; the key contract correction is changing their identifier from numeric pkid to string id.

Archiving marks a notification as read in the backend. Ensure the unread count is updated for other connected clients too: publish a fresh count after archive, or have the client invalidate/refetch its count. The backend should be the source of truth for that count.

Where background tasks fit

Despite the common “background task” framing, this implementation does not use FastAPI BackgroundTasks for notification delivery. It stores a notification and publishes the Redis message during the request.

That is deliberate for a simple first version: the notification record is part of the application workflow, and the live publish is a small operation. FastAPI background tasks run in the API process after the response; they are not a durable queue. A process restart can lose pending work, and sharing a request-scoped SQLAlchemy session with a task is unsafe.

If email, push delivery, retries, or high-volume fan-out become requirements, use a durable worker pattern. A transactional outbox is a practical next step:

1. Save the notification and an outbox event in one database transaction.
2. A worker claims undelivered outbox rows.
3. The worker publishes and retries failures.
4. Mark events delivered idempotently.

Use a task queue when you need durable execution, controlled concurrency, retries, or scheduling—not just because a task happens after a request.

Operational and security checklist

• Derive the user and organization from authenticated dependencies, not request payloads.
• Scope every notification query and mutation by notification ID, user ID, and organization ID.
• Keep Redis channels user- and organization-specific.
• Validate category values and bound cursor page sizes.
• Treat Redis delivery as best effort; retrieve durable history from PostgreSQL.
• Close Redis Pub/Sub subscriptions when clients disconnect.
• Send heartbeats and configure reverse proxies not to buffer the SSE response.
• Avoid logging notification bodies or sensitive metadata.
• Set CORS and cookie settings deliberately for cross-origin EventSource connections.
• Monitor Redis connection health; application startup may succeed in degraded mode, but live delivery will not be available until Redis recovers.
• For reliable publishing after database commit, use an outbox rather than assuming a database flush is a commit.

Integration gaps to close before calling the attached code end-to-end

The backend and frontend snippets show the intended architecture, but the attached files currently have contract differences worth fixing:

1. SSE framing: the backend generator yields preformatted strings and passes them to EventSourceResponse. sse-starlette encodes ordinary strings as data: values. Yield ServerSentEvent objects or event dictionaries for custom event names.
2. Named SSE handlers: the frontend uses source.onmessage, while the backend names events. Use addEventListener("notification", ...) and the matching listeners for connected and unread_count.
3. Identifiers: the backend exposes UUID id; the frontend types and mutations use numeric pkid. Use the UUID string consistently.
4. Cursor fields: the backend cursor page exposes next_page; the frontend query currently reads next_cursor. Read the actual CursorPage field.
5. Categories: align frontend category unions and UI labels with the backend values (SYSTEM, COMMUNICATION, ADMINISTRATIVE, PAYMENTS, EXPENSES, and TRIPS).
6. Unread count after archive: archiving marks the record read, so also publish the updated count or refetch it.
7. Transaction ordering: Redis publication currently occurs before the request-scoped database transaction commits. Treat the event as a hint and refetch, or add a transactional outbox for reliable ordering.

These are not reasons to abandon the design. They are exactly the sorts of details an explicit end-to-end contract makes easier to spot.

Closing

A small notification system can be built from familiar pieces: store user-visible history in PostgreSQL, publish live hints through Redis Pub/Sub, stream them with SSE, and let TanStack Query manage the browser cache.

The simple design works best when its guarantees are stated honestly. Redis Pub/Sub is not durable, SSE is one-way, and a live event should not replace the history API. Align the event names, IDs, pagination fields, and categories across the backend and frontend; then add an outbox or worker only when delivery requirements justify the extra machinery.