Transactions¶

A transaction is a regular function decorated with @transaction (or @async_transaction). By default it is atomic: every database operation inside it either commits on success or rolls back if the function raises.

Transactions are also the unit AMSDAL exposes as remote procedures: each one is discovered at startup, can be called over HTTP, run in the background, or scheduled.

Basic Usage¶

SyncAsync

from amsdal.transactions import transaction

@transaction
def create_person(first_name: str, last_name: str) -> Person:
    person = Person(first_name=first_name, last_name=last_name)
    person.save()
    return person

from amsdal.transactions import async_transaction

@async_transaction
async def create_person(first_name: str, last_name: str) -> Person:
    person = Person(first_name=first_name, last_name=last_name)
    await person.asave()
    return person

If the function raises, all database changes made inside it are rolled back automatically.

Defining a Transaction¶

The decorator accepts a few keyword arguments:

@transaction(name="archive_person", tags=("admin", "maintenance"), hidden=True, atomic=True)
def archive_person(person: Person) -> None:
    ...

Argument	Default	Purpose
`name`	function name	The public name used to address the transaction (see Discovery & Addressing).
`atomic`	`True`	When `True`, wrap the whole function in one transaction. When `False`, do not — see below.
`tags`	`()`	Labels used to filter the transaction listing.
`hidden`	`False`	Exclude the transaction from the listing. It can still be called, submitted, and executed — it is only hidden from discovery listings.

atomic¶

By default a transaction is all-or-nothing. Set atomic=False when you deliberately want each operation to persist on its own, so a later failure does not undo earlier work:

@transaction(atomic=False)
def import_people(rows: list[dict]) -> None:
    for row in rows:
        Person(**row).save()   # each save commits independently
    # if a later row fails, the people already saved above remain in the database

With atomic=True (the default), the same failure would roll back every row.

Arguments¶

Every argument of a transaction must have a type annotation. An unannotated argument raises a TypeError at application startup, when transactions are discovered.

Annotating an argument with a model type controls how a caller may pass that model — as a reference to an existing record, or as inline data for a new one.

Bare model — reference or inline¶

A plain model annotation accepts either a reference to an existing record or an inline object. Either way the function body receives a loaded instance:

@transaction
def rename_person(person: Person, new_name: str) -> None:
    person.first_name = new_name   # already loaded — a reference is fetched automatically
    person.save()

`ReferenceOnly[Model]` — reference only, not loaded¶

Use ReferenceOnly[...] when the caller must pass a reference to an existing record and you do not need the record loaded (e.g. you only need its id, or loading it is unnecessary work):

from amsdal.transactions import ReferenceOnly

@transaction
def enqueue_report(person: ReferenceOnly[Person]) -> None:
    person_id = person.ref.object_id   # no database fetch happens
    ...

`NewOnly` — inline only, always new¶

Use NewOnly when the argument must be brand-new data (a reference is rejected). The body receives a fresh, unsaved instance:

from typing import Annotated
from amsdal.transactions import NewOnly

@transaction
def register_person(person: Annotated[Person, NewOnly]) -> Person:
    person.save()   # always a new record
    return person

Collections and optionals¶

Model arguments compose the same way inside containers:

@transaction
def link_people(
    manager: Person,                 # reference or inline
    reports: list[Person],           # a list of references/inline objects
    by_team: dict[str, Person],      # mapping — keys must be scalars (str/int/...)
    mentor: Person | None = None,    # optional
) -> None:
    ...

Discovery & Addressing¶

Transactions are discovered automatically at startup. For them to be registered they must live under the transactions root of your app — the src/transactions/ directory (or a submodule of it). A transaction defined anywhere else is never registered, so it cannot be called over HTTP or submitted, and resolving it by name raises TransactionNotFoundError.

📁 src
└── 📁 transactions
    ├── 📁 common
    │   └── utils.py
    ├── create_person.py
    └── reports.py

Each transaction is addressed by its name (the name= argument, or the function name). A transaction that ships inside a package (AMSDAL core or a contrib) is addressed by its fully qualified name, app_name.name (for example auth.change_password); the short name also works as long as it is unambiguous.

The following errors (importable from amsdal.errors) surface naming problems:

Error	When
`TransactionNotFoundError`	The name matches no registered transaction.
`AmbiguousTransactionNameError`	A short name matches more than one transaction — use the full `app_name.name`.
`DuplicateTransactionNameError`	Two transactions register the same qualified name (raised at startup).

Invoking a Transaction¶

Direct call¶

Call the decorated function like any other function — it runs inside its transaction:

person = create_person("Ada", "Lovelace")

Over HTTP¶

Every registered transaction is exposed by the AMSDAL server as an RPC endpoint.

Execute — POST /api/transactions/{name}/ with a JSON object of the arguments. The response is the transaction's return value, JSON-encoded.
```
curl -X POST https://<host>/api/transactions/create_person/ \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{"first_name": "Ada", "last_name": "Lovelace"}'
```
Invalid or missing arguments return HTTP 422 with a structured detail list describing each problem. Calling a transaction requires the execute permission on it (see Permissions).
List / describe — GET /api/transactions/ lists the available transactions (hidden ones are excluded), and GET /api/transactions/{name}/ describes a single one, including its arguments.

Permissions¶

By default a transaction requires an authenticated user. You control access by stacking a permission decorator above @transaction (the order matters — the permission must attach to the transaction wrapper):

from amsdal.contrib.auth.decorators import require_auth, allow_any

@require_auth        # only authenticated users
@transaction
def change_password(old_password: str, new_password: str) -> None:
    ...

@allow_any           # no authentication required
@transaction
def public_stats() -> dict:
    ...

Custom permissions¶

For anything beyond "authenticated / anyone", subclass BasePermission and implement has_permission(request, action). Built-in permissions (AllowAny, RequireAuth, RequirePermissions) compose with & and |:

from amsdal.contrib.auth.decorators import permissions
from amsdal.contrib.auth.permissions import BasePermission, RequireAuth

class IsBusinessHours(BasePermission):
    def has_permission(self, request, action) -> bool:
        from datetime import datetime
        return 9 <= datetime.now().hour < 17

@permissions(RequireAuth & IsBusinessHours())
@transaction
def export_report() -> None:
    ...

RequirePermissions checks the caller's scopes, for example RequirePermissions("reports:export").

Overriding a shipped transaction's permissions¶

To change the permissions of a transaction that ships in AMSDAL core or a contrib package — without editing the library — apply the permissions(...) decorator to the imported transaction from your own app. Put this in a module under your transactions/ root so it runs at startup:

# src/transactions/_permission_overrides.py
from amsdal.contrib.auth.decorators import permissions
from amsdal.contrib.auth.permissions import AllowAny
from amsdal.contrib.auth.transactions.mfa_device_transactions import list_mfa_devices

permissions(AllowAny)(list_mfa_devices)   # retune the shipped transaction

Note

This retunes the shared transaction for the whole process (that is the intent). For decisions that depend on request data beyond what a BasePermission can express, register your own authorization event listener instead. See the auth permissions guide for the full permission model.

Background Transactions¶

Run a transaction outside the request/response cycle with .submit():

SyncAsync

@transaction
def create_person_and_notify(first_name: str, last_name: str, email: str) -> Person:
    person = Person(first_name=first_name, last_name=last_name)
    person.save()

    send_email.submit(email)   # runs in the background
    return person

@transaction
def send_email(email: str) -> None:
    ...

@async_transaction
async def create_person_and_notify(first_name: str, last_name: str, email: str) -> Person:
    person = Person(first_name=first_name, last_name=last_name)
    await person.asave()

    await send_email.submit(email)   # runs in the background
    return person

@async_transaction
async def send_email(email: str) -> None:
    ...

A background transaction needs a worker connection in config.yml.

Local development (no broker)¶

For local development you do not need a message broker. Configure the inline connection, which runs each submitted transaction immediately, in-process:

connections:
  - name: local_worker
    backend: amsdal_data.transactions.background.connections.sync_connection.SyncBackgroundTransactionConnection

resources_config:
  worker: local_worker

With this connection .submit() executes the transaction inline — no broker, no amsdal worker run.

Note

The inline connection is for regular (non-async) mode. For async_mode: true, use Celery.

Production (Celery)¶

In production, background transactions run on Celery, which needs a message broker (RabbitMQ, Redis, …).

AMSDAL Cloud

On AMSDAL Cloud the broker and worker are configured automatically — just write and deploy.

SyncAsync

connections:
  - name: celery_worker
    backend: amsdal_data.transactions.background.connections.celery_connection.CeleryConnection
    credentials:
      - broker_url: amqp://guest:guest@localhost:5672/

resources_config:
  worker: celery_worker

async_mode: true

connections:
  - name: celery_worker
    backend: amsdal_data.transactions.background.connections.celery_connection.AsyncCeleryConnection
    credentials:
      - broker_url: amqp://guest:guest@localhost:5672/

resources_config:
  worker: celery_worker

Start the worker to consume tasks:

amsdal worker run

Upgrading

Background task names are now the transaction's qualified name (app_name.name for packaged transactions; the short name for your app's own transactions). Tasks queued or scheduled under an older, bare function name will not match the new name after upgrading.

Scheduled Transactions¶

Schedule a transaction to run periodically. This requires a worker connection that supports scheduling (Celery).

SyncAsync

from datetime import timedelta
from amsdal.transactions import transaction, Crontab, ScheduleConfig

# every 10 minutes (interval in seconds)
@transaction(schedule=60 * 10)
def cleanup_expired() -> None:
    ...

# every midnight — Crontab fields are strings
@transaction(schedule_config=ScheduleConfig(schedule=Crontab(minute="0", hour="0")))
def daily_report() -> None:
    ...

# every 3 days, with keyword arguments (schedule is positional)
@transaction(
    schedule_config=ScheduleConfig(
        schedule=timedelta(days=3),
        kwargs={"scope": "all"},
    )
)
def periodic_task(scope: str) -> None:
    ...

from datetime import timedelta
from amsdal.transactions import async_transaction, Crontab, ScheduleConfig

@async_transaction(schedule=60 * 10)
async def cleanup_expired() -> None:
    ...

@async_transaction(schedule_config=ScheduleConfig(schedule=Crontab(minute="0", hour="0")))
async def daily_report() -> None:
    ...

@async_transaction(
    schedule_config=ScheduleConfig(
        schedule=timedelta(days=3),
        kwargs={"scope": "all"},
    )
)
async def periodic_task(scope: str) -> None:
    ...

Start the worker in scheduler or hybrid mode:

# Scheduler only — enqueues due tasks
amsdal worker run --mode scheduler

# Hybrid — schedules and processes tasks
amsdal worker run --mode hybrid

Rollback¶

AMSDAL records a full history of changes, so you can roll back to a previous state by timestamp or by transaction id. Rolling back creates new versions of the affected objects — it does not delete history.

By timestamp¶

SyncAsync

from amsdal.utils.rollback import rollback_to_timestamp

rollback_to_timestamp(person.get_metadata().updated_at)

from amsdal.utils.rollback import async_rollback_to_timestamp

await async_rollback_to_timestamp((await person.aget_metadata()).updated_at)

By transaction id¶

SyncAsync

from amsdal.utils.rollback import rollback_transaction

rollback_transaction(person.get_metadata().transaction.ref.object_id)

from amsdal.utils.rollback import async_rollback_transaction

await async_rollback_transaction(
    (await person.aget_metadata()).transaction.ref.object_id
)

Limitations¶

You cannot roll back to a point in the middle of a transaction. If two objects were created within the same (atomic) transaction, there is no consistent point between them — an AmsdalTransactionError is raised:

from amsdal_data.transactions.errors import AmsdalTransactionError

@transaction
def create_two() -> None:
    Person(first_name="first", last_name="x").save()
    Person(first_name="second", last_name="y").save()

create_two()

try:
    rollback_to_timestamp(first_person.get_metadata().updated_at)
except AmsdalTransactionError:
    pass  # cannot roll back to a mid-transaction point

Transactions¶

Basic Usage¶

Defining a Transaction¶

atomic¶

Arguments¶

Bare model — reference or inline¶

ReferenceOnly[Model] — reference only, not loaded¶

NewOnly — inline only, always new¶

Collections and optionals¶

Discovery & Addressing¶

Invoking a Transaction¶

Direct call¶

Over HTTP¶

Permissions¶

Custom permissions¶

Overriding a shipped transaction's permissions¶

Background Transactions¶

Local development (no broker)¶

Production (Celery)¶

Scheduled Transactions¶

Rollback¶

By timestamp¶

By transaction id¶

Limitations¶

`ReferenceOnly[Model]` — reference only, not loaded¶

`NewOnly` — inline only, always new¶