Skip to content

Push Notifications

Web Push notification subscription management.

This module handles push notification subscriptions from client browsers using the Web Push protocol (RFC 8030). Subscriptions are persisted to PostgreSQL so they survive container restarts and deployments.

Example

Client subscribes: POST /push/subscribe {"endpoint": "https://...", "keys": {"p256dh": "...", "auth": "..."}}

PushKeys

Bases: BaseModel

Encryption keys for Web Push messages.

Attributes:

Name Type Description
p256dh str

Public key for message encryption (Base64).
auth str

Authentication secret for message encryption (Base64).
Source code in app/push.py 
26
27
28
29
30
31
32
33
34
35
class PushKeys(BaseModel):
    """Encryption keys for Web Push messages.

    Attributes:
        p256dh: Public key for message encryption (Base64).
        auth: Authentication secret for message encryption (Base64).
    """

    p256dh: str
    auth: str

PushSubscriptionIn

Bases: BaseModel

Web Push subscription from a client browser.

Attributes:

Name Type Description
endpoint str

Push service endpoint URL (browser-specific).
expirationTime int | None

Optional subscription expiry timestamp.
keys PushKeys

Encryption keys for secure message delivery.
Source code in app/push.py 
38
39
40
41
42
43
44
45
46
47
48
49
class PushSubscriptionIn(BaseModel):
    """Web Push subscription from a client browser.

    Attributes:
        endpoint: Push service endpoint URL (browser-specific).
        expirationTime: Optional subscription expiry timestamp.
        keys: Encryption keys for secure message delivery.
    """

    endpoint: str
    expirationTime: int | None = None
    keys: PushKeys

subscribe 

subscribe(sub, request, db=Depends(get_session))

Register a new push notification subscription.

Requires authentication via access_token cookie. Upserts by (user_id, endpoint) to prevent duplicates.

Parameters:

Name Type Description Default
sub PushSubscriptionIn

Push subscription from browser's Push API.

 required
request Request

HTTP request (for cookie validation).

 required
db Session

Database session.

 Depends(get_session)

Returns:

Name Type Description
dict dict[str, bool | int]

Status with ok=True and total subscription count for user.

Raises:

Type Description
HTTPException

401 if not authenticated.
Source code in app/push.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@router.post("/subscribe")
def subscribe(
    sub: PushSubscriptionIn,
    request: Request,
    db: Session = Depends(get_session),
) -> dict[str, bool | int]:
    """Register a new push notification subscription.

    Requires authentication via access_token cookie.
    Upserts by (user_id, endpoint) to prevent duplicates.

    Args:
        sub: Push subscription from browser's Push API.
        request: HTTP request (for cookie validation).
        db: Database session.

    Returns:
        dict: Status with ok=True and total subscription count for user.

    Raises:
        HTTPException: 401 if not authenticated.
    """
    user_id = _get_user_id_from_request(request, db)

    # Check if subscription already exists for this user + endpoint
    existing = db.scalar(
        select(PushSubscriptionModel).where(
            PushSubscriptionModel.user_id == user_id,
            PushSubscriptionModel.endpoint == sub.endpoint,
        )
    )

    if existing:
        # Update keys in case they changed
        existing.keys_p256dh = sub.keys.p256dh  # gitleaks:allow
        existing.keys_auth = sub.keys.auth
    else:
        db.add(
            PushSubscriptionModel(
                user_id=user_id,
                endpoint=sub.endpoint,
                keys_p256dh=sub.keys.p256dh,
                keys_auth=sub.keys.auth,
            )
        )

    db.commit()

    count = db.scalar(
        select(func.count(PushSubscriptionModel.id)).where(
            PushSubscriptionModel.user_id == user_id
        )
    )
    return {"ok": True, "count": count or 0}