Metadata-Version: 2.4
Name: vintasend
Version: 3.1.1
Summary: A flexible package for implementing transactional notifications
License-Expression: MIT
License-File: LICENSE.txt
Author: Hugo bessa
Author-email: hugo@vinta.com.br
Requires-Python: >=3.10,<3.15
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Dist: requests (>=2.34.2,<3.0.0)
Requires-Dist: typing-extensions (>=4.10.0)
Description-Content-Type: text/markdown

# VintaSend

A flexible package for implementing transactional notifications in Python projects.

## Features
* **Storing notifications in a Database**: This package relies on a data store to record all the notifications that will be sent. It also keeps it's state column up to date.
* **Scheduling notifications**: Storing notifications to be send in the future. The notification's context for rendering the template is only evaluated at the moment the notification is sent due to the lib's context generation registry.
* **Notification context fetched at send time**: On scheduled notifications, we only get the notification context at the send time, so we always get the most up-to-date information.
* **AsyncIO Support**: We have two different versions of our service, one that only supports sync backends/adapters, and the other that only supports AsyncIO backends/adapters. 
* **File Attachments**: Support for adding file attachments to notifications with various input types including file paths, URLs, bytes data, and file-like objects. File storage is a pluggable seam of its own — see [Attachment storage](#attachment-storage) below and [ATTACHMENTS.md](ATTACHMENTS.md) for the full picture.
* **One-off Notifications**: Send notifications directly to email addresses or phone numbers without requiring user IDs from your database.
* **Flexible backend**: Your projects database is getting slow after you created the first milion notifications? You can migrate to a faster no-sql database with a blink of an eye without affecting how you send the notifications.
* **Flexible adapters**: Your project probably will need to change how it sends notifications overtime. This package allows to change the adapter without having to change how notifications templates are rendered or how the notification themselves are stored.
* **Flexible template renderers**: Wanna start managing your templates with a third party tool (so non-technical people can help maintaining them)? Or even choose a more powerful rendering engine? You can do it independetly of how you send the notifications or store them in the database.
* **Store-backed templates**: Templates don't have to be files. A [template manager](#template-managers) keeps them in a data store with versioning, a publishing lifecycle, and composition of one template from another — so copy changes ship without a deploy, and a notification can be pinned to the exact version it should render.


## Installation

To install `vintasend` you just need to run 

```shell
pip install vintasend
```

**Disclaimer**: Although the VintaSend package is self-sufficient and can be used alone, you'd have to implement at least one Backend, one Adapter, and one Template Renderer. We have a bunch of additional packages that implement these and need to be installed seprately depending on your needs. Please check [Officially supported packages](#officially-supported-packages) section.


## Getting Started

To start using VintaSend you just need to import the notification service and start using it to manage your notifications.

```python
from vintasend.services.notification_service import NotificationService, register_context
from vintasend.services.dataclasses import NotificationAttachment
from vintasend.constants import NotificationTypes


notification_backend = MyNotificationBackend() 
notifications_service = NotificationService(
    notification_adapters=[MyAdapter(MyTemplateRenderer(), notification_backend)], 
    notification_backend=notification_backend,
)


@register_context("my_context_generator")
def my_context_generator(user_id: str):
    user = get_user_by_id(user_id)
    
    return {
        "first_name": user.first_name,
        "last_name": user.last_name,
        "joined_at": user.joined_at,
    }


notifications_service.create_notification(
    user_id=user.id,
    notification_type=NotificationTypes.EMAIL,
    title="My Notification", # this is just for auditing purposes
    body_template="my/notification/template/path.html",
    context_name="my_context_generator",
    context_kwargs={
        "user_id": user.id,
    },
    send_after=datetime.datetime.now(),
    subject_template="my/notification/subject/template/path.txt",
    preheader_template="my/notification/preheader/template/path.html",
)
```

### Adding Attachments to Notifications

VintaSend supports adding attachments to notifications. You can attach files using various input types:

```python
from vintasend.services.dataclasses import NotificationAttachment

# Attach a file from disk
file_attachment = NotificationAttachment(
    filename="document.pdf",
    file="/path/to/document.pdf",  # File path
    content_type="application/pdf",  # Optional, auto-detected if not provided
    description="Important document",
    is_inline=False,
)

# Attach from bytes data
bytes_attachment = NotificationAttachment(
    filename="report.txt",
    file=b"Report content here",
    content_type="text/plain",
)

# Attach from a URL
url_attachment = NotificationAttachment(
    filename="image.jpg",
    file="https://example.com/image.jpg",  # URL will be downloaded
    is_inline=True,  # For inline images in HTML emails
)

# Attach from file-like object
import io
buffer = io.BytesIO(b"File content")
buffer_attachment = NotificationAttachment(
    filename="data.bin",
    file=buffer,
)

# Create notification with attachments
notifications_service.create_notification(
    user_id=user.id,
    notification_type=NotificationTypes.EMAIL,
    title="Email with Attachments",
    body_template="notification/with/attachments.html",
    context_name="my_context_generator",
    context_kwargs={"user_id": user.id},
    attachments=[file_attachment, bytes_attachment, url_attachment],
)
```

### Attachment storage

The bytes behind an attachment are stored by an **attachment manager**, a fourth pluggable seam
alongside the backend, adapter, and template renderer. A notification backend never opens a file,
calls S3, or downloads a URL itself — it only persists a checksum-indexed file record and a join
row, and hands an opaque `storage_identifiers` value back to whichever manager is configured.

Pass a manager to the service the same way you pass a backend or an adapter, either as an instance
or a dotted import string, or configure it once through the `NOTIFICATION_ATTACHMENT_MANAGER`
setting:

```python
notifications_service = NotificationService(
    notification_adapters=[MyAdapter(MyTemplateRenderer(), notification_backend)],
    notification_backend=notification_backend,
    attachment_manager=MyAttachmentManager(),  # or a dotted import string
)
```

A backend written before this seam existed needs no changes: injection is duck-typed, so a
backend without an `inject_attachment_manager` method simply never receives one and keeps working
without attachment support.

Identical uploads are deduplicated by checksum, so attaching the same bytes twice stores one file
record and two lightweight join rows rather than two copies. You can also skip uploading entirely
and attach an already-stored file by id, with `NotificationAttachmentReference(file_id=...)`
instead of `NotificationAttachment(file=...)`.

`AsyncIONotificationService` takes the same `attachment_manager` argument and works against an
`AsyncIOBaseAttachmentManager`.

See [ATTACHMENTS.md](ATTACHMENTS.md) for the manager's ABC, the full data model
(`AttachmentFileRecord`, `StoredAttachment`, `StorageIdentifiers`), the seven methods a
notification backend implements, and the orphaned-file reclamation pattern.

### One-off Notifications

VintaSend also supports one-off notifications for sending notifications directly to email addresses or phone numbers without requiring a user ID from your database:

```python
from vintasend.services.dataclasses import NotificationAttachment

# Send a one-off notification with attachments
notifications_service.create_one_off_notification(
    email_or_phone="user@example.com",  # Email address or phone number
    first_name="John",
    last_name="Doe",
    notification_type=NotificationTypes.EMAIL,
    title="Welcome Email",
    body_template="welcome/email.html",
    context_name="welcome_context",
    context_kwargs={"welcome_message": "Welcome to our service!"},
    send_after=datetime.datetime.now(),
    subject_template="Welcome to Our Service",
    attachments=[
        NotificationAttachment(
            filename="welcome_guide.pdf",
            file="/path/to/guide.pdf",
            description="Getting started guide"
        )
    ],
)
```

This is useful for:
- Welcome emails to new signups before they're fully registered
- Marketing campaigns to external email lists
- Notifications to external parties who aren't users in your system

`email_or_phone` is validated when the notification is created. It must be either an email
address or a phone number of 10 to 15 digits, optionally prefixed with `+`. Empty, whitespace-only
and malformed values raise `InvalidOneOffNotificationRecipientError` before anything is persisted,
so a recipient that can never be delivered to is rejected up front rather than at send time. The
check is on format only -- it does not verify that the address or number actually exists.

### Scheduled notifications

VintaSend schedules notifications by creating them on the database for sending when the `send_after` value has passed. The sending isn't done automatically but we have a service method called `send_pending_notifications` to send all pending notifications found in the database.

You need to call the `send_pending_notifications` service method in a cron job or a tool like Celery Beat.

### In-app notifications (listing, counts and marking as read)

In-app notifications live in the same backend as every other notification. Once an in-app
notification has been sent it has status `SENT`; once the user has seen it, it becomes `READ`.
Internal pipeline states (`PENDING_SEND`, `FAILED`, `CANCELLED`) are never exposed by the in-app
listing methods.

All listing methods are paginated (`page` defaults to `1`, `page_size` to `10`) and pair with a
`count_*` method so you can build `count` / `next` / `previous` envelopes. All of them require an
in-app adapter to be configured, otherwise they raise `NotificationError`.

```python
# Unread only (status == SENT)
unread = notification_service.get_in_app_unread(user_id, page=1, page_size=10)
unread_count = notification_service.get_in_app_unread_count(user_id)

# Read + unread ("all", status in SENT/READ), newest-first
all_notifications = notification_service.get_in_app_notifications(user_id, page=1, page_size=10)
all_count = notification_service.get_in_app_notifications_count(user_id)
```

To mark notifications as read you can mark a single one or do it in bulk:

```python
# Single — raises NotificationUpdateError if the notification is not currently SENT
notification_service.mark_read(notification_id)

# Bulk — idempotent. ids that are already read, missing, not owned by `user_id`
# (when passed) or in a non-SENT state are simply skipped (never an error). Returns
# the final READ state for the requested ids. Always pass `user_id` for endpoints so
# you never touch another user's rows.
read_notifications = notification_service.mark_read_bulk(
    [id_1, id_2, id_3], user_id=user_id
)
```

The `AsyncIONotificationService` exposes the same methods as coroutines (`await ...`).

> Prefer the paginated `get_in_app_notifications` / `get_in_app_unread` plus the matching
> `count_*` methods. The backend's unpaginated `filter_all_*` variants exist mainly for internal
> and count use.

### Filtering and Ordering Notifications

`NotificationService` and `AsyncIONotificationService` also expose a general-purpose query --
`filter_notifications`, `count_notifications`, `get_backend_supported_filter_capabilities` and
`resend_notification` -- built for a monitoring dashboard that needs to search, page through, and
retry notifications across every user and notification type, not just one user's in-app list.

#### The filter grammar

A filter is a plain `dict` built from the `TypedDict`s in
`vintasend.services.notification_backends.filters` (`NotificationFilter` and friends), so it
round-trips through JSON with no adapter layer -- a dashboard backend can accept a filter straight
from a request body and hand it to `filter_notifications` unchanged.

**Field filters.** Multiple keys inside one filter dict are an implicit AND. A scalar value means
equality; a list means membership:

```python
from vintasend.constants import NotificationStatus

# tenant == "acme" AND status == SENT
notification_service.filter_notifications(
    {"tenant": "acme", "status": NotificationStatus.SENT.value},
    page=1,
    page_size=10,
)

# tenant in ("acme", "beta")
notification_service.filter_notifications({"tenant": ["acme", "beta"]}, page=1, page_size=10)
```

**String lookups** apply to `body_template`, `subject_template` and `context_name`. A bare `str`
is a case-sensitive `exact` match. For anything else, pass a `StringFilterLookup` dict with
`lookup` (`"exact"`, `"starts_with"`, `"ends_with"` or `"includes"`), `value`, and an optional
`case_sensitive` (defaults to `True`):

```python
notification_service.filter_notifications(
    {
        "body_template": {
            "lookup": "starts_with",
            "value": "WELCOME",
            "case_sensitive": False,
        }
    },
    page=1,
    page_size=10,
)
```

**Date ranges** apply to `send_after_range`, `created_at_range`, `sent_at_range` and
`read_at_range`. Both bounds are **inclusive** (`from` maps to `>=`, `to` maps to `<=`); a client
computing "today" as midnight-to-midnight will double-count boundary rows if it assumes an
exclusive upper bound instead. Either bound can be left out for an open-ended range:

```python
import datetime

now = datetime.datetime.now(tz=datetime.timezone.utc)

# sent in the last 24 hours
notification_service.filter_notifications(
    {"sent_at_range": {"from": now - datetime.timedelta(days=1), "to": now}},
    page=1,
    page_size=10,
)
```

**Template versions.** `requested_template_version` and `used_template_version` filter on the
two integer fields a notification records when its template renderer versions templates --
what it was pinned to, and what it actually rendered. Scalar or list, like the other
membership fields, but the candidates must be real `int`s: `"3"` is a malformed filter rather
than a request for version 3, and one bad candidate rejects the whole leaf.

```python
# still pinned to v3
notification_service.filter_notifications(
    {"requested_template_version": 3}, page=1, page_size=10
)

# went out on v1 or v2 -- the query after finding a bug in an old version
notification_service.filter_notifications(
    {"used_template_version": [1, 2]}, page=1, page_size=10
)
```

Both are `None` on a notification whose renderer has no versions, and `used_template_version`
is `None` until it has been sent. The NULL rule above applies unchanged, so those rows never
match a positive version filter and are included under negation -- which also means there is
no positive way to ask for "the unpinned ones".

**`and` / `or` / `not`** compose and nest arbitrarily -- each takes the same `NotificationFilter`
shape, so a group can hold field filters or further groups:

```python
# Everything except acme's notifications and every reminder email
notification_service.filter_notifications(
    {
        "not": {
            "or": [
                {"tenant": "acme"},
                {"body_template": "reminder_email"},
            ]
        }
    },
    page=1,
    page_size=10,
)

# acme's sent welcome emails specifically
notification_service.filter_notifications(
    {
        "and": [
            {"tenant": "acme"},
            {"status": NotificationStatus.SENT.value},
            {"body_template": {"lookup": "includes", "value": "welcome"}},
        ]
    },
    page=1,
    page_size=10,
)
```

A positive filter on a field whose stored value is `None` never matches, so a `None` row is
*included* under negation: `{"not": {"tenant": "acme"}}` returns rows whose tenant is anything but
`acme`, plus rows whose tenant is `None`.

**`filter_notifications({}, page, page_size)` -- an empty filter -- is the unrestricted listing.**
There is no separate "get everything" method; the empty dict matches every notification, not none
of them.

#### Ordering

`order_by` selects one primary field and a direction:

```python
notification_service.filter_notifications(
    {}, page=1, page_size=10, order_by={"field": "sent_at", "direction": "desc"}
)
```

The available fields are `send_after`, `sent_at`, `read_at`, `created_at` and `updated_at`.
`created_at` maps to the notification's `created` column and `updated_at` maps to `modified` --
the filter API's field names don't have to match the underlying column names one-to-one. Leaving
`order_by` out defaults to `created_at` descending. Every backend appends `id` as a tiebreaker in
the same direction as the primary field, so paging through rows that share a timestamp still
returns each row exactly once: without a stable tiebreaker, offset pagination over a repeated
`created` value would silently drop or duplicate rows across pages.

#### Pagination

`page` and `page_size` behave like every other paginated method in this library. The result is an
`Iterable`, so you can't call `len()` on it -- use `count_notifications` for the total a dashboard
needs to render `count` / `next` / `previous`:

```python
page_1 = notification_service.filter_notifications({}, page=1, page_size=20)
total = notification_service.count_notifications({})
```

There's no enforced maximum `page_size`; the library leaves that decision to the host
application. A caller that passes an unreasonably large `page_size` loads that many rows into
memory in one call, so an endpoint that exposes this to a client should apply its own cap.

**Pages are 1-indexed: `page=1` is the first page.** Every backend in this library follows that,
and so does every paginated method, not just `filter_notifications`. Don't hardcode it if you're
writing something that paginates a backend you didn't write -- read `pagination.oneIndexed` from
the capability report below. Getting this wrong is silent: nothing raises, you just serve the
wrong page or an empty first one. It matters most when crossing ecosystems, because the
`vintasend-ts` sibling library's backends are 0-indexed.

#### Capability introspection

A backend doesn't have to support every filter field, string lookup and sort field.
`get_backend_supported_filter_capabilities()` returns a `dict[str, bool]` a dashboard can use to
grey out controls the configured backend can't handle:

```python
capabilities = notification_service.get_backend_supported_filter_capabilities()
capabilities["fields.tenant"]          # True unless the backend declines it
capabilities["fields.usedTemplateVersion"]
capabilities["stringLookups.startsWith"]
capabilities["orderBy.sentAt"]
capabilities["logical.or"]             # can it assemble `or` groups?
capabilities["negation.sentAtRange"]   # can it negate a date range?
capabilities["pagination.oneIndexed"]  # True when page 1 is the first page
```

The namespaces are:

| Namespace | Reports |
| --- | --- |
| `fields.*` | Whether a filter field can be used at all |
| `stringLookups.*` | Which string lookups, and which case sensitivities, are available |
| `logical.*` | Whether `and` / `or` / `not` groups can be assembled |
| `negation.*` | Whether a date range specifically can be negated |
| `orderBy.*` | Which sort fields are available |
| `pagination.*` | The backend's pagination convention |

`logical.not` and `logical.notNested` are separate questions: a query builder that can
negate a single predicate can't always negate a whole subtree, so a backend may support
`{"not": {"tenant": "acme"}}` and decline `{"not": {"or": [...]}}`.

`negation.*` covers only date ranges because that's where backends actually struggle. A
negated membership or string lookup is a plain `NOT IN` / `NOT LIKE`, but a negated range
has to include `None` rows to satisfy this library's negation semantics (see above), and
not every query builder expresses that. So `fields.sentAtRange: True` with
`negation.sentAtRange: False` is a coherent, expected combination.

`pagination.oneIndexed` is the odd one out: it reports a *convention* rather than a feature, and
it's there because that convention is invisible when you get it wrong. A caller that assumes the
wrong base doesn't get an exception, it gets the wrong rows. Anything translating between its own
page numbers and a backend's -- an HTTP layer, a report exporter, a sync job -- should read this
key instead of assuming:

```python
caps = notification_service.get_backend_supported_filter_capabilities()
backend_page = page if caps["pagination.oneIndexed"] else page - 1
```

A backend declares only what it *cannot* do. Its report is merged over an all-`True` default, so a
missing key means supported:

```python
class TenantUnsupportedBackend(FakeFileBackend):
    def get_filter_capabilities(self) -> dict[str, bool]:
        return {"fields.tenant": False}


limited_service = NotificationService(
    notification_adapters=[my_adapter],
    notification_backend=TenantUnsupportedBackend(),
)
caps = limited_service.get_backend_supported_filter_capabilities()
caps["fields.tenant"]  # False
caps["fields.status"]  # still True -- everything not declared stays supported
```

This means a filter field added in a later `vintasend` release doesn't force every existing
backend to re-declare support for everything else.

Notice that capability keys are camelCase and dotted (`'fields.notificationType'`,
`'orderBy.sentAt'`) while filter field names are snake_case (`notification_type`,
`send_after_range`). This is deliberate, not an inconsistency: filter fields are an in-process
Python API that `mypy` checks and that you type by hand, so snake_case is correct there.
Capability keys are data a client reads over the wire, and they're kept byte-identical to the
`vintasend-ts` sibling library's keys wherever both libraries have the key, so one dashboard can
consume a capability report from a Python backend or a TypeScript one without a translation table
in between.

One key means different things by default in the two ecosystems, which matters if you're
writing something that consumes both:

| Key | Note |
| --- | --- |
| `pagination.oneIndexed` | Defined by both, but the **defaults differ because the backends do**: `True` here, `false` in `vintasend-ts`, whose backends page from 0. Never assume a value across ecosystems — read it. |

The only remaining difference is `read_at`: this library defines `fields.readAtRange` and
`negation.readAtRange`, which `vintasend-ts` has no equivalent for. Since a missing key
reads as supported, a client written against either map is safe in both directions.

`stringLookups.caseSensitive` and `stringLookups.caseInsensitive` are **two independent
capabilities, not one flag and its negation.** Don't derive either from the other:

- A backend on a case-insensitive collation (MySQL's `*_ci`) reports
  `caseSensitive: False` — it matches case-insensitively and cannot do anything else.
- A backend with no way to fold case reports `caseInsensitive: False` — the opposite
  constraint.

Reading one as the inverse of the other inverts the answer for exactly the backends that
had a constraint worth reporting, and the failure is silent: you decline the one lookup
the backend actually supports.

#### Resending a notification

`resend_notification` is the retry action a dashboard drives when a notification failed or a user
asks for it again:

```python
clone = notification_service.resend_notification(notification_id)

# Reuse the exact context the original notification rendered with, instead of
# regenerating it through the context registry
clone = notification_service.resend_notification(
    notification_id, use_stored_context_if_available=True
)
```

The source notification is left completely untouched -- its id, status and timestamps never
change. The clone is a brand-new row: same user, template, context configuration and attachments,
sent immediately (`send_after` is not carried over, since resending means "send now").
`resend_notification` raises `NotificationResendError` for a one-off notification, and for one
still scheduled in the future -- these are refusals, not silent no-ops.

The object `resend_notification` returns reflects the clone's state right after `send()` is
called on it, the same convention `create_notification` already follows -- treat it the same way.
A backend that returns a detached object from `persist_notification` may hand back something that
still looks `PENDING_SEND` even though the clone was, in fact, sent; re-fetch through
`get_notification` or `filter_notifications` if you need the backend's own freshest copy.

#### Rendering a notification from historical template content

`render_email_template_from_content` reproduces how an email notification rendered in the past,
for a preview or an audit trail -- without touching the notification's currently configured
templates, and without sending or persisting anything:

```python
from vintasend.services.notification_template_renderers.base_templated_email_renderer import (
    EmailTemplateContent,
)

notification = notification_service.get_notification(notification_id)

historical_content = EmailTemplateContent(
    subject_template="Your order {{ order_id }} has shipped",
    body_template="<p>Hi {{ first_name }}, your order is on its way.</p>",
    # preheader_template is optional -- a Python-only concept with no TS counterpart
    preheader_template="Your order is on its way",
)

templated_email = notification_service.render_email_template_from_content(
    notification,
    historical_content,
    # Render with the context the notification actually used, not a freshly generated one
    notification.context_used,
)

templated_email.subject   # "Your order 123 has shipped"
templated_email.body      # "<p>Hi Ada, your order is on its way.</p>"
templated_email.preheader # "Your order is on its way"
```

`AsyncIONotificationService.render_email_template_from_content` is the same signature, awaited.

This is a pure, read-shaped operation: it performs no I/O, generates no context (the context you
pass is used verbatim -- typically a notification's stored `context_used`), and never sends or
writes anything. It raises `NotificationRenderError` if the notification's type has no email
adapter configured, or if the configured renderer is not a `BaseTemplatedEmailRenderer`.

**Injection-safety caveat.** `template_content` is rendered as-is through the configured
renderer -- an inline template string is an injection surface if it comes from user input.
Only pass template content sourced from your own application's template history (for example, a
version you stored alongside the notification, or a git-tracked template file at a known past
revision), never a string a user typed in. Jinja2-backed renderers still apply their normal
autoescape defaults on top of this, but that's a mitigation, not a substitute for keeping the
template source trusted.

#### `tenant` is a filter field, not an access control

`tenant` behaves like `user_id` elsewhere in this library: it's an opaque string this package
doesn't interpret. The host application decides what a tenant is and who may see it.
**Filtering by `tenant` is not authorization.** Nothing stops a caller from omitting the `tenant`
filter and reading across every tenant, and nothing in this library checks whether the caller is
allowed to see the tenant it asked for. `update_notification` refuses to change a notification's
`tenant` after creation, raising `TenantReassignmentError` -- that closes off one accidental way to
move a notification between tenants, but it's defence in depth, not a security boundary. The host
application must still enforce who is allowed to query which tenant.

#### Query governance is the host application's job

Because the filter is composable, a caller can build an arbitrarily deep `and` / `or` / `not`
tree. A deeply nested `or` over an unindexed column will produce a slow scan, and this library has
no query planner to stop it. There's no built-in depth limit or leaf-filter cap -- an arbitrary
number here would be arbitrary for every host, not a real safeguard. If a filter comes from
untrusted input, the host application should validate or bound it (a maximum nesting depth, a
maximum number of leaf filters, indexes on the columns it lets clients filter by) before handing
it to `filter_notifications`.


## Background Sending via Queue Service

Background sending lets you decouple notification delivery from the web request. Adapters opt in by subclassing `BackgroundNotificationAdapter` (sync) or `AsyncIOBackgroundNotificationAdapter` (AsyncIO). When you send a notification through a background adapter, the service enqueues only the notification id and returns immediately — the worker processes the queue and actually delivers the notification.

### Configuration

Set `NOTIFICATION_QUEUE_SERVICE` to your queue service implementation:

```python
from vintasend.services.notification_service import NotificationService
from myapp.queue_services import MyQueueService

service = NotificationService(
    notification_adapters=[...],
    notification_backend=...,
    notification_queue_service=MyQueueService(),  # or as an import string
)
```

Or via environment variable:

```bash
export NOTIFICATION_QUEUE_SERVICE="myapp.queue_services.MyQueueService"
```

### Queue Service Factory

The worker needs `NOTIFICATION_SERVICE_FACTORY` to rebuild the service in its own process, since the queue payload carries only the notification id. The factory is called once per worker process and cached:

```python
# myapp/worker.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from vintasend.services.notification_service import NotificationService
from vintasend_sqlalchemy.backends import SQLAlchemyNotificationBackend

_engine = create_engine("postgresql://...")
_SessionLocal = sessionmaker(bind=_engine)

def notification_service_factory():
    """Return a notification service for this worker process."""
    session = _SessionLocal()
    backend = SQLAlchemyNotificationBackend(session)
    return NotificationService(
        notification_backend=backend,
        notification_adapters=[...],
    )
```

Then point the environment variable:

```bash
export NOTIFICATION_SERVICE_FACTORY="myapp.worker.notification_service_factory"
```

The worker and web process must read the same `NOTIFICATION_BACKEND` and `NOTIFICATION_QUEUE_SERVICE` settings.

### Worker Entrypoints

Register `send_notification` or `async_send_notification` as your queue task:

```python
# Sync task (Celery example)
from celery import Celery
from vintasend.tasks.background_tasks import send_notification

celery_app = Celery()
celery_app.task(send_notification)

# AsyncIO task
from vintasend.tasks.background_tasks import async_send_notification
celery_app.task(async_send_notification)
```

Both entrypoints take the notification id and an optional explicit service override.

### Example: Background Email

```python
from vintasend.services.notification_adapters.async_base import BackgroundNotificationAdapter
from vintasend.services.notification_template_renderers.base import BaseNotificationTemplateRenderer

class CeleryEmailAdapter(BackgroundNotificationAdapter):
    """Email adapter that uses Celery for delivery."""
    
    def send(self, notification, context):
        # Core calls this from the worker after the queue service enqueues
        email_content = self.template_renderer.render(notification, context)
        # Send email (attachments are loaded from backend)
        self._send_email(notification.email_or_phone, email_content, notification.attachments)

    def delayed_send(self, notification_id):
        # Marker only -- core never calls this. Delivery happens in send() above.
        pass
```

### Attachments in Background Sends

File attachments now work on the background path because the worker reloads the real notification from the backend, not from a serialized payload.

## Multi-Backend Configuration

A service can hold one primary backend plus zero or more **additional backends** -- extra
copies of the same data, kept in sync by replication. This is for hosts that want a hot standby,
a region-local read replica, or a bridge while migrating from one storage backend to another. A
service configured with no `additional_backends` behaves exactly like a single-backend
deployment; every read, write, and setting below is inert until you pass at least one.

### Configuring additional backends

Pass extra backend instances (or import strings) through `additional_backends`:

```python
from vintasend.services.notification_service import NotificationService

service = NotificationService(
    notification_adapters=[...],
    notification_backend=primary_backend,
    additional_backends=[replica_backend, another_replica_backend],
)
```

Each configured backend -- primary and additional -- gets a stable string identifier used to
address it later. A backend can declare its own identifier by implementing
`get_backend_identifier() -> str | None`; if it returns `None` (the concrete default every
backend gets for free), the service assigns `backend-{n}`, where `n` is the backend's position
(the primary is always `backend-0`; the first additional backend is `backend-1`, and so on).
Two configured backends that resolve to the same identifier raise
`DuplicateBackendIdentifierError` at construction -- fix a colliding custom
`get_backend_identifier()` or drop the duplicate entry.

```python
service.get_primary_backend_identifier()      # "backend-0" (or the primary's own identifier)
service.get_all_backend_identifiers()          # every identifier, primary first, in order
service.get_additional_backend_identifiers()   # every identifier except the primary's
service.has_backend("backend-1")               # True / False
```

### Read routing

Every read method (`get_notification`, `filter_notifications`, `get_future_notifications`, and
the rest of the getters) takes an optional trailing `backend_identifier` keyword:

```python
# Reads the primary (the default -- identical to a single-backend service)
service.get_notification(notification_id)

# Reads a specific replica directly
service.get_notification(notification_id, backend_identifier="backend-1")

# Unknown identifier -- raises BackendNotFoundError
service.get_notification(notification_id, backend_identifier="does-not-exist")
```

There is no fan-out read that queries every backend and merges the result -- a read always
targets exactly one backend. Comparing backends is what `verify_notification_sync` (below) is
for.

### Inline vs. queued replication

With additional backends configured, every write -- create, update, mark sent/failed/read,
cancel, store context, and so on -- replicates to them after the primary write succeeds.
`replication_mode` controls how:

* **`"inline"` (the default).** Replication happens on the request path, right after the
  primary write returns. No extra infrastructure needed; the right choice unless the added
  request latency of replicating to every backend matters.
* **`"queued"`.** The write enqueues one replication task per destination backend and returns
  immediately; a background worker (the same host-factory worker model background sending
  uses) drains the queue and applies the replication. Pick this once you already run that
  worker and want replica writes off the request path -- at the cost of replicas converging
  slightly later and of one task per additional backend per write.

```python
service = NotificationService(
    notification_adapters=[...],
    notification_backend=primary_backend,
    additional_backends=[replica_backend],
    replication_mode="queued",
    replication_queue_service=MyReplicationQueueService(),  # or an import string
)
```

Or via environment variables:

```bash
export NOTIFICATION_REPLICATION_MODE="queued"
export NOTIFICATION_REPLICATION_QUEUE_SERVICE="myapp.queue_services.MyReplicationQueueService"
```

`NOTIFICATION_REPLICATION_MODE` defaults to `"inline"`; `NOTIFICATION_REPLICATION_QUEUE_SERVICE` defaults to unset. As with every setting, an environment variable overrides the framework config value, and an instance or import string passed to the constructor overrides both.

A replication queue service implements `BaseNotificationReplicationQueueService` (sync) or `AsyncIOBaseNotificationReplicationQueueService` (AsyncIO) -- a single method, `enqueue_replication(notification_id, backend_identifier)`. A worker drains the queue by calling `service.process_replication(notification_id, backend_identifier)`, which converges that replica to the primary's current snapshot. You can also call `register_replication_queue_service(...)` to inject the queue service after construction, the same way `register_queue_service` works for background sending.

Key semantics:

* **A broken queue never silently drops replication.** If no replication queue service is configured, or the record's id cannot be resolved, queued mode logs a warning and falls back to inline replication for every backend. If enqueueing a single backend raises, only that backend is replicated inline; the ones that enqueued successfully are left to the worker.
* **Single-backend services never replicate**, regardless of `replication_mode`.
* **The worker and web process must share settings** -- the same `NOTIFICATION_BACKEND`, `NOTIFICATION_REPLICATION_QUEUE_SERVICE`, and `NOTIFICATION_SERVICE_FACTORY`, exactly as background sending requires.

### Monitoring and repair

Four methods give a dashboard everything it needs to watch replication and fix drift:

```python
# Per-field, per-backend agreement report for one notification
report = service.verify_notification_sync(notification_id)
report["in_sync"]                  # True only if every backend holds the record and every
                                    # comparable field agrees across all of them
report["backends_with_record"]     # identifiers that have the record at all
report["backends_missing_record"]  # identifiers that don't
report["fields"]                   # per-field {"field", "in_agreement", "differing_values"}

# Per-backend health: total notification count, or the error if a backend is unreachable
stats = service.get_backend_sync_stats()
stats["backend-1"]  # {"total_notifications": 1234, "status": "healthy"}

# Converge one notification to every additional backend, or a single named one
service.replicate_notification(notification_id)
service.process_replication(notification_id, target_backend_identifier="backend-1")

# Copy every notification from one backend to another, batch_size rows at a time
result = service.migrate_to_backend(
    destination_backend_identifier="backend-2",
    batch_size=500,
    source_backend_identifier=None,  # defaults to the primary
)
result["migrated"]   # count of records confirmed present on the destination
result["failures"]   # [{"notification_id": ..., "error": ...}, ...] -- doesn't abort the batch
```

`migrate_to_backend` is copy-only -- it never deletes from the source, and re-running it is
idempotent (an already-migrated record is converged, never duplicated). Retiring the source is
a separate, deliberate step you take after verifying the destination, not something this method
does for you.

### Failure semantics

Multi-backend replication is **not** synchronous multi-master, and it makes no
read-your-writes guarantee across backends. Be precise about what it actually promises:

* **The primary write commits first, and its failure is the caller's failure.** If the primary
  rejects a write, the exception propagates exactly as it would on a single-backend service.
* **Replica writes are best-effort.** A replica that rejects a write is logged, not raised --
  the primary write already succeeded and is never rolled back. A rejected replica write is
  reconciled by the next write to that record, or by calling `process_replication` /
  `verify_notification_sync` yourself.
* **No automatic failover.** A down primary is an error, full stop -- this library never
  promotes a replica to primary for you. Promoting a replica is an operational decision a human
  or an external tool makes, not something `NotificationService` does automatically.
* **Reading right after writing can see a stale replica.** A read against `backend_identifier="backend-1"` immediately after a write may not yet reflect that write, especially in `"queued"` mode. Read the primary (the default, no `backend_identifier`) when you need the current state.

## Git Commit SHA Tracking

Every notification can record which source-code revision rendered and sent it. This is opt-in: with no provider configured, nothing changes -- no SHA is ever resolved or written, and `git_commit_sha` stays `None` on every notification.

To turn it on, implement `BaseGitCommitShaProvider` (sync) or `AsyncIOBaseGitCommitShaProvider` (AsyncIO) -- a single method, `get_current_git_commit_sha() -> str | None`, that returns the commit SHA of the revision currently running:

```python
from vintasend.services.git_commit_sha_providers.base import BaseGitCommitShaProvider

class EnvVarGitCommitShaProvider(BaseGitCommitShaProvider):
    def get_current_git_commit_sha(self) -> str | None:
        return os.environ.get("GIT_COMMIT_SHA")
```

Then configure it on the service, the same way as the queue service or attachment manager -- an instance, a dotted import string, or the `NOTIFICATION_GIT_COMMIT_SHA_PROVIDER` setting:

```python
service = NotificationService(
    notification_adapters=[...],
    notification_backend=...,
    git_commit_sha_provider=EnvVarGitCommitShaProvider(),  # or as an import string
)
```

```bash
export NOTIFICATION_GIT_COMMIT_SHA_PROVIDER="myapp.git.EnvVarGitCommitShaProvider"
```

Key semantics:

* **Resolved at send time, not creation time.** The provider is called at the top of both `send()` and `delayed_send()`, so a scheduled notification records the revision that actually delivered it -- foreground or from a background worker -- not the one that enqueued it.
* **One SHA per notification**, not a history: `git_commit_sha` holds the revision that last sent it.
* **Written only when it changes.** The provider is called on every send, but the backend is only asked to persist a new value when it differs from what is already stored, so re-sending under the same revision is a no-op write.
* **Normalized and validated.** A returned SHA is trimmed, lowercased, and must match 40 hexadecimal characters; a malformed non-`None` value raises `InvalidGitCommitShaError` rather than being silently stored. A `None` return means "unknown this call" and is simply skipped.
* **Provider failures never block a send.** If the provider raises, the exception is caught and logged, and treated exactly like a `None` return -- audit metadata is never worth failing a delivery over.
* **System-managed.** `git_commit_sha` is never settable through `create_notification` or `update_notification` -- passing it to `update_notification` raises `GitCommitShaReassignmentError`. It is only ever written by the service itself, at send time.

## Template Version Pinning

Some template renderers version their templates. A file on disk has no version -- editing it changes what every notification renders, past and future. A store-backed renderer such as [vintasend-managed-templates](https://github.com/vintasoftware/vintasend-managed-templates) keeps every edit as a new version, and that makes two questions answerable that were not before: *which version should this notification render*, and *which version did it actually go out with*.

Two fields on `Notification` (and `OneOffNotification`) hold the answers:

| Field | Set by | Means |
|---|---|---|
| `requested_template_version` | the caller, or the service at create time | Render this exact version. `None` means "whatever is current at send time". |
| `used_template_version` | the service, at send time | The version the renderer reported it actually used. Read-only to callers. |

With a renderer whose templates are not versioned -- which is most of them -- both stay `None` and nothing about this changes.

### Pinning a single notification

Pass the version and it is used, whatever the service is configured with:

```python
service.create_notification(
    user_id=user.id,
    notification_type="EMAIL",
    title="Welcome",
    body_template="welcome",          # a template key, not a path
    context_name="welcome_context",
    context_kwargs={"user_id": user.id},
    requested_template_version=3,     # render v3, now and forever
)
```

Repointing an existing notification works the same way through `update_notification`:

```python
service.update_notification(notification.id, requested_template_version=4)
```

### Pinning without naming a version

`pin_template_versions=True` resolves whatever version is current *now* and records it, so a later edit to the template cannot change what an already-recorded notification renders. Ask for it per call:

```python
service.create_notification(
    ...,
    body_template="welcome",
    pin_template_versions=True,      # pin to whatever "welcome" is right now
)

service.update_notification(
    notification.id,
    pin_template_versions=True,
    body_template="farewell",        # re-pinned to farewell's current version
)
```

...or set the default for every call on the service, and override it where it does not apply:

```python
service = NotificationService(
    notification_adapters=[...],
    notification_backend=...,
    pin_template_versions=True,      # the default for every create and update
)

service.create_notification(..., pin_template_versions=False)   # not this one
```

The default is **off**, because turning it on changes what an existing deployment sends: unpinned, a notification scheduled for next week renders whatever the template says next week, which is sometimes exactly what a team wants. The flag is never stored -- it decides what `requested_template_version` is set to at that moment, and nothing afterwards consults it.

Key semantics:

* **The call wins over the service.** `pin_template_versions` on a create or update overrides the constructor's setting in both directions; leave it `None` (the default) to defer to the service.
* **An explicit version always wins over both.** `pin_template_versions` only decides what happens when the caller does not name a version, on create and on update alike.
* **Resolved through the renderer.** The service asks the renderer for the notification type it is creating, via `BaseNotificationTemplateRenderer.get_latest_template_version()`. The default implementation returns `None` -- so with a file-based renderer this flag has nothing to pin and quietly does nothing.
* **Best-effort.** A renderer that raises while resolving is logged and the notification is created unpinned, rather than failing the write. An unpinned notification still sends.
* **Updates re-pin only when the template changes.** An update carrying a new `body_template` is re-pinned to that template's current version; an update to the title leaves an existing pin exactly as it was. Silently moving a pin forward is the one thing pinning exists to prevent.

### Recording what actually rendered

At send time the renderer reports the version it used on the send input it returns, the adapter hands that back from `send()`, and the service stores it:

```python
notification = service.get_notification(notification_id)
notification.requested_template_version   # 3  -- what it asked for (or None)
notification.used_template_version        # 3  -- what rendered
```

On an unpinned notification `used_template_version` is the *only* record of which version went out, since the template has moved on by the time anyone asks.

* **System-managed.** Passing `used_template_version` to `update_notification` raises `UsedTemplateVersionReassignmentError`. Set `requested_template_version` instead.
* **Never blocks a send.** It is written after delivery, so a failure to record it is logged and the notification stays sent.
* **A pin covers the template named, not what that template builds on.** With a renderer that
  composes templates from other templates, pinning a notification to v3 of `welcome` renders v3 of
  `welcome` -- but a base it extends without naming a version still resolves to whatever that base
  is today. Pin the reference too (`{% managed_extends "base-email" version=2 %}`) when a template
  must keep composing against an exact parent. See
  [Template Managers](#template-managers).

### What an implementation package has to do

Everything below is optional. A renderer, adapter, or backend that ignores all of it keeps working exactly as it did.

* **A renderer whose templates are versioned** overrides `get_latest_template_version(template_key)`, honours the notification's `requested_template_version` in `render()`, and sets `template_version` on the `NotificationSendInput` it returns.
* **An adapter** returns the send input from `send()` instead of discarding it. Returning `None` is still valid -- the service then records nothing.
* **A backend** stores `requested_template_version` (an optional keyword on `persist_notification` / `persist_one_off_notification`, passed only when a version was pinned) and overrides `store_template_version()`. That method is concrete and a no-op on the base, so a backend with nowhere to put it needs no changes at all.

## Glossary

* **Notification Backend**: It is a class that implements the methods necessary for VintaSend services to create, update, and retrieve Notifications from da database.
* **Notification Adapter**: It is a class that implements the methods necessary for VintaSend services to send Notifications through email, SMS or even push/in-app notifications.
* **Template Renderer**: It is a class that implements the methods necessary for VintaSend adapter to render the notification body.
* **Notification Context**: It's the data passed to the templates to render the notification correctly. It's generated when the notification is sent, not on creation time
* **Context generator**: It's a function registered with a name that, when called, generates the data necessary to render a notification.
* **Context name**: The registered name of a context generator. It's stored in the notification so the context generator is called at the moment the notification will be sent.
* **Context registry**: We store all registered context generators on a Singleton class, we call it context registry.
* **Notification Attachment**: Files that can be attached to notifications, supporting various input types including file paths, URLs, bytes data, and file-like objects.
* **Attachment Manager**: It is a class that implements the methods necessary to store, read, and delete the bytes behind a notification attachment, so the notification backend only ever handles rows, never files.
* **Template Manager**: A class that implements the methods necessary to store, version, and publish notification templates in a data store rather than as files, so they can be edited without a deploy. Paired with a template renderer that reads from it.
* **Managed Template**: One version of one template held by a template manager, identified by a key and a version number, with a draft/active/inactive/archived lifecycle. A notification's `body_template` names its key rather than a file path.
* **Abstract Template**: A managed template meant to be built on rather than sent -- it declares a `{% managed_children %}` hole, or blocks without extending anything. Recorded on the template as `is_abstract`.
* **One-off Notification**: A notification sent directly to an email address or phone number without requiring a user ID from your database. 
* **Git Commit SHA Provider**: A class that implements a single method returning the current git commit SHA, so a notification records which source-code revision sent it.


## Community

VintaSend has many backend, adapter, and template renderer implementations. If you can't find something that fulfills your needs, the package has very clear interfaces you can implement and achieve the exact behavior you expect without loosing VintaSend's friendly API.

### Officially supported packages 

#### Backends

* **[vintasend-django](https://github.com/vintasoftware/vintasend-django/)**: Uses Django ORM to manage the notifications in the database.
* **[vintasend-sqlalchemy](https://github.com/vintasoftware/vintasend-sqlalchemy/)**: Uses SQLAlchemy to manage the notifications in the database. It supports both sync and async engines/sessions.

#### Adapters

* **[vintasend-fastapi-mail](https://github.com/vintasoftware/vintasend-fastapi-mail/)**: AsyncIO implementation that sends emails using FastAPI-Mail
* **[vintasend-flask-mail](https://github.com/vintasoftware/vintasend-flask-mail/)**: Sync implementation that sends emails using Flask-Mail.
* **[vintasend-celery](https://github.com/vintasoftware/vintasend-celery/)**: Adapter factory that allows sending emails asynchronously on sync backends by using Celery.
* **[vintasend-django](https://github.com/vintasoftware/vintasend-django/)**: Sync implementation that sends emails using Django's builtin email sender.

#### Template Renderers
* **[vintasend-django](https://github.com/vintasoftware/vintasend-django/)**: Renders emails using Django's templating system.
* **[vintasend-jinja](https://github.com/vintasoftware/vintasend-jinja/)**: Renders emails using Jinja2.

#### Attachment Managers
* **[vintasend-aws-s3-attachments](https://github.com/vintasoftware/vintasend-aws-s3-attachments/)**: Stores attachment files as objects in an AWS S3 bucket using boto3. Supports both sync and AsyncIO.

#### Template Managers

Regular template renderers read templates from wherever the rendering engine looks -- files on
disk, usually. A **template manager** puts them in a data store instead, so non-technical people
can edit a notification's copy without a deploy, with versioning and a draft/active/inactive/archived
lifecycle on top.

* **[vintasend-managed-templates](https://github.com/vintasoftware/vintasend-managed-templates/)**: The template management layer itself -- the `BaseTemplateManagerBackend` seam, `ManagedTemplateService` for creating/versioning/publishing templates, and `ManagedTemplateEmailRenderer` / `ManagedTemplateSMSRenderer`, which wrap any existing `BaseNotificationTemplateRenderer` and feed it the stored template body instead of a template path. Storage-agnostic on its own; pair it with a manager backend.
* **[vintasend-django-templates-manager](https://github.com/vintasoftware/vintasend-django-templates-manager/)**: Stores managed templates in the database through the Django ORM, with `ManagedTemplate` / `ManagedTemplateStatusRecord` models, an admin, and filtering + pagination over template versions and their status history.
* **[vintasend-templates-management-api](https://github.com/vintasoftware/vintasend-templates-management-api/)**: A django-ninja REST API over a template manager backend, for a UI where non-technical people create, version, publish and preview templates. Tracked here as a submodule under `tools/`.

Because the renderer wraps another renderer rather than replacing it, a notification's
`body_template` stops being a path and becomes a managed template's key -- everything else about
how you create and send notifications stays the same.

**Templates can be built from other templates.** Moving templates into a store costs you what
the rendering engine gave you for free: Django's `{% extends %}` and `{% include %}` resolve
against the template loader, which never sees a stored template. `vintasend-managed-templates`
replaces that with its own layer -- `managed_extends`, `managed_children`, `managed_block` /
`managed_endblock`, `managed_super` and `managed_include` -- resolved into one flat string
*before* the engine runs, so Django or Jinja renders exactly what it always did:

```
base-email    <html><body>
                {% managed_block header %}<h1>Acme</h1>{% managed_endblock %}
                {% managed_children %}
              </body></html>

welcome       {% managed_extends "base-email" %}
              <p>Hi {{ name }}, welcome aboard.</p>
```

A template that declares a `{% managed_children %}` hole, or blocks without extending anything,
is a base to build on rather than one to send. That is what `is_abstract` records -- denormalized
onto the template by the manager backend, so "list the templates I can actually send" is a column
lookup rather than a parse of every row. A reference can name a version --
`{% managed_extends "base-email" version=2 %}` -- which is the same pin
[Template Version Pinning](#template-version-pinning) applies to a notification.

See the [package's README](https://github.com/vintasoftware/vintasend-managed-templates#composition-bases-blocks-and-includes)
for the full tag reference, the cycle and depth limits, and how a composition failure is reported.

#### Working on them from this repo

Each officially supported package lives in its own repository and is linked here as a git
submodule -- the backends, adapters, renderers, attachment managers and template managers under
`implementations/`, and the applications built on top of them (the notifications dashboard and
its API, the templates management API) under `tools/`. A single checkout gives you the core
package plus everything that has to stay compatible with it.

```bash
# Fresh clone, with the implementations
git clone --recurse-submodules git@github.com:vintasoftware/vintasend.git

# Existing clone
git submodule update --init --recursive

# Pull the latest commit of every implementation
git submodule update --remote
```

Each submodule is an ordinary checkout of its own repo: `cd` into it, branch, commit, and
push there as usual. This repo's lint, type-check, and test commands deliberately skip
`implementations/` — each package has its own dependencies, its own tooling config, and
its own CI. Committing here only records which commit of each implementation this repo
points at.

#### Creating a new implementation

If none of the packages above cover what you need, don't start from a blank directory.
`templates/vintasend-implementation-template/` is a ready-to-clone skeleton with one `TODO`
stub per seam (backend, adapter, template renderer, queue service, attachment manager) and a
matching test for each, so your clone installs and passes its test suite before you write any
real logic:

```bash
python templates/vintasend-implementation-template/scripts/clone.py /path/to/vintasend-your-integration --package-name vintasend-your-integration
```

See that package's `README.md` for the full clone-and-rename workflow and a per-component
checklist of exactly which methods to implement.


## Advanced Usage

### AsyncIO Notification Service

To use the AsyncIO Notification Service your backend and adapters must all support AsyncIO as well.

```python
from vintasend.services.notification_service import AsyncIONotificationService, register_context
from vintasend.services.dataclasses import NotificationAttachment
from vintasend.constants import NotificationTypes


notification_backend = MyAsyncIONotificationBackend() 
notifications_service = AsyncIONotificationService(
    notification_adapters=[MyAsyncIOAdapter(MyTemplateRenderer(), notification_backend)], 
    notification_backend=notification_backend,
)


@register_context("my_context_generator")
async def my_context_generator(user_id: str):
    user = await get_user_by_id(user_id)
    
    return {
        "first_name": user.first_name,
        "last_name": user.last_name,
        "joined_at": user.joined_at,
    }


await notifications_service.create_notification(
    user_id=user.id,
    notification_type=NotificationTypes.EMAIL,
    title="My Notification", # this is just for auditing purposes
    body_template="my/notification/template/path.html",
    context_name="my_context_generator",
    context_kwargs={
        "user_id": user.id,
    },
    send_after=datetime.datetime.now(),
    subject_template="my/notification/subject/template/path.txt",
    preheader_template="my/notification/preheader/template/path.html",
)
```

**Note**: Attachments work the same way with `AsyncIONotificationService` - just pass the `attachments` parameter with a list of `NotificationAttachment` objects.

**Background sending** also works with `AsyncIONotificationService`. Subclass `AsyncIOBackgroundNotificationAdapter` and pass an `AsyncIOBaseNotificationQueueService` to send via a queue:

```python
from vintasend.services.notification_service import AsyncIONotificationService
from vintasend.services.notification_adapters.asyncio_background_base import AsyncIOBackgroundNotificationAdapter

class MyAsyncIOBackgroundAdapter(AsyncIOBackgroundNotificationAdapter):
    async def send(self, notification, context):
        # Worker calls this to deliver
        email_content = self.template_renderer.render(notification, context)
        await self._send_email(notification, email_content)

    async def delayed_send(self, notification_id):
        # Marker only -- core never calls this. Delivery happens in send() above.
        pass

notifications_service = AsyncIONotificationService(
    notification_adapters=[MyAsyncIOBackgroundAdapter(...)],
    notification_backend=...,
    notification_queue_service=...,  # AsyncIO queue service
)
```

### Using frameworks that don't use a globally-available configuration

Frameworks like FastAPI don't have a centralized configuration object that's globally accessible by default. Because of that, we need to manually pass the configuration object to the service in order to initialize VintaSend configuration.


```python
from vintasend.services.notification_service import AsyncIONotificationService, register_context
from vintasend.services.dataclasses import NotificationAttachment
from vintasend.constants import NotificationTypes
from fastapi import FastAPI
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str
    items_per_user: int = 50

settings = Settings()
app = FastAPI()

notification_backend = MyAsyncIONotificationBackend() 
notifications_service = AsyncIONotificationService(
    notification_adapters=[MyAsyncIOAdapter(MyTemplateRenderer(), notification_backend)], 
    notification_backend=notification_backend,
    config=settings,
)
```

