Metadata-Version: 2.4
Name: notifika
Version: 0.1.1
Summary: Reusable, technology-agnostic notification & event bus for FastAPI + Postgres with identity-aware real-time delivery.
Project-URL: Homepage, https://github.com/vanmarkic/audit-logger
Project-URL: Repository, https://github.com/vanmarkic/audit-logger
License-Expression: MIT
Keywords: event-bus,fastapi,keycloak,notifications,postgres,sse
Classifier: Framework :: FastAPI
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.11
Provides-Extra: all
Requires-Dist: asyncpg>=0.24.0; extra == 'all'
Requires-Dist: fastapi<1.0,>=0.110; extra == 'all'
Requires-Dist: httpx>=0.27; extra == 'all'
Requires-Dist: sse-starlette>=2.0; extra == 'all'
Provides-Extra: dev
Requires-Dist: asyncpg>=0.24.0; extra == 'dev'
Requires-Dist: fastapi<1.0,>=0.110; extra == 'dev'
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: sse-starlette>=2.0; extra == 'dev'
Provides-Extra: keycloak
Requires-Dist: httpx>=0.27; extra == 'keycloak'
Provides-Extra: postgres
Requires-Dist: asyncpg>=0.24.0; extra == 'postgres'
Provides-Extra: sse
Requires-Dist: fastapi<1.0,>=0.110; extra == 'sse'
Requires-Dist: sse-starlette>=2.0; extra == 'sse'
Description-Content-Type: text/markdown

# 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):

```python
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(...))`.

<details>
<summary>…or wire the object graph yourself (what <code>in_memory</code> does under the hood)</summary>

```python
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"}))
```

</details>

## For AI agents & coding assistants

The library carries its own orientation — [`AGENTS.md`](./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:

```bash
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:

```python
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.

```python
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`):

```python
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):

```python
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

```bash
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:

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

Or against any reachable Postgres directly:

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