notifika

A reusable, technology-agnostic notification & event bus for Python, with identity-aware real-time delivery. Postgres + Keycloak are the preferred adapters, but the core depends only on Protocol ports, so any backend fits.

Status: MVP (v0.1, in development). Implemented and tested: dependency-free core, in-memory adapters, an optional FastAPI/SSE integration, and the preferred-stack adapters — Postgres (NotificationStore, cached PolicyStore, LISTEN/NOTIFY EventBus), Keycloak (IdentityResolver), and a multi-worker SSE relay (RelayRealtimePush).

What it does

Match an event against configurable policies, resolve recipients via an identity provider (e.g. Keycloak groups/roles), and deliver to those users over pluggable channels — including a real-time in-app channel (SSE) for live UI toasts. Example: "when an L1 group performs an action, only the L2 group sees the toast." That routing is just a policy — fully configurable at runtime, not hardcoded.

The gap it fills: existing Postgres-native Python libraries are single-consumer job queues; the event-driven libraries with great DX don't support a Postgres/Keycloak-first, identity-aware notification model. notifika targets that intersection at low-to-moderate (human-driven) volume.

Architecture

Ports & adapters. The dependency-free core owns the domain:

• Models — Event (a generic, owned envelope; Event.from_audit_event() adapts a duck-typed audit event), NotificationPolicy, Target, Notification.
• Logic — PolicyEngine (match), NotifyService (publish → resolve → dispatch), Dispatcher (persist → deliver, failures isolated per channel).
• Ports (notifika.ports) — EventBus, IdentityResolver, NotificationStore, NotificationChannel, RealtimePush, PolicyStore / MutablePolicyStore.

Adapters are optional extras. In-memory implementations ship for tests and single-process use; the FastAPI/SSE integration lives behind the sse extra.

pip install notifika                 # core only (zero deps)
pip install "notifika[sse]"          # + FastAPI SSE integration
pip install "notifika[postgres,keycloak,sse]"   # full preferred stack (planned adapters)

Quickstart

One call wires the whole in-memory stack (demos, tests, single-process apps):

from notifika import Event, NotificationPolicy, Target, in_memory

nk = in_memory(
    groups={"/ops/l2": ["alice", "bob"]},
    policies=[NotificationPolicy.on("contract.updated", Target.group("/ops/l2"))],
)

queue = nk.connect("alice")                          # what an SSE endpoint opens per connection
await nk.publish(Event(type="contract.updated", actor_id="carol"))  # an L1 user acts...
msg = await queue.get()                              # ...only the L2 group gets the live toast

in_memory(...) returns an InMemoryNotifika bundle exposing the wired .service plus the adapters it was built from — .realtime, .store, .policies, .resolver — so tests can drive them. There's exactly one publish path: nk.publish(...) forwards to service.publish(Event(...)).

from notifika import (
    Dispatcher, Event, InAppChannel, InMemoryIdentityResolver,
    InMemoryNotificationStore, InMemoryPolicyStore, InMemoryRealtimePush,
    NotificationPolicy, NotifyService, PolicyEngine, Target,
)

realtime = InMemoryRealtimePush()
resolver = InMemoryIdentityResolver(groups={"/ops/l2": ["alice", "bob"]})
store = InMemoryNotificationStore()

policies = InMemoryPolicyStore([
    NotificationPolicy(
        name="l1-action-escalates-to-l2",
        match={"type": "contract.updated"},
        targets=(Target(kind="group", value="/ops/l2", channels=("in_app",)),),
    )
])

service = NotifyService(
    policy_store=policies,
    engine=PolicyEngine(),
    resolver=resolver,
    dispatcher=Dispatcher({"in_app": InAppChannel(realtime)}, store=store, resolver=resolver),
)

# An L1 user acts -> only the L2 group is notified in real time.
await service.publish(Event(type="contract.updated", actor_id="carol", payload={"resource_id": "42"}))

For AI agents & coding assistants

The library carries its own orientation — AGENTS.md — and ships it inside the wheel (installed at <site-packages>/notifika/AGENTS.md). So it reaches you with no docs site and no network, even from an airgapped Nexus PyPI mirror:

python -m notifika                                   # prints the AGENTS.md guide
python -c "import notifika; print(notifika.overview())"

AGENTS.md is the single source of truth: the mental model (Event → policy match → resolve recipients → dispatch → channel), the one-call quickstart, the ergonomic constructors, the production adapter swap, and what the library deliberately does not do. notifika.overview() returns its text and tests/guide_test.py executes its quickstart, so it can't drift from the API. Everything is py.typed; help(notifika) / help(notifika.NotifyService) expand each piece.

Production adapters (Postgres + Keycloak)

Swap the in-memory adapters for the preferred stack — same ports, so nothing in your wiring logic changes:

import asyncpg
from notifika import Dispatcher, InAppChannel, NotifyService, PolicyEngine, RelayRealtimePush, InMemoryRealtimePush
from notifika.postgres import PostgresNotificationStore, PostgresPolicyStore, PgListenNotifyBus
from notifika.keycloak import KeycloakIdentityResolver

pool = await asyncpg.create_pool(dsn)
store = PostgresNotificationStore(pool)
policies = PostgresPolicyStore(pool)            # cached: sync reads, async upsert/delete/refresh
await store.ensure_schema(); await policies.ensure_schema(); await policies.refresh()

resolver = KeycloakIdentityResolver(
    server_url="https://kc.example.com", realm="app",
    client_id="notifika-svc", client_secret=secret,   # service account (client_credentials)
)

# Multi-worker real-time: a push on any worker reaches the user's SSE connection
# on whichever worker holds it, via Postgres LISTEN/NOTIFY.
bus = PgListenNotifyBus(pool, connect=lambda: asyncpg.connect(dsn))
realtime = RelayRealtimePush(InMemoryRealtimePush(), bus)
await realtime.start()                            # stop() on shutdown

service = NotifyService(
    policy_store=policies, engine=PolicyEngine(), resolver=resolver,
    dispatcher=Dispatcher({"in_app": InAppChannel(realtime)}, store=store, resolver=resolver),
)

PgListenNotifyBus payloads cross Postgres NOTIFY (< 8000 bytes) — it carries compact refs (user id, event type, small payloads), never large bodies. In a multi-worker deployment, broadcast a "policies changed" signal over the bus and call policies.refresh() so every worker's policy cache converges.

Adding channels (email, SMS, webhook, …)

A channel is any object implementing the NotificationChannel port — a channel_name and async send(recipient_id, contact, payload). notifika resolves contact for you via IdentityResolver.get_user_contact(user_id, channel) (the Keycloak adapter returns the user's email for "email", or a matching attribute such as a phone number for "sms"), then the Dispatcher hands it to your channel. InAppChannel is the reference implementation.

class EmailChannel:
    channel_name = "email"
    def __init__(self, smtp): self._smtp = smtp
    async def send(self, recipient_id, contact, payload):
        if not contact:
            raise ValueError(f"no email on file for {recipient_id}")  # MUST raise on permanent failure
        await self._smtp.send(to=contact, subject=payload["event_type"], body=render(payload))

# register it, then reference the name from a policy target:
dispatcher = Dispatcher({"in_app": InAppChannel(realtime), "email": EmailChannel(smtp)}, store=store, resolver=resolver)
# Target(kind="group", value="/ops/l2", channels=("in_app", "email"))

Guidance: raise on permanent failure so the Dispatcher marks the notification failed (delivery is already failure-isolated per channel); keep send quick and idempotent (a payload may be re-sent on retry); SMS/push/webhook are identical — implement send, resolve the address via get_user_contact, register under a channel name, list that name in the target's channels. Automatic retry of failed/transient deliveries belongs in a worker over NotificationStore.get_pending (roadmap, not yet built).

Configuring policies at runtime

NotifyService re-reads the policy store on every publish, so changes apply live. The library provides the building blocks for a management surface — MutablePolicyStore (upsert/get/delete) and policy (de)serialization (NotificationPolicy.from_dict / .to_dict) — so an application can expose policy CRUD however it likes.

notifika deliberately does not ship policy-CRUD HTTP routes. Editing "who-gets-notified-about-what" is privileged, and the authorization model is application-specific; an opinionated router would also freeze HTTP shape into the library's public API. So your app owns that route and its admin gate — it's ~15 lines on top of the building blocks (see examples/fastapi_app.py):

from fastapi import APIRouter, Body, Depends
admin = APIRouter(dependencies=[Depends(require_admin)])   # YOUR admin gate (fail-closed)

@admin.post("/notification-policies", status_code=201)
async def create_policy(body: dict = Body(...)):
    policies.upsert(NotificationPolicy.from_dict(body))    # MutablePolicyStore + serialization
    return {"ok": True}

notifika does ship an optional, low-stakes router for the per-user read path (SSE stream + list/mark-read your own notifications):

from notifika.asgi import build_router
app.include_router(build_router(realtime=realtime, store=store, current_user=current_user))

Because the browser EventSource API can't send an Authorization header, present the Keycloak token to the SSE stream via an HttpOnly cookie or a fetch-based SSE client; resolve it in current_user.

Development

pip install pytest
cd notifika && python -m pytest          # core tests need no extra deps
pip install "fastapi" "sse-starlette" httpx && python -m pytest   # incl. FastAPI tests

Tests use plain pytest + asyncio.run() (no pytest-asyncio), matching the repo's house style. Adapter unit tests run against fakes (no live services). Real-database integration tests are skipped unless NOTIFIKA_TEST_DSN is set. Run them against a throwaway Postgres via the bundled compose file:

docker compose run --rm tests        # spins up postgres:16, runs tests/integration_test.py
docker compose down -v

Or against any reachable Postgres directly:

NOTIFIKA_TEST_DSN=postgresql://user:pw@localhost:5432/db \
  pip install asyncpg && python -m pytest tests/integration_test.py