Write a driver¶

Goal: a package that adds a backend to a capability, selected by configuration, with no change to business modules or to the kernel.

Use it for a notification backend that Sandloom does not ship, such as an email API, or for a capability of your own.

Do not use it for business logic: that is a module. A contract is public only with a conformance kit, so the cache, objectstore and taskqueue contracts are not available yet. The database and eventbus backends carry the transaction and delivery guarantees and are not open to drivers (why).

Prerequisites¶

• An application created with sandloom new.
• A separate package for the driver, outside <app>/modules. A driver may import its backend's SDK and sandloom.driverkit; a module may not (SANDLOOM-C004).

Steps¶

1. Implement the contract¶

This complete driver is for development: it appends messages, including their bodies, to a file.

"""A development notification driver: appends every message to a JSON-lines file."""

import asyncio
import json
import os
from collections.abc import Mapping
from pathlib import Path
from typing import Any

from sandloom.driverkit import Notification, NotifierProvider


class FileNotifier(NotifierProvider):
    contract_version = "v1"
    supported_features = frozenset({"email"})

    def __init__(self, *, name: str, driver: str, options: Mapping[str, Any]) -> None:
        super().__init__(name=name, driver=driver, options=options)
        path = options.get("path")
        if not path:
            raise ValueError("the file notification driver needs a 'path' option")
        self._path = Path(path)

    async def send(self, notification: Notification) -> None:
        line = json.dumps(
            {
                "channel": notification.channel,
                "recipient": notification.recipient,
                "subject": notification.subject,
                "body": notification.body,
            }
        )
        await asyncio.to_thread(self._append, line)

    def _append(self, line: str) -> None:
        # Bodies can hold secret links: only the owner may read the file.
        descriptor = os.open(self._path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
        os.fchmod(descriptor, 0o600)
        with os.fdopen(descriptor, "a", encoding="utf-8") as target:
            target.write(line + "\n")

• contract_version is the contract the driver was written for. It is required, and another major version is refused when the driver loads.
• supported_features names what the driver really does, and is its one declaration. A module that requires features={"email"} is refused at startup on a driver that does not advertise it. If a feature depends on an option, override the classmethod features_for(options). Do not override features(): an instance that reports other features than it declared is refused.
• options holds the keys of the INFRA entry, and nothing from another driver. Options can hold secrets, so never log them. For validation before the driver is built, set options_schema to a pydantic model.
• The constructor is called on first use, or at startup if an instance is configured. Open no connection at import. To open a pool at startup, report health in /diagnostics/infra or release resources, override aopen, health and aclose.
• Notification.body_fields holds the values body was built from, for a driver that renders its own template. It is as sensitive as the body: never log it.

2. Publish it by name¶

The entry-point name is <capability>.<driver>, and its value is your class:

[project]
name = "sandloom-notify-file"
version = "0.1.0"
dependencies = ["sandloom>=0.7.0a6,<0.8"]

[project.entry-points."sandloom.drivers"]
"notification.file" = "sandloom_notify_file.driver:FileNotifier"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

Add the package to the application's dependencies. For a package next to the application:

uv add --editable ../sandloom-notify-file

Sandloom imports the class only when INFRA selects its name. A package nothing selects runs no code, and the order of installed packages does not matter.

3. Select it with INFRA¶

Put this in .env for the application and in .env.test for its tests. Everything other than driver becomes an option:

INFRA={"notification":{"default":{"driver":"file","path":"/tmp/myapp-mail.log"}}}

Module code does not change: await notifier().send(...) now reaches your driver. To keep the default and add a second instance, use another name ("audit" instead of "default") and resolve it with notifier("audit").

4. Check it against the contract¶

sandloom.driverkit.testing holds the conformance kit. Subclass it in the driver's tests; each obligation is its own test, so a failure names what to fix. The kit needs pytest and pytest-asyncio, and it sends real messages, so point it at a test backend.

"""The conformance kit for the file notification driver."""

from collections.abc import Callable
from pathlib import Path

import pytest
from sandloom_notify_file.driver import FileNotifier

from sandloom.driverkit import NotifierProvider
from sandloom.driverkit.testing import NotifierConformance


class TestFileNotifier(NotifierConformance):
    @pytest.fixture
    def make_provider(self, tmp_path: Path) -> Callable[[], NotifierProvider]:
        options = {"path": str(tmp_path / "mail.log")}
        return lambda: FileNotifier(name="default", driver="file", options=options)

A skipped test means the driver advertises no channel, so nothing was sent. Read the skip reason rather than treating it as a pass.

5. Prove the application uses it¶

A passing application test does not show the driver was selected: the default log driver also accepts every send. Assert on the effect. With INFRA in .env.test, a module test can check it:

from pathlib import Path

from sandloom.sdk.infra import Notification, notifier  # sandloom: advanced infra


async def test_mail_goes_through_the_file_driver(client) -> None:
    await notifier().send(
        Notification(channel="email", recipient="a@b.test", subject="Hi", body="x")
    )
    assert "a@b.test" in Path("/tmp/myapp-mail.log").read_text()

Verify¶

uv run sandloom test

On a running application, /diagnostics/infra lists the installed driver packages and the driver of each instance. It needs INTERNAL_API_TOKEN (Operator reference).

Errors you may hit¶

Related¶

• Infrastructure: interfaces, drivers and instances.
• Add your own capability: a new kind of infrastructure.
• sandloom.driverkit: every name it exports.
• Customize a standard module: what a notification driver can do for identity's mail.
• ADR-0011: why it works this way.