# Austial documentation

> Austial is a batteries-included, NestJS-style web framework for Python built on FastAPI: modules, controllers, dependency injection, guards, pipes, interceptors, an ORM, auth, multi-tenancy, security, jobs, storage and a CLI.

---

Source: https://docs.austial.com

# Introduction

Austial is a **batteries-included web framework for Python**, built on top
of [FastAPI](https://fastapi.tiangolo.com/) and inspired by
[NestJS](https://nestjs.com/). It gives Python decorator-driven
modules/controllers/providers, real constructor-based dependency injection,
the full request lifecycle (guards, pipes, interceptors, exception filters,
middleware), a config layer, a database layer, Terminus-style health checks,
a testing module — and a CLI so you can scaffold new projects and generate
artifacts.

FastAPI is a fantastic ASGI toolkit, but it doesn't prescribe an application
architecture. Austial is opinionated on top of FastAPI: modules own
controllers and providers, providers get dependency-injected by type, and
the request pipeline

```
guards -> pipes -> interceptors (wrapping) -> handler -> exception filters
```

is a first-class concept instead of something you hand-roll with
`Depends()` everywhere.

## Why Austial

Routing, guards, and interceptors are all declared with decorators that
read top to bottom, in the order they execute:

```python
from austial import Controller, Get, UseGuards, UseInterceptors
from austial.common.interceptors import TransformInterceptor

from .api_key_guard import ApiKeyGuard

@Controller("cats")
class CatsController:
    @Get(":id")
    @UseGuards(ApiKeyGuard)
    @UseInterceptors(TransformInterceptor())
    async def find_one(self, id: int):
        ...
```

Path params use a familiar `:id` syntax (translated to FastAPI's `{id}`
under the hood), so route strings stay short, and the decorator stack --
guard, interceptor, handler -- reads top to bottom in the order it executes.

## Architecture at a glance

- **Modules** — `@Module(imports=, controllers=, providers=, exports=)` compose the application graph.
- **Controllers** — `@Controller(prefix)` plus `@Get`/`@Post`/`@Put`/`@Patch`/`@Delete`/`@Options`/`@Head` declare routes.
- **Providers** — `@Injectable(scope=Scope.DEFAULT | Scope.TRANSIENT)`; the container resolves `__init__` type hints recursively, singleton by default. Custom providers use provider dicts: `{"provide": TOKEN, "useClass"/"useValue"/"useFactory": ..., "inject": [...]}`.
- **Bootstrap** — `AustialFactory.create(AppModule)` returns an `AustialApplication`; `await app.listen(8000)` starts it.
- **Guards** — `CanActivate` ABC + `ExecutionContext`, applied with `@UseGuards(...)`.
- **Interceptors** — `NestInterceptor` ABC (`intercept(context, call_next)`), applied with `@UseInterceptors(...)`.
- **Pipes** — `PipeTransform` ABC, including the built-in `ValidationPipe`.
- **Exception filters** — `ExceptionFilter` ABC + `@Catch(...)`, with a default `AllExceptionsFilter`.
- **Middleware** — `NestMiddleware` ABC, wired up via `configure(consumer)` on modules.
- **Config** — `ConfigModule.for_root(env_file=".env")` / `ConfigService`.
- **Database** — `austial.orm`, a from-scratch TypeORM-shaped ORM: `@Entity()`, `Repository`, `QueryBuilder`, wired up via `OrmModule.for_root(...)`.
- **Health checks** — `austial.terminus.HealthCheckService` + `MemoryHealthIndicator`, `DatabaseHealthIndicator`.
- **Testing** — `austial.testing.Test.create_testing_module(...).compile()`.
- **CLI** — `austial new <name>` / `austial generate|g module|controller|service|resource <name>`.

## A note on file names

Some JavaScript frameworks name files with dots (`health.controller.ts`)
because Node resolves modules by explicit relative path strings. Python's
`import a.b` instead resolves `a` as a package and `b` as a submodule — a
literal file `health.controller.py` isn't importable via a normal `import`
statement. So Austial uses underscore suffixes instead: `health_controller.py`,
`health_service.py`, `health_module.py`, `health_dto.py`. The directory
layout is one folder per feature under `src/modules/`.

## What's not (yet) implemented

Austial is deliberately opinionated but not a 1:1 port of any single
framework. A few concepts are intentionally out of scope for now, so you
don't go looking for them:

- **Request-scoped providers** (`Scope.REQUEST`) — only `Scope.DEFAULT`
  (singleton) and `Scope.TRANSIENT` exist today.
- **Module `imports`/`exports` visibility enforcement** — every module's
  providers currently register into one flat, application-wide container
  (visibility is documentation-only, not enforced at resolve time).
- **RxJS-style interceptor streams** — `CallHandler.handle()` is a plain
  async continuation, not an Observable.

None of this blocks building a real API with the full request lifecycle --
it's just worth knowing up front.

## Where to go next

- [First Steps](/first-steps)
- [Controllers](/overview/controllers)
- [Providers](/overview/providers)
- [CLI](/cli/overview)

---

Source: https://docs.austial.com/first-steps

# First Steps

In this section you'll install the Austial CLI, scaffold a new project, and
get it running with auto-reload. Austial requires **Python 3.11+** and uses
[uv](https://github.com/astral-sh/uv) for dependency management.

## Installation

**uv**

```bash
uv tool install austial
```

**pipx**

```bash
pipx install austial
```

**pip**

```bash
pip install austial
```

This installs the `austial` console script.

**`austial: command not found`?** The folder these tools install into is not
on your `PATH` yet. This is common when uv comes from Homebrew, and uv only
warns about it the first time you install.

- **uv:** run `uv tool update-shell`, then open a new terminal window.
- **pipx:** run `pipx ensurepath`, then open a new terminal window.

To skip the install and run the CLI once, use `uvx austial new my-app`.

## Scaffolding a new project

```bash
austial new my-app
cd my-app
uv sync
cp .env.example .env
uv run austial serve
```

`austial new` scaffolds a small "hello world" app, running on
`http://localhost:8000` with auto-reload, exposing:

- `GET /` — a root health-check style greeting
- `GET /health` — a public Terminus-style health check
- `GET /health/protected` — the same check, gated behind an `ApiKeyGuard`
- `/docs` — FastAPI's Swagger UI, free from FastAPI since every route is
  registered as a real `APIRouter`

## Project structure

```
my-app/
├── pyproject.toml          # dependencies=["austial>=..."], dev extras, ruff/mypy config
├── .env.example             # PORT=8000 / API_KEY=changeme
├── AGENTS.md                 # rules for AI coding tools (CLAUDE.md, GEMINI.md, ... are symlinks to it)
├── .pre-commit-config.yaml   # ruff + mypy + pytest hooks, same as this framework's own repo
├── src/
│   ├── main.py                # AustialFactory.create(AppModule) -> app.listen(8000)
│   ├── app_module.py            # root @Module(...): imports, controllers, providers, configure()
│   ├── app_controller.py         # GET / -> {"message": "Hello from my-app, built with Austial!"}
│   ├── app_service.py
│   └── modules/
│       └── health/
│           ├── health_module.py
│           ├── health_controller.py     # GET /health, GET /health/protected
│           ├── health_service.py         # HealthCheckService + MemoryHealthIndicator
│           ├── health_dto.py
│           └── guards/api_key_guard.py    # ApiKeyGuard(CanActivate)
└── tests/
    ├── unit/health_service_spec.py
    └── e2e/app_e2e_spec.py
```

This is a one-folder-per-feature layout under `src/`, using
underscore-suffixed file names (see
[the note on file names](/#a-note-on-file-names)).

## Running the app

```bash
uv run austial serve
```

is equivalent to:

```bash
uv run uvicorn src.main:app --reload
```

`austial serve` just locates your project root (by walking up for
`src/app_module.py`), puts it on `sys.path`, and calls
`uvicorn.run("src.main:app", reload=True)` for you.

## The bootstrap file

Every Austial app boots from `src/main.py`:

```python
from austial import AustialFactory
from austial.common.exceptions import AllExceptionsFilter
from austial.common.pipes import ValidationPipe

from .app_module import AppModule

def _build():
    app = AustialFactory.create(AppModule, title="my-app")
    app.use_global_pipes(ValidationPipe())
    app.use_global_filters(AllExceptionsFilter())
    app.enable_cors()
    return app

app = _build()

async def bootstrap() -> None:
    await app.listen(8000)
```

`AustialFactory.create(AppModule)` walks the `@Module` import graph starting
at `AppModule`, registers every provider/controller into one DI container,
wires up any middleware declared via `configure()`, and returns an
`AustialApplication` -- a thin wrapper around a real `fastapi.FastAPI`
instance. Because it's ASGI-callable itself (`app(scope, receive, send)`),
`uvicorn src.main:app` works directly.

## Developing without installing from PyPI

If you're contributing to the framework itself rather than consuming it,
point a new project at your local checkout instead:

```bash
git clone https://github.com/austial/austial-py
cd austial-py && uv sync
uv run austial new ../my-app --link .
```

`--link` adds an editable `[tool.uv.sources]` entry so `uv sync` resolves
`austial` from your local checkout instead of PyPI.

## Next steps

Continue to [Controllers](/overview/controllers) to see how routing works,
or jump to the [CLI](/cli/overview) reference to see the full
`generate`/`g` schematic list.

---

Source: https://docs.austial.com/overview/controllers

# Controllers

Controllers are responsible for handling incoming requests and returning
responses to the client. A controller's
purpose is to receive requests and delegate business logic to a
**provider** (see [Providers](/overview/providers)).

## Routing

`@Controller()` attaches a route prefix; the HTTP-verb decorators
(`@Get`, `@Post`, `@Put`, `@Patch`, `@Delete`, `@Options`, `@Head`) attach a
sub-path and register the method as a route handler:

```python
from austial import Controller, Get, Post

@Controller("cats")
class CatsController:
    def __init__(self, cats_service: CatsService):
        self.cats_service = cats_service

    @Get()
    async def find_all(self) -> list[Cat]:
        return self.cats_service.find_all()

    @Post()
    async def create(self, dto: CreateCatDto = Body()) -> Cat:
        return self.cats_service.create(dto)
```

`@Controller("cats")` + `@Get()` maps to `GET /cats`, `@Post()` maps to
`POST /cats` — a prefix-plus-suffix composition.

## Route parameters

Austial route strings use a familiar `:paramName` syntax. A decoration-time
regex rewrites `:id` into FastAPI's `{id}` before FastAPI ever sees the path:

```python
@Get(":id")
async def find_one(self, id: int) -> Cat:
    return self.cats_service.find_one(id)
```

```python
@Get("users/:userId/posts/:postId")
async def find_post(self, user_id: int, post_id: int):
    ...
```

A trailing `*` on a param name (`:key*`) declares a catch-all that also
matches `/`, translated to FastAPI's `{key:path}` converter instead of the
plain `{key}`. Use it for path-shaped values such as storage keys:

```python
@Get("files/:key*")
async def download(self, key: str = Param("key")) -> Response:
    ...  # matches GET /files/a/b/c.pdf with key == "a/b/c.pdf"
```

Because Python has no parameter *decorators*, request data is bound the same
way FastAPI already does it — through type-hinted parameters, optionally
given an explicit default from `austial`'s param helpers:

```python
from austial import Body, Headers, Param, Query

@Get(":id")
async def find_one(
    self,
    id: int = Param(),                 # equivalent to plain `id: int`
    verbose: bool = Query(default=False),
    x_request_id: str | None = Headers("x-request-id"),
):
    ...

@Post()
async def create(self, dto: CreateCatDto = Body()) -> Cat:
    ...
```

| Decorator/helper | Binds to |
|---|---|
| `Param(name=None, ...)` | path parameter |
| `Query(name=None, default=None, ...)` | query string |
| `Body(default=..., embed=False, ...)` | request body (usually a pydantic `BaseModel`) |
| `Headers(name=None, default=None, ...)` | request header |
| `Req` / `Res` | raw `fastapi.Request` / `fastapi.Response` |

In practice, plain type-hinted path params (`id: int`) already resolve
correctly via FastAPI's own inference from `{id}` in the path — `Param()` is
there for explicitness. `Body()` is the one you'll reach for
most, since it's what turns a plain constructor parameter into an
explicitly-bound request body.

## Status codes and headers

```python
from austial import HttpCode, Header

@Post()
@HttpCode(202)
@Header("X-Powered-By", "Austial")
async def create(self, dto: CreateCatDto = Body()) -> Cat:
    ...
```

Without `@HttpCode`, `POST` handlers return `201 Created`, every other verb
returns `200 OK`.

## Guards, pipes, interceptors on a controller

Any of `@UseGuards`, `@UsePipes`, `@UseInterceptors`, `@UseFilters` can be
stacked on a controller class (applies to every route in it) or on an
individual handler (applies to that route only):

```python
from austial import Controller, Get, UseGuards, UseInterceptors

from .api_key_guard import ApiKeyGuard
from austial.common.interceptors import TransformInterceptor

@Controller("cats")
@UseGuards(ApiKeyGuard)          # every route in this controller requires the guard
class CatsController:
    @Get(":id")
    @UseInterceptors(TransformInterceptor())   # only this route wraps its response
    async def find_one(self, id: int):
        ...
```

See [Guards](/overview/guards), [Pipes](/overview/pipes),
[Interceptors](/overview/interceptors) and
[Exception Filters](/overview/exception-filters) for what each of these
actually do in the request pipeline.

## Full CRUD example

This is exactly what `austial generate resource cats` scaffolds for you
(see [CLI: Generating Resources](/cli/generating)):

```python
from austial import Body, Controller, Delete, Get, Injectable, Patch, Post

@Controller("cats")
class CatController:
    def __init__(self, service: CatService):
        self.service = service

    @Post()
    async def create(self, dto: CreateCatDto = Body()) -> Cat:
        return self.service.create(dto)

    @Get()
    async def find_all(self) -> list[Cat]:
        return self.service.find_all()

    @Get(":id")
    async def find_one(self, id: int) -> Cat:
        return self.service.find_one(id)

    @Patch(":id")
    async def update(self, id: int, dto: UpdateCatDto = Body()) -> Cat:
        return self.service.update(id, dto)

    @Delete(":id")
    async def remove(self, id: int) -> dict:
        self.service.remove(id)
        return {"deleted": True}
```

---

Source: https://docs.austial.com/overview/providers

# Providers

Providers are the core concept in Austial. Services, repositories,
factories, helpers — anything that can be injected as a
dependency is a provider. The main idea is that a provider can be
**injected** as a dependency, meaning objects can create various
relationships with each other, and the "wiring up" of these objects can
largely be delegated to Austial's DI container.

## A basic provider

```python
from austial import Injectable, NotFoundException

@Injectable()
class CatsService:
    def __init__(self) -> None:
        self._cats: list[Cat] = []
        self._next_id = 1

    def create(self, dto: CreateCatDto) -> Cat:
        cat = Cat(id=self._next_id, **dto.model_dump())
        self._next_id += 1
        self._cats.append(cat)
        return cat

    def find_all(self) -> list[Cat]:
        return self._cats

    def find_one(self, id: int) -> Cat:
        for cat in self._cats:
            if cat.id == id:
                return cat
        raise NotFoundException(f"Cat #{id} not found")
```

`@Injectable()` marks a class as manageable by the DI container. Register
it on a module's `providers=[...]` and it becomes available for
constructor injection anywhere within that module's dependency graph:

```python
@Module(controllers=[CatsController], providers=[CatsService])
class CatsModule:
    pass
```

## Dependency injection

Austial's container resolves `__init__` type hints recursively — no
decorators or tokens required for the common case:

```python
@Controller("cats")
class CatsController:
    def __init__(self, cats_service: CatsService):
        self.cats_service = cats_service
```

When `CatsController` is instantiated, the container inspects
`CatsService`'s own constructor, resolves *its* dependencies first, and so
on, recursively. Providers are **singletons by default**: the first `resolve()`
builds and caches the instance; every subsequent resolution (including
across different controllers/services within the same app) returns that
same cached object.

```python
first = container.resolve(CatsService)
second = container.resolve(CatsService)
assert first is second
```

## Injection scope

```python
from austial import Injectable, Scope

@Injectable(scope=Scope.TRANSIENT)
class RequestIdGenerator:
    ...
```

| Scope | Behavior |
|---|---|
| `Scope.DEFAULT` | Singleton — one instance for the lifetime of the application (default). |
| `Scope.TRANSIENT` | A new instance is built every time it's resolved. |

Austial does not currently implement a request scope (a fresh instance per
inbound request) — see
[Injection Scopes](/fundamentals/injection-scopes) for details.

## Injecting by token instead of type

Not every dependency is a class you can type-hint against — a connection
string, a raw value, a repository bound to a specific entity. For those,
use `Inject(token)` as an explicit default (`InjectRepository(Entity)` in
[Database](/techniques/database) is built the same way):

```python
from austial import Inject

@Injectable()
class AnalyticsService:
    def __init__(self, api_prefix=Inject("ANALYTICS_API_PREFIX")):
        self.api_prefix = api_prefix
```

`Optional_()` marks a dependency as optional — if it can't be resolved,
the container passes `None` instead of raising:

```python
from austial import Optional_

@Injectable()
class MetricsService:
    def __init__(self, statsd_client=Optional_()):
        self.statsd_client = statsd_client
```

## Custom providers

Beyond a plain class, a module's `providers=[...]` list also accepts
provider dicts for `useValue`/`useClass`/`useFactory` --
see [Custom Providers](/fundamentals/custom-providers) for the full
reference, and [Dynamic Modules](/fundamentals/dynamic-modules) for how
`ConfigModule.for_root()`/`OrmModule.for_root_async()` are built on top
of the same mechanism.

## Provider registration errors

If a provider isn't registered anywhere reachable from the module graph and
can't be auto-registered (i.e. it's not a plain type), resolving it raises
`ProviderNotFoundError`:

```
Nest can't resolve dependency 'CatsService'. Is it registered in a module's
`providers=[...]`?
```

A dependency cycle (`A` needs `B`, `B` needs `A`) raises
`CircularDependencyError` instead of recursing forever — see
[Circular Dependency](/fundamentals/circular-dependency).

---

Source: https://docs.austial.com/overview/modules

# Modules

A module is a class annotated with `@Module()`. The decorator provides
metadata that Austial uses to organize the application structure.

```python
from austial import Module

from .cats_controller import CatsController
from .cats_service import CatsService

@Module(
    controllers=[CatsController],
    providers=[CatsService],
)
class CatsModule:
    pass
```

| Property | Meaning |
|---|---|
| `imports` | Other modules whose exported providers this module needs. |
| `controllers` | Controllers defined in this module, instantiated by Austial. |
| `providers` | Providers registered by this module, resolvable within it (and, currently, application-wide — see the note below). |
| `exports` | The subset of `providers` (or re-exported imported tokens) other modules can use once they `import` this module. |

## The root module

Every Austial app has at least one root module, conventionally
`AppModule`, which `AustialFactory.create()` starts scanning from:

```python
from austial import Module
from austial.common.middleware import LoggingMiddleware, MiddlewareConsumer
from austial.config import ConfigModule

from .app_controller import AppController
from .app_service import AppService
from .modules.health.health_module import HealthModule

@Module(
    imports=[ConfigModule.for_root(), HealthModule],
    controllers=[AppController],
    providers=[AppService],
)
class AppModule:
    def configure(self, consumer: MiddlewareConsumer) -> None:
        consumer.apply(LoggingMiddleware).for_routes("*")
```

`AustialFactory.create(AppModule)` walks this `imports` graph recursively
(`austial.core.module_scanner.scan_module`), registering every module's
`controllers`/`providers` into one shared `Container`, and de-duplicating
modules that are imported more than once.

## Feature modules

Group a feature's controller + service + DTOs + entity behind its own
module, then `import` it from `AppModule` — one folder per feature, which
is exactly what `austial generate module <name>` /
`austial generate resource <name>` scaffold automatically (see
[CLI](/cli/generating)):

```
src/modules/cats/
├── cats_module.py
├── cats_controller.py
├── cats_service.py
├── entities/cats_entity.py
└── dto/
    ├── create_cats_dto.py
    └── update_cats_dto.py
```

```python
@Module(imports=[CatsModule], controllers=[AppController], providers=[AppService])
class AppModule:
    pass
```

## Global modules

```python
from austial import Global, Module

@Global()
@Module(providers=[ConfigService], exports=[ConfigService])
class ConfigModule:
    pass
```

`@Global()` marks a module's exports as available everywhere without every
consumer needing to list it in their own `imports` (used internally by
`ConfigModule.for_root()`).

## A note on module visibility

Austial's DI container is currently **application-wide and flat**: every
module's `providers` register into the same `Container` instance
regardless of `imports`/`exports`. In practice this means anything
registered anywhere in the graph is resolvable anywhere else — `imports`
and `exports` are honored for *scanning* (so you must still `import` a
module for its controllers/providers to be registered at all) but are not
yet enforced as a visibility boundary. Write your `imports`/`exports`
correctly anyway: it documents intent and means nothing changes for your
code if/when strict visibility lands.

## Dynamic modules

`ConfigModule.for_root(...)` and `OrmModule.for_root_async(...)` are
**dynamic modules** — static methods that build and return module metadata
at runtime instead of a fixed `@Module(...)` decorator. See
[Dynamic Modules](/fundamentals/dynamic-modules) for how to write your own.

---

Source: https://docs.austial.com/overview/middleware

# Middleware

Middleware is a function (here, a class implementing `.use()`) called
**before** the route handler, sitting in front of the whole
guards/pipes/interceptors/handler/filters pipeline — the same position it
occupies in most web framework middleware chains.

## Writing middleware

```python
from collections.abc import Awaitable, Callable

from fastapi import Request, Response

from austial.common.middleware import NestMiddleware

class LoggingMiddleware(NestMiddleware):
    async def use(
        self,
        request: Request,
        call_next: Callable[[], Awaitable[Response]],
    ) -> Response:
        print(f"[Request] {request.method} {request.url.path}")
        response = await call_next()
        print(f"[Response] {response.status_code}")
        return response
```

`call_next()` invokes the rest of the pipeline (the next middleware, or the
route handler if this is the last one) and returns its `Response` — call it
exactly once.

Austial ships this exact pattern built in: `austial.common.middleware.LoggingMiddleware`,
which logs `{METHOD} {path} {status_code} - {duration}ms` through the
framework's own `Logger`.

## Applying middleware

Middleware is wired up in a module's `configure()` method using a
`MiddlewareConsumer`:

```python
from austial import Module
from austial.common.middleware import LoggingMiddleware, MiddlewareConsumer

@Module(...)
class AppModule:
    def configure(self, consumer: MiddlewareConsumer) -> None:
        consumer.apply(LoggingMiddleware).for_routes("*")
```

```python
consumer.apply(LoggingMiddleware, AuthMiddleware).for_routes("/cats")
```

`for_routes(...)` accepts one or more route patterns:

| Pattern | Matches |
|---|---|
| `"*"` or `"(.*)"` | every route |
| `"/cats"` | an exact path |
| `"/cats/*"` | any path prefixed with `/cats/` |

`AustialFactory.create()` resolves every module's `configure()` after
scanning the module graph, collects all `MiddlewareBinding`s, and installs
a single dispatcher that chains matched middlewares' `use()` calls in
declaration order — one real Starlette `BaseHTTPMiddleware` under the hood,
so this composes cleanly with anything else mounted on the underlying
FastAPI app via `app.get_http_adapter()`.

## Where middleware sits in the pipeline

```
middleware -> guards -> pipes -> interceptors (wrapping) -> handler -> exception filters
```

Because middleware runs before routing is even resolved, it can't access
route-specific metadata the way a guard can via `ExecutionContext` — use a
[guard](/overview/guards) instead when you need to know *which* controller
or handler is about to run.

---

Source: https://docs.austial.com/overview/exception-filters

# Exception Filters

Austial ships a built-in exceptions layer that converts thrown exceptions
into a consistent JSON error response — no `try/except` in every handler
required.

## The default error response

Any unhandled exception becomes:

```json
{
  "statusCode": 404,
  "message": "Widget #999 not found",
  "error": "Not Found",
  "timestamp": "2026-07-12T10:15:00.000000+00:00",
  "path": "/widgets/999"
}
```

`statusCode`/`message`/`error`/`timestamp`/`path` — the shape every
default exception filter response uses.

## Built-in HTTP exceptions

```python
from austial import NotFoundException

raise NotFoundException(f"Cat #{id} not found")
```

| Class | Status |
|---|---|
| `BadRequestException` | 400 |
| `UnauthorizedException` | 401 |
| `ForbiddenException` | 403 |
| `NotFoundException` | 404 |
| `MethodNotAllowedException` | 405 |
| `ConflictException` | 409 |
| `UnprocessableEntityException` | 422 |
| `TooManyRequestsException` | 429 |
| `InternalServerErrorException` | 500 |
| `NotImplementedException` | 501 |
| `ServiceUnavailableException` | 503 |

Every subclass takes the same constructor: `(message: str | dict | None = None)`.
Pass a string for a plain `message`, or a dict to control the body more
precisely:

```python
raise BadRequestException({"message": ["name is required"], "error": "Bad Request"})
```

Any of these raised from inside a handler (or from a guard/pipe/interceptor)
is automatically caught and turned into the JSON shape above with the
matching status code — you never construct the response yourself.

## Writing a custom filter

```python
from austial.common.filters import ArgumentsHost, Catch, ExceptionFilter
from fastapi.responses import JSONResponse

@Catch(NotFoundException)
class NotFoundFilter(ExceptionFilter):
    async def catch(self, exception: NotFoundException, host: ArgumentsHost) -> JSONResponse:
        request = host.get_request()
        return JSONResponse(
            status_code=404,
            content={
                "statusCode": 404,
                "message": str(exception),
                "path": request.url.path,
                "hint": "Check the ID and try again.",
            },
        )
```

`@Catch(*exceptions)` declares which exception types a filter handles;
`@Catch()` with no arguments catches everything.

## Applying filters

```python
app.use_global_filters(AllExceptionsFilter())
```

registers a filter application-wide — this is what every generated
project's `src/main.py` does with the built-in `AllExceptionsFilter`.
`@UseFilters(*filters)` attaches filters to a single controller or handler
instead, the same controller/handler-scoped pattern as guards, pipes, and
interceptors:

```python
from austial import UseFilters

@Controller("cats")
@UseFilters(NotFoundFilter())
class CatsController:
    ...
```

## Dispatch order

When multiple filters are registered, Austial walks them in order and uses
the first one whose `@Catch(...)` types match the raised exception
(an empty `@Catch()` always matches). If none match, the exception
re-raises — this is why `AllExceptionsFilter` (registered with an empty
`@Catch()`) is the sane default to always have installed as a catch-all.

Validation errors raised by FastAPI itself (`RequestValidationError`) and
any other uncaught exception are automatically funneled through the same
filter chain, so a forgotten `try/except` still produces the same
consistent error body instead of a raw traceback.

---

Source: https://docs.austial.com/overview/pipes

# Pipes

A pipe transforms or validates input data before it reaches a route handler.

## The `PipeTransform` interface

```python
from austial.common.pipes import ArgumentMetadata, PipeTransform

class ParseIntPipe(PipeTransform):
    def transform(self, value, metadata: ArgumentMetadata):
        try:
            return int(value)
        except (TypeError, ValueError) as exc:
            raise BadRequestException(f"{metadata.data} must be an integer") from exc
```

`ArgumentMetadata` describes the argument being processed:

```python
@dataclass
class ArgumentMetadata:
    type: Literal["body", "query", "param", "custom"]
    data: str | None       # the parameter name
    metatype: type | None  # the parameter's runtime type
```

## The built-in `ValidationPipe`

```python
from austial.common.pipes import ValidationPipe

app.use_global_pipes(ValidationPipe())
```

`ValidationPipe` re-validates pydantic-model values (already parsed by
FastAPI against your type hints) and, on failure, raises a
`BadRequestException` shaped as:

```json
{
  "statusCode": 400,
  "message": ["field required: name"],
  "error": "Bad Request"
}
```

Since FastAPI already parses and type-checks request bodies against your
pydantic `BaseModel` DTOs before your handler runs, `ValidationPipe`'s main
job is normalizing *how* a validation failure is reported — into the same
error body shape every other exception produces, rather than FastAPI's
native `422` shape.

Every project scaffolded by `austial new` registers it globally in
`src/main.py` by default.

## Applying pipes

Globally (as above), or scoped to a controller/handler with `@UsePipes`:

```python
from austial import UsePipes

@Post()
@UsePipes(ValidationPipe())
async def create(self, dto: CreateCatDto = Body()) -> Cat:
    ...
```

## Pipeline position

```
guards -> pipes -> interceptors (wrapping) -> handler -> exception filters
```

Pipes run after guards have authorized the request, but before interceptors
wrap the handler call — so a pipe can reject a request (by raising) before
any interceptor logic (like response transformation or timing) ever runs.

---

Source: https://docs.austial.com/overview/guards

# Guards

A guard determines whether a given request will be handled by the route
handler or not, based on certain conditions (permissions, roles, API keys,
ACLs, ...).

## The `CanActivate` interface

```python
from austial import CanActivate, ExecutionContext

class ApiKeyGuard(CanActivate):
    async def can_activate(self, context: ExecutionContext) -> bool:
        request = context.switch_to_http().get_request()
        return request.headers.get("x-api-key") == "expected-value"
```

`ExecutionContext` gives a guard access to the current request and the
handler/controller about to run:

```python
class ExecutionContext:
    def switch_to_http(self) -> "ExecutionContext": ...
    def get_request(self) -> Request
    def get_class(self) -> type      # the controller class
    def get_handler(self) -> Any     # the handler method
```

If `can_activate` returns a falsy value, Austial short-circuits the
pipeline and raises `ForbiddenException("Forbidden resource")` — a
`403`, formatted through the same [exception filter](/overview/exception-filters)
chain as everything else.

## Applying guards

```python
from austial import Controller, Get, UseGuards

from .api_key_guard import ApiKeyGuard

@Controller("health")
class HealthController:
    @Get("protected")
    @UseGuards(ApiKeyGuard)
    async def protected(self):
        return {"status": "ok"}
```

Stack `@UseGuards` on the controller class to protect every route in it,
or on a single handler for one route only — guards attached at multiple
levels combine.

```python
@Controller("cats")
@UseGuards(ApiKeyGuard)   # applies to every route below
class CatsController:
    ...
```

## Binding guards outside a controller

`ApiKeyGuard` itself is just an `@Injectable()`-free plain class here, but
nothing stops it from being a real provider with injected dependencies
(a `ConfigService` to read the expected key from, for example) — the
container resolves guard classes the same way it resolves controllers and
services.

```python
class ApiKeyGuard(CanActivate):
    def __init__(self, config: ConfigService):
        self._expected = config.get("API_KEY", "changeme")

    async def can_activate(self, context: ExecutionContext) -> bool:
        request = context.switch_to_http().get_request()
        return request.headers.get("x-api-key") == self._expected
```

## Pipeline position

```
guards -> pipes -> interceptors (wrapping) -> handler -> exception filters
```

Guards run **first** — before any pipe validates input or any interceptor
wraps the call — so unauthorized requests never reach handler logic (or
pay the cost of body validation).

---

Source: https://docs.austial.com/overview/interceptors

# Interceptors

An interceptor wraps a route handler's execution, letting you run logic
before *and* after the handler runs — transform the response, add timing,
log calls, cache results. There's no RxJS here: `CallHandler.handle()` is
a plain async continuation rather than an Observable.

## The `NestInterceptor` interface

```python
from austial import CallHandler, ExecutionContext, NestInterceptor

class TimingInterceptor(NestInterceptor):
    async def intercept(self, context: ExecutionContext, next: CallHandler):
        start = time.perf_counter()
        result = await next.handle()          # runs the handler (and any inner interceptors)
        elapsed_ms = (time.perf_counter() - start) * 1000
        print(f"{context.get_handler().__name__} took {elapsed_ms:.1f}ms")
        return result
```

`next.handle()` is the continuation — call it to run the rest of the
pipeline and get the handler's return value; skip calling it to short-circuit
entirely and return something else instead (a cached value, for example).

## The built-in `TransformInterceptor`

```python
from austial.common.interceptors import TransformInterceptor
```

```python
class TransformInterceptor(NestInterceptor):
    async def intercept(self, context: ExecutionContext, next: CallHandler):
        result = await next.handle()
        return {"data": result, "timestamp": datetime.now(UTC).isoformat()}
```

Wraps every response in a familiar `{ data, timestamp }` envelope:

```python
@Get(":id")
@UseInterceptors(TransformInterceptor())
async def find_one(self, id: int):
    return self.cats_service.find_one(id)
```

```json
{
  "data": { "id": 1, "name": "Whiskers" },
  "timestamp": "2026-07-12T10:15:00.000000+00:00"
}
```

## Applying interceptors

Same controller/handler scoping as guards and pipes:

```python
@Controller("cats")
@UseInterceptors(LoggingInterceptor())   # every route
class CatsController:
    @Get(":id")
    @UseInterceptors(TransformInterceptor())   # this route only, stacks with the class-level one
    async def find_one(self, id: int):
        ...
```

When multiple interceptors apply, the first-declared interceptor is
outermost — it's the first to run on the way in, and the last to see the
result on the way out, just like nested middleware.

## Pipeline position

```
guards -> pipes -> interceptors (wrapping) -> handler -> exception filters
```

Interceptors sit closest to the handler: by the time one runs, guards have
already authorized the request and pipes have already validated/transformed
its arguments — an interceptor only ever sees a request that's cleared both.

---

Source: https://docs.austial.com/overview/custom-decorators

# Custom Decorators

Austial is built around a single mechanism — a metadata bag attached to
every decorated class or function (`austial.core.metadata.set_metadata` /
`get_metadata`). Every framework decorator (`@Module`,
`@Controller`, the route decorators, `@UseGuards`, `@Injectable`, ...) is
built on top of it, and it's directly available to you for your own
decorators, guards, and interceptors.

## `set_metadata` / `get_metadata`

```python
from austial.core.metadata import get_metadata, set_metadata

ROLES_KEY = "roles"

def Roles(*roles: str):
    def decorator(target):
        set_metadata(ROLES_KEY, list(roles), target)
        return target

    return decorator
```

```python
@Controller("cats")
class CatsController:
    @Get()
    @Roles("admin", "moderator")
    async def find_all(self):
        ...
```

This is a decorator factory that stamps metadata onto whatever it's applied
to (class or method).

## Reading metadata with `Reflector`

`austial.core.reflector.Reflector` is the read-side counterpart, injectable
anywhere the container resolves dependencies — most commonly inside a
guard:

```python
from austial import CanActivate, ExecutionContext
from austial.core.reflector import Reflector

from .roles_decorator import ROLES_KEY

class RolesGuard(CanActivate):
    def __init__(self, reflector: Reflector):
        self._reflector = reflector

    async def can_activate(self, context: ExecutionContext) -> bool:
        required_roles = self._reflector.get(ROLES_KEY, context.get_handler())
        if not required_roles:
            return True   # no @Roles(...) on this handler -> no restriction

        request = context.get_request()
        user_roles = request.headers.get("x-roles", "").split(",")
        return any(role in user_roles for role in required_roles)
```

```python
@Get()
@Roles("admin")
@UseGuards(RolesGuard)
async def find_all(self):
    ...
```

`Reflector` also exposes `get_all_and_merge`/`get_all_and_override` for
reading the same metadata key stamped at both the controller and handler
level and combining them.

## Where this is used internally

Every `@Use*` decorator (`@UseGuards`, `@UsePipes`, `@UseInterceptors`,
`@UseFilters`), `@HttpCode`, `@Header`, and `@Catch` are themselves thin
wrappers around `set_metadata`/`get_metadata` with a dedicated key
(`GUARDS_METADATA`, `PIPES_METADATA`, `INTERCEPTORS_METADATA`,
`FILTERS_METADATA`, `CATCH_METADATA` in `austial/core/metadata.py`) — so a
custom decorator you write with the same primitives is a first-class
citizen, not a workaround.

---

Source: https://docs.austial.com/fundamentals/custom-providers

# Custom Providers

A module's `providers=[...]` list accepts more than a plain `@Injectable()`
class — `useValue`/`useClass`/`useFactory` shapes, expressed as plain dicts
(Austial has no decorators-only metaprogramming trick for this, so a dict
with a `"provide"` key is the literal provider definition).

## Value providers — `useValue`

Bind a token to a fixed, already-constructed value — handy for constants,
mocks, or third-party client instances:

```python
providers=[
    {"provide": "API_KEY", "useValue": "super-secret"},
]
```

```python
from austial import Inject

@Injectable()
class ExternalApiService:
    def __init__(self, api_key: str = Inject("API_KEY")):
        self._api_key = api_key
```

## Class providers — `useClass`

Bind a token to a class the container should instantiate (resolving *its*
constructor dependencies as usual) — useful for swapping implementations
per environment:

```python
providers=[
    {"provide": CacheService, "useClass": RedisCacheService},
]
```

Any consumer that type-hints `CacheService` receives a `RedisCacheService`
instance instead.

## Factory providers — `useFactory`

Bind a token to the return value of a function, optionally with its own
injected dependencies via `inject=[...]`:

```python
providers=[
    {
        "provide": "DATABASE_URL",
        "useFactory": lambda config: config.get("DATABASE_URL", "sqlite+aiosqlite:///./app.db"),
        "inject": [ConfigService],
    },
]
```

The factory function is called with each of `inject`'s tokens resolved and
passed positionally, in order — this is exactly how
`OrmModule.for_root_async(use_factory=, inject=)` and
`ConfigModule.for_root()` build their own providers under the hood (see
[Dynamic Modules](/fundamentals/dynamic-modules) and
[Database](/techniques/database)).

## Non-class tokens

Any hashable value works as a `"provide"` token — a string (as above), or
a computed per-entity key:

```python
def repository_token(entity: type) -> str:
    return f"austial:orm:repository:{entity.__module__}.{entity.__qualname__}"
```

is precisely the pattern `austial.orm`'s `OrmModule.for_feature(...)` /
`InjectRepository(...)` use internally, so each entity gets its own
injectable `Repository` token without a shared string constant to collide on.

---

Source: https://docs.austial.com/fundamentals/dynamic-modules

# Dynamic Modules

A dynamic module is a module whose metadata (`imports`/`controllers`/
`providers`/`exports`) is computed at runtime by a factory method instead of
being fixed at decoration time. `ConfigModule.for_root()` and
`OrmModule.for_root_async()` are both built this way.

## Writing a dynamic module

```python
from austial.core.dynamic_module import DynamicModule

class CacheModule:
    @staticmethod
    def for_root(*, ttl_seconds: int = 60) -> DynamicModule:
        return DynamicModule(
            module=CacheModule,
            providers=[
                {"provide": "CACHE_TTL", "useValue": ttl_seconds},
                CacheService,
            ],
            exports=[CacheService],
        )
```

```python
@Module(imports=[CacheModule.for_root(ttl_seconds=300)])
class AppModule:
    pass
```

`DynamicModule` is a plain dataclass with the same shape as `@Module`'s
metadata (`module`, `imports`, `controllers`, `providers`, `exports`) --
the module scanner treats a `DynamicModule` instance in an `imports=[...]`
list exactly like a decorated module class, registering its
`controllers`/`providers` and recursing into its own `imports`.

## Real examples in the framework

```python
@Module()
class ConfigModule:
    @staticmethod
    def for_root(env_file: str = ".env", *, is_global: bool = True) -> DynamicModule:
        ...
```

```python
@Module()
class OrmModule:
    @staticmethod
    def for_root_async(*, use_factory, inject=(), entities, synchronize=False) -> DynamicModule:
        ...
```

See [Configuration](/techniques/configuration) and
[Database](/techniques/database) for what each of these actually
provisions.

## Async dynamic modules

There's no separate `forRootAsync` mechanism to learn beyond what
[factory providers](/fundamentals/custom-providers) already give you --
`OrmModule.for_root_async(use_factory=, inject=, entities=)` builds its
`DataSource` lazily via a `useFactory` provider under the hood, resolved
the first time something actually depends on the `DataSource` token. Note
that `use_factory` itself always runs synchronously (there's no
async-lifecycle hook in the module scanner yet — see
[Database](/techniques/database)'s note on calling `DataSource.initialize()`
explicitly). There's no meaningful difference between a "sync" and "async"
dynamic module in Austial beyond whether the values you're providing need
other injected dependencies to compute.

---

Source: https://docs.austial.com/fundamentals/injection-scopes

# Injection Scopes

```python
from austial import Injectable, Scope
```

| Scope | Behavior |
|---|---|
| `Scope.DEFAULT` | **Singleton** (default). One instance shared across the whole application lifetime. |
| `Scope.TRANSIENT` | A brand-new instance is constructed every time the provider is resolved. |

```python
@Injectable(scope=Scope.DEFAULT)     # same as plain @Injectable()
class CatsService:
    ...

@Injectable(scope=Scope.TRANSIENT)
class RequestIdGenerator:
    def __init__(self):
        self.id = uuid4()
```

Every place `RequestIdGenerator` is a constructor dependency gets its own
fresh instance; every place `CatsService` is a dependency shares the exact
same object.

```python
a = container.resolve(RequestIdGenerator)
b = container.resolve(RequestIdGenerator)
assert a is not b   # transient: different instances

x = container.resolve(CatsService)
y = container.resolve(CatsService)
assert x is y       # default: singleton
```

## No request scope (yet)

A request scope — a fresh instance per inbound HTTP request, letting a
provider safely hold per-request state (like the current user) as instance
attributes — is not implemented yet. If you need per-request state today,
read it off the `Request` object directly (via a guard, interceptor, or
`Req` param) rather than expecting a request-scoped provider to carry it
for you.

---

Source: https://docs.austial.com/fundamentals/circular-dependency

# Circular Dependency

A circular dependency occurs when two providers depend on each other:
`ServiceA` needs `ServiceB` in its constructor, and `ServiceB` needs
`ServiceA` in its. Austial's container detects this during resolution
instead of recursing forever:

```python
@Injectable()
class ServiceA:
    def __init__(self, b: "ServiceB"):
        self.b = b

@Injectable()
class ServiceB:
    def __init__(self, a: ServiceA):
        self.a = a
```

Resolving either one raises `CircularDependencyError` describing the cycle,
instead of an unbounded recursion / stack overflow.

## Avoiding it

Three escape hatches apply:

1. **Redesign** — a circular dependency is usually a sign two providers
   should be merged, or that shared logic belongs in a third provider both
   depend on. This is the preferred fix.
2. **Lazy resolution via a token** — inject a factory/callable instead of
   the concrete type directly, deferring resolution until it's actually
   called rather than at construction time.
3. **`Optional_()`** — if one side of the relationship is genuinely
   optional, mark it as such so the container returns `None` instead of
   raising:

   ```python
   from austial import Optional_

   @Injectable()
   class ServiceA:
       def __init__(self, b: "ServiceB" = Optional_()):
           self.b = b
   ```

Prefer option 1 whenever possible — circular dependencies between services
are usually a design smell worth addressing directly rather than working
around.

---

Source: https://docs.austial.com/techniques/configuration

# Configuration

`austial.config` provides a `ConfigModule.for_root()` dynamic module that
loads a `.env` file and exposes its values (plus the rest of the process
environment) through an injectable `ConfigService`.

## Setup

```python
from austial import Module
from austial.config import ConfigModule

@Module(imports=[ConfigModule.for_root()])
class AppModule:
    pass
```

`ConfigModule.for_root(env_file=".env", is_global=True)`:

- loads `env_file` via `python-dotenv`'s `load_dotenv(env_file, override=False)`
  (a missing file or missing `python-dotenv` is a no-op, not an error),
- builds one `ConfigService` snapshotting `os.environ`,
- is registered `@Global()` by default (`is_global=True`), so you don't need
  to re-import `ConfigModule` in every feature module that wants
  `ConfigService`.

## Using `ConfigService`

```python
from austial import Injectable
from austial.config import ConfigService

@Injectable()
class ApiKeyGuard:
    def __init__(self, config: ConfigService):
        self._expected = config.get("API_KEY", "changeme")
```

```python
class ConfigService:
    def get(self, key: str, default: Any = None) -> Any: ...
    def get_or_throw(self, key: str) -> Any: ...   # raises KeyError if the key is missing
```

Use `get_or_throw` for configuration that has no sane default (a secret
key, an external API URL) so a missing value fails fast at
startup-adjacent code rather than silently falling through as `None`.

## `.env` files

A scaffolded project ships a `.env.example` — copy it to `.env` before
running:

```bash
cp .env.example .env
```

```env
PORT=8000
API_KEY=changeme
```

`.env` is git-ignored by default; `.env.example` is the checked-in template
documenting which variables a deployment needs to set.

---

Source: https://docs.austial.com/techniques/database

# Database

`austial.orm` is a from-scratch, [TypeORM](https://typeorm.io)-shaped ORM: `@Entity()`/`@Column()`
decorators, all four relation types, a `Repository`/`EntityManager` data-mapper API, a fluent
`QueryBuilder`, and transactions — wired natively into Austial's DI container via `OrmModule`,
the same way `@nestjs/typeorm` wires up `typeorm` in Nest.

SQLAlchemy Core is used internally, purely as a hidden SQL-compilation/connection-pooling layer
behind a pluggable `Driver` interface (PostgreSQL via `asyncpg`, SQLite via `aiosqlite` today) —
you never import or see SQLAlchemy anywhere in application code.

## Defining an entity

```python
from austial.orm import Column, Entity, PrimaryGeneratedColumn

@Entity()
class Cat:
    id: int = PrimaryGeneratedColumn()
    name: str = Column()
    age: int = Column(nullable=True)
```

Python has no field-level decorators, so (exactly like `Inject()` elsewhere in Austial) each
column/relation marker is used as the attribute's *default value* — `@Entity()` reads these
markers back off the class, converts it into a real `@dataclass`, and registers its schema
metadata. `Cat` instances are just plain mutable objects (`Cat(name="Tom")`, attribute
access/assignment) — no active-record magic, no lazy proxies, no dirty-checking.

## Setup

```python
from austial import Module
from austial.orm import OrmModule

from src.entities.cat import Cat

@Module(
    imports=[
        OrmModule.for_root(
            type_="postgres",
            url="postgresql+asyncpg://user:pass@localhost/mydb",
            entities=[Cat],
        ),
    ],
)
class AppModule:
    pass
```

Zero-config local development/tests with SQLite:

```python
OrmModule.for_root(type_="sqlite", url="sqlite+aiosqlite:///./austial.db", entities=[Cat])
```

Connection options resolved at runtime from other providers (the common case — reading a URL out
of `ConfigService`):

```python
from austial.config import ConfigService
from austial.orm import OrmModule

OrmModule.for_root_async(
    use_factory=lambda config: {
        "type_": "postgres",
        "url": config.get("DATABASE_URL"),
    },
    inject=[ConfigService],
    entities=[Cat],
)
```

`use_factory` is called with each of `inject`'s tokens resolved and passed positionally, and must
return a dict shaped like `for_root()`'s keyword arguments (`type_`, `url`, any extra
driver-specific kwargs). `entities`/`synchronize` stay static — they describe your app's schema,
not runtime connection config, so they're passed directly to `for_root_async()` itself rather
than through the factory.

### The one manual step: initializing the `DataSource`

`austial-core`'s module scanner walks `@Module(imports=[...])` synchronously and eagerly, and
there's no async module-init lifecycle hook (yet) for `OrmModule` to hook into. So
`OrmModule.for_root()` only *constructs* the `Driver`/`DataSource` object at module-scan time
(cheap — no connection opens yet); your entrypoint must explicitly initialize it before serving
requests, exactly like vanilla (non-Nest) TypeORM:

```python
from austial import AustialFactory
from austial.orm import DataSource

from src.app_module import AppModule

app = AustialFactory.create(AppModule)

async def bootstrap() -> None:
    await app.get(DataSource).initialize()
    await app.listen(8000)
```

`synchronize=True` (dev/test only — not the production path once migrations exist) makes
`initialize()` also issue `CREATE TABLE` for every entity in `entities=[...]`.

#### Production deployments (`uvicorn src.main:app` / `austial serve`)

The `bootstrap()` function above only runs for `python src/main.py` — `uvicorn src.main:app` and
`austial serve` (the deployment target in Docker/production) both just import `src.main:app` and
hand it straight to uvicorn, so `bootstrap()` is never called and `DataSource.initialize()` never
runs. For those entry points, register it as an ASGI `"startup"` lifespan event handler instead,
via `get_http_adapter()`:

```python
from austial import AustialFactory
from austial.orm import DataSource

from src.app_module import AppModule

app = AustialFactory.create(AppModule)

@app.get_http_adapter().on_event("startup")
async def _init_data_source() -> None:
    await app.get(DataSource).initialize()
```

This is safe with any global guard/pipe/interceptor/filter that constructor-injects a
`Repository` via `@InjectRepository()`, even though route-building resolves those constructors:
`AustialApplication` defers route-building until the first non-lifespan ASGI scope, so it always
runs *after* every `"startup"` handler registered here has already completed, for both
`app.listen()` and the bare-ASGI (`uvicorn`/`austial serve`) entry points.

## Repositories

Register entities with `OrmModule.for_feature([...])` in whichever feature module owns them, then
inject a `Repository` with `@InjectRepository()`:

```python
from austial import Injectable
from austial.orm import InjectRepository, Repository

from src.entities.cat import Cat

@Injectable()
class CatsService:
    def __init__(self, repository: Repository[Cat] = InjectRepository(Cat)):
        self.repository = repository

    async def create(self, name: str) -> Cat:
        return await self.repository.save(Cat(name=name))

    async def find_all(self) -> list[Cat]:
        return await self.repository.find()

    async def find_one(self, id: int) -> Cat | None:
        return await self.repository.find_one_by({"id": id})
```

`InjectRepository(Cat)` is typed `Any` (the marker `Container` actually resolves at DI time is an
implementation detail), so it type-checks as the default for a parameter annotated
`Repository[Cat]` — the same trick every other param-marker helper (`Body()`, `Inject()`, ...)
uses.

```python
@Module(
    imports=[OrmModule.for_feature([Cat])],
    providers=[CatsService],
)
class CatsModule:
    pass
```

`Repository[T]` methods: `find(options)`, `find_one(options)`, `find_by(criteria)`,
`find_one_by(criteria)`, `count(options)`, `save(entity | [entities])` (INSERT when the primary
key is unset, else a full-column UPDATE), `insert(entity | [entities])` (always an INSERT — see
below), `remove(entity | [entities])`, `update(criteria,
partial)` and `delete(criteria)` (bulk, no prior SELECT — same distinction as TypeORM's
`Repository.update()`/`.delete()` vs. `.save()`/`.remove()`), and `create_query_builder(alias)`.

### Application-assigned keys: `insert()`

`save()` picks INSERT or UPDATE from the primary key, so an entity whose key you assign yourself
(a UUID, a natural key) cannot be inserted through it — it would issue an UPDATE that matches
nothing. Use `insert()` (on `Repository` and `EntityManager`), which always issues an INSERT and
still stamps `CreateDateColumn`/`UpdateDateColumn`:

```python
await self.repository.insert(Invoice(id=str(uuid.uuid4()), number="INV-1", tenant_id=tenant_id))
```

A duplicate key raises the driver's integrity error. It is the only write an `append_only` entity
with a pre-set key will accept.

### Mixins

`@Entity()` also collects `Column()`/relation markers from plain (non-entity) base classes, so a
mixin can contribute columns to every entity that inherits it — `austial.tenancy.Tenanted` is one.
Attributes declared on the entity class itself win over the mixin's; mixin columns are appended after
the class's own and are regular constructor fields.

`FindOptions` supports `where` (a dict is one AND-of-equality group, a `list[dict]` is
OR-of-AND-groups), `relations` (single-level eager-loading via `LEFT JOIN`), `order_by`, `skip`,
`take`.

A `where` value can be wrapped in `In`/`NotIn` (mirrors TypeORM's `In()`) for membership tests
instead of equality:

```python
from austial.orm import In, NotIn

pending = await repository.find(FindOptions(where={"status": In(["PENDING", "MANUAL_REVIEW"])}))
active = await repository.find(FindOptions(where={"status": NotIn(["REJECTED"])}))
```

`IsNull()` and `NotNull()` compile to `IS NULL` / `IS NOT NULL`. A plain `None` value is `= NULL`, which
never matches, so use these to filter on an unset nullable column — and to claim a row atomically, for
example consuming a single-use token with `update({"token_hash": h, "used_at": IsNull()}, {"used_at": now})`
and checking `affected == 1`:

```python
from austial.orm import IsNull

unused = await repository.find(FindOptions(where={"used_at": IsNull()}))
```

Comparing a native Postgres `ENUM` column this way (equality or `In`/`NotIn`) "just works" without
any `CAST` — `EntityManager` resolves the column's real enum type from schema metadata and binds
the parameter with it, so no app-code workaround is needed here.

### Append-only entities

Pass `append_only=True` to `@Entity()` for rows that must only ever be inserted, never mutated or
removed — audit logs, ledger entries, anything with a regulatory retention requirement:

```python
from austial.orm import Column, Entity, PrimaryGeneratedColumn

@Entity(append_only=True)
class LedgerEntry:
    id: int = PrimaryGeneratedColumn()
    description: str = Column()
    amount: int = Column()
```

Composes with the optional table-name positional, e.g. `@Entity("ledger_entry",
append_only=True)`.

This is enforced at the `EntityManager` level — the single choke point every `Repository` method
funnels through — not by convention in your service code:

- `repository.insert(entity)` works for any key (including an application-assigned one);
  `repository.save(entity)` still works for the INSERT path (primary key unset); calling it again
  on an already-persisted instance (primary key set, which `save()` would otherwise treat as a
  full-column UPDATE) raises.
- `repository.update(criteria, partial)`, `repository.remove(entity)`, and
  `repository.delete(criteria)` always raise, unconditionally — there's no legitimate partial
  update or delete path for an append-only entity.

All four raise `austial.orm.AppendOnlyViolationError` (a subclass of `OrmError`), naming the
entity and the attempted operation.

**What this guard does *not* cover:** it's an application-level check inside `EntityManager`,
not a database-level trigger, `REVOKE`, or row-level-security policy — raw SQL run outside the
ORM (a migration, a `psql` session, a DB role with direct table access) is not stopped by this.
Pair it with an actual DB-level guard (e.g. a `REVOKE UPDATE, DELETE` grant or a trigger) if you
need enforcement that survives someone bypassing the ORM entirely.

### Aggregation

Alongside `count(options)`, `Repository[T]` has `sum(column, criteria=None, group_by=None)` and
the lower-level `aggregate(fn, column, criteria=None, group_by=None)` (`fn` is
`"SUM" | "AVG" | "MIN" | "MAX"`) for numeric rollups — e.g. total AUM across matching holdings:

```python
total_aum: float = await holdings_repository.sum("quantity", criteria={"status": "ACTIVE"})
```

Pass `group_by` (a sequence of attribute names) to get a breakdown instead of a single scalar —
e.g. AUM per currency:

```python
by_currency: list[dict] = await holdings_repository.sum("quantity", group_by=["currency"])
# [{"currency": "USD", "value": 125000.0}, {"currency": "EUR", "value": 42000.0}]
```

Without `group_by`, `sum`/`aggregate` return a single `float` (`0.0`, not `None`, when no rows
match — SQL's `SUM` of zero rows is `NULL`, normalized here). With `group_by`, they return one
`dict` per group actually present in the data (the group-by attribute names as keys, plus
`"value"`) — groups with no matching rows don't appear, standard SQL `GROUP BY` semantics. Same
scope as `count()`: WHERE (+ here, `GROUP BY`) only, no joins/paging, single-level (no dotted
relation paths) group-by columns.

## Relations

```python
from austial.orm import Column, Entity, JoinColumn, ManyToOne, OneToMany, PrimaryGeneratedColumn

@Entity()
class Owner:
    id: int = PrimaryGeneratedColumn()
    name: str = Column()
    cats: list["Cat"] = OneToMany(lambda: Cat, inverse_side="owner")

@Entity()
class Cat:
    id: int = PrimaryGeneratedColumn()
    name: str = Column()
    owner: Owner = ManyToOne(lambda: Owner, join_column=JoinColumn(name="owner_id"))
```

`OneToOne`, `OneToMany`, `ManyToOne`, `ManyToMany` all work the same way TypeORM's do, with one
deliberate deviation: TypeORM stacks `@JoinColumn()`/`@JoinTable()` decorators above the relation
decorator, but Python's "marker as default value" trick only allows one right-hand side per
attribute — so `JoinColumn`/`JoinTable` are passed as a `join_column=`/`join_table=` kwarg
instead. The target entity accepts either the class directly or a zero-arg callable (`lambda:
Cat`) for forward references.

Eager-load a relation with `FindOptions.relations` or `QueryBuilder.left_join_and_select()`:

```python
owners = await owner_repository.find(FindOptions(relations=["cats"]))
```

## `QueryBuilder`

For anything `FindOptions` can't express, drop to the fluent query builder — mirrors TypeORM's
own `createQueryBuilder()`:

```python
cats = await (
    repository.create_query_builder("cat")
    .left_join_and_select("cat.owner", "owner")
    .where("cat.age > :age", {"age": 2})
    .and_where("owner.name = :name", {"name": "Alice"})
    .order_by("cat.name", "ASC")
    .take(10)
    .get_many()
)
```

WHERE clauses are always a raw SQL fragment plus a **separate** bound-params dict — values are
never string-formatted into the SQL text, only ever bound through the DBAPI driver. This is what
makes `.where()`/`.and_where()`/`.or_where()` (and everything built on top of them, including
`FindOptions.where`) SQL-injection-safe by construction.

For the common cases of equality and membership, `.where_eq()`/`.and_where_eq()`/`.or_where_eq()`
and `.where_in()`/`.and_where_in()`/`.or_where_in()` (plus `.where_not_in()` and its `and_`/`or_`
variants) generate the bind params for you — no need to invent placeholder names:

```python
qb.and_where_eq("cat.name", "Tom").and_where_in("cat.age", [2, 3, 4])
```

Pass `cast_text=True` when the target column is a native Postgres `ENUM` (or any column where a
string-typed bind param needs it) — it wraps the column reference in `CAST(<column> AS TEXT)`,
ANSI SQL so it also works against the SQLite driver used in tests:

```python
qb.and_where_eq("s.status", "MANUAL_REVIEW", cast_text=True)
qb.and_where_not_in("s.status", ["VERIFIED", "REJECTED"], cast_text=True)
```

`QueryBuilder` deals with arbitrary joined aliases, so unlike `FindOptions.where`'s `In`/`NotIn`
(which resolves the real column type automatically), this is an explicit opt-in rather than
auto-detected.

**Reserved-word table/column names:** a table or column name that collides with a SQL reserved
keyword (`user`, `order`, `group`, `check`, ... — `user` is reserved in Postgres, for example) is
safe everywhere the framework builds a query *for* you — `FindOptions.where`, the
bare-scalar-primary-key shorthand (`repository.update(id, {...})`/`.delete(id)`),
`find_one_by`/`find_by`, and `FindOptions.order_by` all quote table/column identifiers
automatically. Raw fragments you write yourself via
`.where()`/`.and_where()`/`.or_where()`/`.order_by()`/`.add_order_by()` on `QueryBuilder` are
**not** auto-quoted — same contract as TypeORM's own querybuilder — so quote them yourself (e.g.
`.where('"order".total > :n', {"n": 2})`) if your alias or column is a reserved word.

## Transactions

```python
async def transfer_cat(data_source: DataSource, cat_id: int, new_owner_id: int) -> None:
    async def _run(manager: EntityManager) -> None:
        cat = await manager.find_one_by(Cat, {"id": cat_id})
        cat.owner_id = new_owner_id
        await manager.save(cat)

    await data_source.transaction(_run)
```

`DataSource.transaction(callback)` opens one connection, binds a fresh `EntityManager` to it, and
runs `callback` — commits on clean return, rolls back automatically if `callback` raises.

## Migrations with Alembic

`austial.orm` doesn't ship its own migration runner — use [Alembic](https://alembic.sqlalchemy.org)
directly. `DataSource.get_metadata()` returns the `sqlalchemy.MetaData` describing every entity the
`DataSource` was constructed with, which is what Alembic's `env.py` needs as `target_metadata`:

```python
# migrations/env.py
from src.app_module import entities  # whatever list you pass OrmModule.for_root(entities=...)
from austial.orm import DataSource, SQLiteDriver

data_source = DataSource(SQLiteDriver("sqlite+aiosqlite:///./austial.db"), entities)
target_metadata = data_source.get_metadata()
```

Building the entity catalog only walks entity/relation metadata — it opens no DB connection, so
this is safe to call before `DataSource.initialize()` too, which matters for Alembic: its process
must not open the app's real connection pool just to import `target_metadata` at module load time.

This is the one deliberate exception to `austial.orm`'s "SQLAlchemy stays hidden behind `Driver`"
rule — Alembic's own API requires a real `sqlalchemy.MetaData` instance, so `get_metadata()` is the
single documented seam where that type crosses the public API boundary.

## Swapping in a different database

Nothing above the `Driver` interface (`Repository`, `EntityManager`, `QueryBuilder`,
`DataSource`, every decorator) is dialect-specific — pick `type_="postgres"` or `type_="sqlite"`
in `OrmModule.for_root()`/`for_root_async()` and everything else keeps working unchanged. See
[Health Checks](/techniques/health-checks) for how `DatabaseHealthIndicator` uses the same
`DataSource` to verify connectivity via `data_source.driver.execute_raw("SELECT 1")`.

---

Source: https://docs.austial.com/techniques/migrations

# Migrations

`austial.migrations` wires [Alembic](https://alembic.sqlalchemy.org) to `austial.orm`'s
`DataSource`, so a product can generate and apply schema migrations without leaving the
framework. Alembic runs against the same SQLAlchemy `MetaData` the ORM builds from your
`@Entity()` classes — `DataSource.get_metadata()` is the single documented seam where
that metadata crosses the ORM's public API boundary (see [Database](/techniques/database)
for the ORM itself).

The framework ships:

- `MigrationsModule.for_root(...)` — dynamic module that registers `MigrationRunner`
  in the DI container.
- `MigrationRunner` — an injectable service with `revision`, `upgrade`, `downgrade`,
  `current` and `history` methods.
- `alembic_env(config, metadata)` — the online/offline runner your `env.py` delegates
  to, so upgrading Alembic itself never forces a `env.py` regeneration.
- `austial-migrate` — a standalone Typer CLI (`revision`, `upgrade`, `downgrade`,
  `current`, `history`, `init`) that runs without booting the DI container, so
  migrations can happen before the application imports cleanly at deploy time.

## Setup

Install the package (already there if you use `austial` as an umbrella dependency,
otherwise `uv add austial-migrations`) and register the module:

```python
from austial import Module
from austial.migrations import MigrationsConfig, MigrationsModule
from austial.orm import DataSource

from src.entities import Cat, Owner

_metadata = DataSource(...).get_metadata()  # cheap: no connection opens

@Module(
    imports=[
        MigrationsModule.for_root(
            MigrationsConfig(script_location="alembic"),
            url="postgresql+psycopg://user:pass@localhost/mydb",
            metadata=_metadata,
        ),
    ],
)
class AppModule:
    pass
```

`MigrationsConfig` accepts:

- `script_location` — on-disk directory holding `env.py` + `versions/` (default
  `"alembic"`).
- `version_table` — Alembic bookkeeping table (default `"alembic_version"`).
- `compare_type` — pass through to Alembic's autogenerate (default `True`).
- `compare_server_default` — same (default `True`).

`MigrationsModule` is registered `@Global()` by default; any module can then inject
`MigrationRunner`.

## Bootstrapping the ``alembic/`` directory

Once, per project:

```bash
austial-migrate init
```

Writes an `env.py`, a `script.py.mako`, an empty `versions/` folder and a short README
at `script_location`. The generated `env.py` is a one-line delegation to
`austial.migrations.alembic_env`, so future framework-side changes to how migrations run
online/offline never require regenerating it.

Its `target_metadata` comes from `context.config.attributes["target_metadata"]` — the
runner sets this before invoking Alembic, and the CLI resolves it from the
`AUSTIAL_METADATA_MODULE` env var (see below).

## Generating and applying migrations

From inside the application (typical during local development):

```python
from austial import Injectable
from austial.migrations import MigrationRunner

@Injectable()
class SchemaAdmin:
    def __init__(self, runner: MigrationRunner):
        self._runner = runner

    def bump(self, message: str) -> str:
        return self._runner.revision(message, autogenerate=True)

    def deploy(self) -> None:
        self._runner.upgrade("head")
```

From a shell (the standard deploy path):

```bash
export AUSTIAL_DATABASE_URL="postgresql+psycopg://user:pass@localhost/mydb"
export AUSTIAL_METADATA_MODULE="src.app_module:target_metadata"

austial-migrate revision -m "add users table" --autogenerate
austial-migrate upgrade head
austial-migrate current
austial-migrate history
austial-migrate downgrade -1
```

The CLI reads its URL from, in order: an explicit `--url` flag, the
`AUSTIAL_DATABASE_URL` env var, or `sqlalchemy.url` in a local `alembic.ini`. For
autogenerate, the CLI additionally imports `target_metadata` from the dotted path in
`AUSTIAL_METADATA_MODULE` (`"my_pkg.module:attribute_name"`); every other subcommand
works without it.

## `env.py` template

The generated `env.py` is deliberately short — the framework owns everything below the
`alembic_env(...)` call:

```python
from __future__ import annotations

from logging.config import fileConfig

from alembic import context

from austial.migrations import alembic_env

target_metadata = context.config.attributes.get("target_metadata")

if context.config.config_file_name is not None:
    try:
        fileConfig(context.config.config_file_name)
    except Exception:
        pass

alembic_env(context.config, target_metadata)
```

If you want to write `env.py` yourself (e.g. because you already have a heavily
customised Alembic setup), just call `alembic_env(context.config, your_metadata)` at the
bottom and the framework will handle online/offline, `NullPool` engine construction, and
threading the `version_table` / `compare_type` / `compare_server_default` settings
through.

## Autogenerate caveats

Alembic's autogenerate compares SQLAlchemy metadata against the current DB and writes
its best guess into a new revision file. Well-known gaps every product should read
before trusting it blindly:

- **Rename detection**: Alembic sees a rename as a drop + create. Edit the generated
  revision to use `op.alter_column(...)`/`op.rename_table(...)` before applying.
- **CHECK constraints and server defaults**: often not inferred; the two `compare_*`
  flags help but do not close the gap entirely.
- **Postgres `ENUM` types**: create/drop is emitted, but rename/reorder isn't — hand-edit.
- **Data migrations**: never inferred. Add `op.bulk_insert(...)` or a plain `op.execute()`
  step yourself when the schema change needs backfill.

Always read the generated revision before committing it. Alembic's own documentation
[Autogenerate caveats](https://alembic.sqlalchemy.org/en/latest/autogenerate.html#what-does-autogenerate-detect-and-what-does-it-not-detect)
is the reference; nothing in `austial.migrations` changes that contract.

## Change management (ISO/IEC 27002 A.8.32)

`austial.migrations` is the schema-change control point ISO/IEC 27002 A.8.32
("Change management") expects a production system to have: every DDL change lives in a
version-controlled revision file, applied in order, reversibly, with a persisted
`alembic_version` bookmark. Pair it with the usual code-review workflow — one PR per
migration, reviewed by someone other than the author — and the audit trail A.8.32 asks
for falls out of standard git + CI, not a separate change-management system.

Never edit an already-applied revision file. Add a new revision that fixes it forward;
`downgrade` is a rollback tool for staging or a fresh incident, not a way to rewrite
history in place.

---

Source: https://docs.austial.com/techniques/validation

# Validation

Austial's validation story leans entirely on FastAPI/pydantic — DTOs are
just pydantic `BaseModel` classes, type-hinted as a handler's `= Body()`
parameter, exactly the shape you'd already reach for in plain FastAPI.

## Defining a DTO

```python
from pydantic import BaseModel

class CreateCatDto(BaseModel):
    name: str

class UpdateCatDto(BaseModel):
    name: str | None = None
```

```python
@Post()
async def create(self, dto: CreateCatDto = Body()) -> Cat:
    return self.service.create(dto)
```

FastAPI parses and validates the incoming JSON body against `CreateCatDto`
before your handler is even called — a request missing `name`, or with the
wrong type, never reaches `create()`.

## `ValidationPipe`

```python
from austial.common.pipes import ValidationPipe

app.use_global_pipes(ValidationPipe())
```

Registering `ValidationPipe` globally (as every `austial new` project does
by default in `src/main.py`) ensures validation failures come back in the
same error body shape as every other exception in the app:

```json
{
  "statusCode": 400,
  "message": ["field required: name"],
  "error": "Bad Request"
}
```

See [Pipes](/overview/pipes) for the full `PipeTransform` reference if you
need custom transformation/validation logic beyond what pydantic gives you
for free.

## Nested and reused DTOs

Because DTOs are plain pydantic models, everything pydantic supports —
nested models, `Field(...)` constraints, custom validators, `model_config`
— works exactly as it does in a plain FastAPI app:

```python
from pydantic import BaseModel, Field

class Address(BaseModel):
    street: str
    city: str

class CreateCustomerDto(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    address: Address
```

There's no separate Austial-specific validation DSL to learn on top of
this — the framework's job stops at wiring pydantic's own validation into
the consistent error response format.

---

Source: https://docs.austial.com/techniques/authentication

# Authentication

`austial.auth` provides authentication primitives — password hashing, JWT
access tokens, TOTP for multi-factor authentication, a refresh-token store
contract, and an `AuthGuard` that verifies bearer tokens on every request.

The framework does not define your user model or the persistence layer.
Instead it gives you strategies and a guard; you wire them together against
your own user entity. This mirrors the `@nestjs/passport` + `@nestjs/jwt`
split.

## Setup

```python
from austial import Module
from austial.auth import AuthConfig, AuthModule

@Module(
    imports=[
        AuthModule.for_root(
            AuthConfig(
                jwt_secret="a-real-secret-at-least-32-characters",
                issuer="fundos",
            )
        ),
    ],
)
class AppModule:
    pass
```

`AuthConfig` accepts:

- `jwt_secret` — HS256 signing secret (at least 32 characters; the strategy
  fails at startup otherwise, per OWASP ASVS V6).
- `issuer` — the `iss` claim; typically your app name.
- `algorithm` — `"HS256"` (default) or `"RS256"`.
- `access_ttl` — access-token lifetime (default 15 minutes).
- `refresh_ttl` — refresh-token lifetime (default 14 days).
- `totp_digits`, `totp_window` — TOTP configuration.

`AuthModule` is registered `@Global()` by default so any module can inject
`JwtStrategy`, `TotpStrategy`, `PasswordHasher`, `AuthService` and
`AuthGuard` without a re-import.

## Password hashing (Argon2id)

```python
from austial import Injectable
from austial.auth import Argon2Hasher

@Injectable()
class UsersService:
    def __init__(self, hasher: Argon2Hasher):
        self._hasher = hasher

    def register(self, password: str) -> str:
        return self._hasher.hash(password)  # PHC-string, safe to persist

    def check(self, password: str, stored: str) -> None:
        self._hasher.verify(password, stored)  # raises PasswordVerificationError
        if self._hasher.needs_rehash(stored):
            # opportunistically upgrade on next successful login
            ...
```

Defaults are calibrated to OWASP's 2024 interactive-login profile
(≥19 MiB memory, 2 iterations, 1 lane). Override in the constructor if
your infrastructure allows heavier parameters.

## JWT access tokens

```python
from austial.auth import JwtStrategy

def sign_login(jwt: JwtStrategy, user_id: str, tenant_id: str) -> str:
    return jwt.sign(user_id, claims={"tenant": tenant_id, "roles": ["ops"]})

def whoami(jwt: JwtStrategy, token: str) -> str:
    payload = jwt.verify(token)   # raises TokenExpiredError / InvalidTokenError
    return payload.subject
```

Tokens carry `iss`, `iat`, `exp` and `jti` in addition to your application
claims. The `jti` gives you a hook for per-token revocation without changing
the wire format.

## TOTP multi-factor authentication

```python
from austial.auth import TotpStrategy

def enrol(totp: TotpStrategy, account: str) -> tuple[str, str, list[str]]:
    secret = totp.generate_secret()
    uri = totp.provisioning_uri(secret, account)  # QR-code payload
    recovery = totp.generate_recovery_codes()
    return secret, uri, recovery

def check(totp: TotpStrategy, secret: str, code: str) -> bool:
    return totp.verify(secret, code)  # one-window drift allowed
```

Persist the secret against the user record. Persist recovery codes
**hashed** (via `Argon2Hasher`) so a database leak does not compromise them.

## Refresh tokens

Refresh tokens are opaque strings (not JWTs) that the store persists so
they can be revoked. Framework ships an `InMemoryRefreshTokenStore` for
tests and local dev; product code supplies a persistent implementation
against its ORM:

```python
from austial.auth import RefreshTokenStore

# supply a persistent implementation
AuthModule.for_root(config, refresh_token_store=MyOrmRefreshTokenStore())
```

Pass a **class**, not just an instance, when the store itself needs DI (a repository, another
service): `for_root` registers it with `useClass`, so the container resolves its own constructor
dependencies once the app boots, instead of you having to build it by hand before the DI container
exists. `password_hasher` accepts the same class-or-instance shape. An instance (as above) still
works for a store with no dependencies of its own.

`AuthService` orchestrates `issue_tokens`, `refresh` and `logout`:

```python
from austial.auth import AuthService

class LoginController:
    def __init__(self, auth: AuthService, users: UsersService):
        self._auth = auth
        self._users = users

    async def login(self, dto):
        user = self._users.by_email(dto.email)
        self._auth.verify_password(dto.password, user.password_hash)
        self._auth.require_mfa(user.totp_secret, dto.totp_code)
        return await self._auth.issue_tokens(user.id, claims={"tenant": user.tenant_id})
```

Rotation on refresh is atomic (the old refresh token is revoked as the new
one is issued). `logout_everywhere(subject)` revokes every active refresh
token for a user — call it on password change, MFA re-enrolment, or an
admin action.

`AuthService.refresh`'s `claims` argument also accepts a **callable** — `(record:
RefreshTokenRecord) -> dict | Awaitable[dict]` — instead of a fixed dict, for when the new access
token's claims depend on the rotated token itself (its `session_id`, or the subject's *current*
tenant/roles, which may have changed since the refresh token was issued):

```python
async def claims_for(record):
    session = await sessions.get_active(record.session_id)
    membership = await memberships.find_one_by({"tenant_id": session.tenant_id, "user_id": record.subject})
    return {"tenant": session.tenant_id, "roles": [membership.role], "sid": session.id}

pair = await auth.refresh(opaque_refresh_token, claims=claims_for)
```

A `PasswordVerificationError`/`AuthenticationError` (or anything else) raised inside the callable
propagates to the caller of `refresh()` as-is.

## Guarding routes

```python
from austial import Controller, Get, UseGuards
from austial.auth import AuthGuard, CurrentUser, Public

@Controller("/me")
@UseGuards(AuthGuard)
class MeController:
    @Get()
    def me(self, user=CurrentUser()):
        return {"id": user.subject, "claims": user.claims}

    @Public()
    @Get("/robots.txt")
    def robots(self):
        return "User-agent: *\nDisallow:"
```

`AuthGuard` reads the `Authorization: Bearer <jwt>` header, verifies it with
`JwtStrategy`, and populates `request.state.user` with an `AuthPrincipal`.
`CurrentUser()` retrieves the principal in a handler parameter — resolved
*after* every guard for the route has run (including a global one registered
via `app.use_global_guards(AuthGuard, ...)`, which is the common case: nothing
above requires `@UseGuards(AuthGuard)` on every controller). `@Public()` skips
the guard for a specific route or an entire controller, and `CurrentUser()`
then resolves to `None`.

Under the hood `CurrentUser()` is `austial.common.FromRequestState("user")`: a
handler parameter bound to `request.state.<attribute>`, read only once the
router's guard loop has finished (a plain FastAPI `Depends` would resolve
before that loop runs, over an empty `request.state`, however the guard was
registered). Any guard that publishes something onto `request.state` — a
product `TenantGuard` subclass, a session guard — can use the same helper for
its own param:

```python
from austial.common import FromRequestState

def CurrentTenant() -> Any:
    return FromRequestState("tenant", default=None)
```

To layer additional checks (session freshness, per-tenant claims) subclass
`AuthGuard` and override `after_verify`.

## Errors

All auth failures extend `UnauthorizedException` (mapping to `401`), with
sub-types for programmatic handling:

- `AuthenticationError` — generic parent.
- `InvalidTokenError` — malformed, tampered, wrong issuer.
- `TokenExpiredError` — `exp` in the past.
- `MfaRequiredError` — password OK but MFA still pending.

Never surface which failure occurred to the client verbatim; log the sub-type
and return a generic `401` message to callers.

## Security posture (OWASP ASVS L2)

- V2 auth: Argon2id with OWASP-2024 parameters; password verification runs
  in constant time via `argon2-cffi`.
- V3 session: short-lived (15-minute) access tokens, opaque rotating refresh
  tokens with server-side revocation and per-user bulk revoke.
- V6 crypto: minimum 32-character HS256 secret enforced at startup; RS256
  is a one-line change.
- V13 API: bearer authentication only; guard verifies before every handler
  unless explicitly `@Public`.

---

Source: https://docs.austial.com/techniques/authorization

# Authorization

`austial.authz` provides role- and permission-based access control on top of
`austial.auth`. The framework enforces "does this principal have the claim?";
product code decides "what roles does this user have?" by populating the
JWT's `roles` and `permissions` claims at login time.

## Setup

```python
from austial import Module
from austial.authz import AuthzModule

@Module(imports=[AuthzModule.for_root()])
class AppModule:
    pass
```

Registered `@Global()` by default -- `RolesGuard` and `PermissionService`
are injectable everywhere.

## Guarding routes

```python
from austial import Controller, Delete, Get, UseGuards
from austial.auth import AuthGuard
from austial.authz import AuthorizeObject, Permissions, Roles, RolesGuard

def _owns_invoice(principal, request) -> bool:
    return request.path_params.get("owner_id") == principal.subject

@Controller("/invoices")
@UseGuards(AuthGuard, RolesGuard)
class InvoicesController:
    @Roles("ops", "admin")
    @Get()
    def list(self):
        return {"invoices": []}

    @Permissions("write:invoice")
    @Get("/:id")
    def get(self):
        return {"id": ...}

    @Roles("admin")
    @AuthorizeObject(_owns_invoice)
    @Delete("/:id")
    def delete(self):
        return {"deleted": True}
```

Semantics (all checks pass = 200; any check fails = 403):

- `@Roles(*names)` — caller must hold **any** of the listed roles.
- `@Permissions(*names)` — caller must hold **all** of the listed permissions.
- `@AuthorizeObject(rule)` — `rule(principal, request) -> bool` runs last;
  returning `False` (or raising) rejects the request. Use for object-level
  checks (ownership, tenant match, status).

Class-level and handler-level metadata are unioned — a class-level `@Roles`
is added to whatever the handler declares.

### Deny-by-default

A handler that declares none of `@Roles`, `@Permissions` or `@AuthorizeObject` is **allowed** by
the stock guard, so a forgotten decorator is silently open to every authenticated caller. To make a
policy mandatory, subclass and set `deny_undeclared = True`; an undeclared handler is then a 403.
Let `@Public` routes through before delegating:

```python
from austial.auth.decorators import is_public

class StrictRolesGuard(RolesGuard):
    deny_undeclared = True

    async def can_activate(self, context):
        if is_public(context.get_handler()) or is_public(context.get_class()):
            return True
        return await super().can_activate(context)
```

## Programmatic checks

```python
from austial.authz import PermissionService

class ReportsService:
    def export(self, principal) -> bytes:
        if not PermissionService.has_all_permissions(principal, ("read:report",)):
            raise PermissionError("nope")
        ...
```

Useful in services, background jobs and event handlers where decorators
don't apply.

## Errors

- `InsufficientRolesError` — required roles absent (403).
- `InsufficientPermissionsError` — required permissions absent (403).
- `ForbiddenException` — object rule denied, or metadata present without a
  principal.

Every error carries the requested vs actual claims so log observability
(never response bodies) can pinpoint the mismatch.

## Populating claims at login

Product code decides what roles/permissions a user has and includes them in
the JWT at issue time:

```python
tokens = await auth.issue_tokens(
    user.id,
    claims={
        "tenant": user.tenant_id,
        "roles": user.role_names,
        "permissions": user.permission_names,
    },
)
```

Keep the list bounded (dozens, not thousands) — a bloated JWT hurts every
request. If permissions are large or fine-grained, put an opaque
`session_id` in the claims and look permissions up server-side per request.

## Security posture (OWASP API Security Top 10)

- **API1 broken object-level authorisation**: `@AuthorizeObject(rule)` is
  the first-class primitive for the ownership/tenant check that global
  role checks cannot express.
- **API3 broken object-property-level authorisation**: keep response DTOs
  narrow (see `austial.common.pipes`); RolesGuard cannot substitute for
  field-level filtering.
- **API5 broken function-level authorisation**: every mutating handler must
  carry either `@Roles`/`@Permissions` or `@Public`. A code review that
  spots a handler without either is a bug report.

---

Source: https://docs.austial.com/techniques/multi-tenancy

# Multi-tenancy

`austial.tenancy` scopes every request to a tenant. It provides a
request-scoped `TenantContext`, a `TenantGuard` that populates it from the
JWT's `tenant` claim, a `Tenanted` entity mixin that adds a `tenant_id`
column, and a `@RequireTenant` decorator that marks routes as tenant-only.

The tenant boundary is enforced at the perimeter (the guard); the persistence
boundary stays explicit — product code writes `.where(tenant_id=current)`
in queries. That keeps the filter visible in every query plan review rather
than magically injected.

## Setup

```python
from austial import Module
from austial.tenancy import TenancyModule

@Module(imports=[TenancyModule.for_root()])
class AppModule:
    pass
```

`TenancyModule` is registered `@Global()` by default so `TenantContext` and
`TenantGuard` are injectable everywhere.

## Reading the current tenant

```python
from austial import Injectable
from austial.tenancy import TenantContext

@Injectable()
class InvoicesService:
    def __init__(self, ctx: TenantContext):
        self._ctx = ctx

    def list_mine(self):
        tenant_id = self._ctx.current_id()      # raises if unset
        # ... query with .where(tenant_id=tenant_id)
```

`TenantContext.get()` returns `TenantContextValue | None`;
`TenantContext.require()` raises `MissingTenantError` (403) when unset. Use
`require` in code paths where the tenant must be present.

The class methods are useful outside a DI context (mixins, pure functions,
background jobs holding a snapshot):

```python
from austial.tenancy import TenantContext, TenantContextValue

token = TenantContext.set(TenantContextValue(tenant_id="acme"))
try:
    ...  # code runs with tenant="acme"
finally:
    TenantContext.reset(token)
```

Contextvars are used under the hood, so the value is isolated across `await`
points and threadpool executors.

## Guarding routes

```python
from austial import Controller, Get, UseGuards
from austial.auth import AuthGuard
from austial.tenancy import RequireTenant, TenantGuard

@Controller("/invoices")
@UseGuards(AuthGuard, TenantGuard)
@RequireTenant()
class InvoicesController:
    @Get()
    def list(self, ctx: TenantContext):
        return {"tenant": ctx.current_id()}
```

`TenantGuard` runs after `AuthGuard`. It reads the `tenant` claim from the
authenticated principal, populates `TenantContext`, and stores a reset token
on `request.state` for teardown.

- On success the tenant id is also published on `request.state.tenant`, which
  [`@Audited`](/techniques/audit-log) reads, so audit rows carry the tenant.
- Handlers or controllers with `@RequireTenant()` and no claim: 403
  (`MissingTenantError`).
- Path parameter `tenant_id` disagrees with the claim: 403
  (`TenantMismatchError`).

Handlers without `@RequireTenant` and without a claim pass through — this
lets `@Public` routes coexist in the same controller when needed.

### Deny-by-default

For a product where every non-public route is tenant-scoped, subclass the guard and set
`require_by_default = True`, so a forgotten `@RequireTenant()` cannot ship a route that runs without
a tenant. Let your public routes through *before* delegating (the tenancy package knows nothing about
`@Public`):

```python
from austial.auth.decorators import is_public

class AppTenantGuard(TenantGuard):
    require_by_default = True

    async def can_activate(self, context):
        if is_public(context.get_handler()) or is_public(context.get_class()):
            return True
        return await super().can_activate(context)
```

## Tenanted entities

```python
from austial.orm import Entity, PrimaryGeneratedColumn, Column
from austial.tenancy import Tenanted

@Entity("invoices")
class Invoice(Tenanted):
    id: int = PrimaryGeneratedColumn()
    number: str = Column(nullable=False)
    total_cents: int = Column(nullable=False)
```

The mixin adds `tenant_id: str` (indexed, non-nullable). Set it explicitly
at insert from `TenantContext.require().tenant_id` so a row without a tenant
is a hard write error rather than a silent misfile.

**mypy caveat:** `dataclass_transform` (what makes `@Entity()` type-check a
synthesized `__init__`) only recognises fields declared on a class that is
itself part of the transform chain. `Tenanted` is a plain mixin, not
`@Entity()`-decorated, so mypy does not see `tenant_id` as a constructor
parameter of the entity that inherits it — passing `tenant_id=...` type-checks
as an "unexpected keyword argument" even though it works correctly at
runtime. If strict mypy on entity construction matters more than reusing the
mixin, declare the column directly on the entity instead (a one-line
`Column(index=True, default=None)`, same shape `Tenanted` uses internally).

## Errors

- `MissingTenantError` — the tenant context is absent when required. Maps to 403.
- `TenantMismatchError` — a path/body tenant disagrees with the claim. Maps to 403.

Both extend `ForbiddenException` so the framework's exception dispatch renders
them without extra work; product filters can inspect the sub-type if they want
a domain-specific response body.

## Security posture (OWASP API Security Top 10)

- **API1 broken object-level authorisation**: every tenant-scoped query
  filters by `tenant_id` (product code); the guard guarantees the value.
- **API5 broken function-level authorisation**: `@RequireTenant` marks routes
  that must have a tenant; the guard fails-closed on absence.
- Guard runs after `AuthGuard`, so a missing/invalid JWT rejects the request
  before this guard runs — the tenant claim is trusted only when its
  container has been verified.

---

Source: https://docs.austial.com/techniques/security

# Security

`austial.security` bundles the baseline defences every fintech backend needs
before it ships: hardened response headers, per-route rate limiting,
idempotency-key replay, double-submit-cookie CSRF, structured PII redaction
for logs, and app-level column encryption. Each primitive is a small,
injectable class the DI container wires for you; the module is registered
`@Global()` so any feature module can pull the piece it needs.

The primitives map cleanly onto OWASP ASVS V5 (validation), V6 (secrets and
cryptography), V8 (data protection), V9 (transport) and OWASP API Top-10
API4 (unrestricted resource consumption), API6 (unrestricted access to
business flows) and API8 (security misconfiguration).

## Setup

```python
from austial import Module
from austial.security import SecurityConfig, SecurityModule

@Module(
    imports=[
        SecurityModule.for_root(
            SecurityConfig(
                csp="default-src 'self'; frame-ancestors 'none'",
                encryption_keys={"2026-09": _load_kek("2026-09")},
                active_encryption_key="2026-09",
            ),
        ),
    ],
)
class AppModule:
    pass
```

`SecurityConfig` accepts:

- `csp` — Content-Security-Policy header value. Default denies everything but
  same-origin and blocks framing.
- `hsts_max_age` — `Strict-Transport-Security` `max-age` (default 1 year).
- `permissions_policy` — Permissions-Policy value; default disables
  geolocation, microphone and camera.
- `encryption_keys` — `{version: 32-byte KEK}` for the `EnvelopeEncryptor`.
- `active_encryption_key` — the version to encrypt new writes with; older
  versions stay readable.
- `default_rate_limit` — `(limit, period_seconds)` used when a route has no
  `@RateLimit` override. Default `(60, 60)`.
- `trusted_proxy_hops` — how many reverse proxies you operate in front of the app. Default `0`:
  `X-Forwarded-For` is ignored and the socket peer is the client. See
  [Rate limiting](#rate-limiting).
- `idempotency_ttl_seconds` — how long an `Idempotency-Key` record is kept.

## Secure response headers

Mount `SecureHeadersMiddleware` module-wide to stamp defensive headers on
every response:

```python
from austial import MiddlewareConsumer
from austial.security import SecureHeadersMiddleware

class AppModule:
    def configure(self, consumer: MiddlewareConsumer) -> None:
        consumer.apply(SecureHeadersMiddleware).for_routes("*")
```

Headers set (each `setdefault`, so a handler can still override for a specific
response):

- `Content-Security-Policy`
- `Strict-Transport-Security: max-age=…; includeSubDomains`
- `X-Content-Type-Options: nosniff`
- `Referrer-Policy: strict-origin-when-cross-origin`
- `X-Frame-Options: DENY`
- `Permissions-Policy`

## Rate limiting

`RateLimitGuard` is a fixed-window limiter keyed on `(client_ip, method, path)`.
The budget is `SecurityConfig.default_rate_limit` unless a route or its
controller carries `@RateLimit(limit, period_seconds)`:

```python
from austial import Controller, Post, UseGuards
from austial.security import RateLimit, RateLimitGuard

@Controller("/auth")
@UseGuards(RateLimitGuard)
class AuthController:
    @Post("/login")
    @RateLimit(5, 60)  # 5 attempts per minute
    async def login(self, body: LoginRequest) -> LoginResponse:
        ...
```

The client IP is the socket peer. `X-Forwarded-For` is client-controlled, so it is ignored unless you
set `SecurityConfig.trusted_proxy_hops = n`; the client is then the `n`-th entry counted from the
**right** (each trusted proxy appends the address it saw, so entries to the left of those can be
forged). If the header has fewer than `n` entries it did not come through your proxy and is ignored.
Trusting the left-most entry would let an attacker mint a fresh bucket per request.

The backing store is a `RateLimitStore` protocol; the default
`InMemoryRateLimitStore` is process-local. For multi-worker deployments
supply a Redis-backed store via `SecurityModule.for_root(config,
rate_limit_store=RedisStore())`.

On a breach the guard raises `RateLimitExceededError` (HTTP 429) with a
`retry_after` attribute your exception filter can turn into a `Retry-After`
response header.

To key on something other than IP + route (session id, api key, tenant id)
subclass `RateLimitGuard` and override `build_key(request)`.

## Idempotency

`IdempotencyKeyGuard` enforces the semantics every payments API needs: a
retried `POST`/`PUT`/`PATCH`/`DELETE` with the same `Idempotency-Key` header
replays the earlier response instead of executing the mutation twice.

```python
from austial import Controller, Post, UseGuards
from austial.security import IdempotencyKeyGuard, IdempotencyRecord, IdempotencyStore

@Controller("/payments")
@UseGuards(IdempotencyKeyGuard)
class PaymentsController:
    def __init__(self, store: IdempotencyStore):
        self._store = store

    @Post("/")
    async def create(self, request, body: CreatePayment) -> Payment:
        # If the guard flagged a replay, echo the cached body.
        cached = getattr(request.state, "idempotent_replay", None)
        if cached is not None:
            return cached.body

        payment = await self._service.create(body)
        self._store.put(
            IdempotencyRecord(
                key=request.state.idempotency_key,
                request_hash=request.state.idempotency_request_hash,
                status_code=201,
                body=payment,
                stored_at=time.time(),
            )
        )
        return payment
```

Behaviour:

- Missing `Idempotency-Key` — pass through (product code may still require it
  via a validation pipe).
- New key — populate `request.state.idempotency_key` and
  `request.state.idempotency_request_hash` so the handler can persist the
  outcome.
- Same key + same body hash within TTL — flag `request.state.idempotent_replay`
  so the handler returns the cached response.
- Same key + different body — raise `IdempotencyConflictError` (HTTP 409).

The default `InMemoryIdempotencyStore` keeps records for
`SecurityConfig.idempotency_ttl_seconds` (24 hours by default); swap it for a
persistent store in production.

## CSRF (double-submit cookie)

For cookie-authenticated flows attach `CsrfInterceptor` to any controller
that mutates state. The interceptor compares the `X-CSRF-Token` header
against the `csrf_token` cookie in constant time.

```python
from austial import Controller, Post, UseInterceptors
from austial.security import CsrfInterceptor

@Controller("/settings")
@UseInterceptors(CsrfInterceptor)
class SettingsController:
    @Post("/notifications")
    async def update(self, body: NotificationsUpdate) -> None:
        ...
```

Read-only methods (`GET`, `HEAD`, `OPTIONS`) skip the check. A request that
carries `Authorization: Bearer …` also skips the check — a bearer token that
an attacker cannot possess is a stronger primitive than a CSRF cookie. A
missing or mismatched token raises `ForbiddenException` (HTTP 403).

Rotate the cookie value at login and again on every privilege change; the
interceptor is only the read-side check.

## PII redaction

`PIIRedactor` masks common PII patterns before you hand a record to a logger
or an audit sink. It recognises emails, international phone numbers, Indian
PAN, Aadhaar and passport numbers, IBANs (mod-97 verified) and card numbers
(Luhn verified).

```python
from austial import Injectable
from austial.security import PIIRedactor

@Injectable()
class AuditLogger:
    def __init__(self, redactor: PIIRedactor, sink: LogSink):
        self._redactor = redactor
        self._sink = sink

    def record(self, event: dict) -> None:
        self._sink.emit(self._redactor.redact(event))
```

`redact` accepts a string, dict or list and recurses; non-matching text
passes through unchanged. Values that fail their check (Luhn, mod-97) are
also passed through, so log spam like `1234 5678 9012 3456` is not falsely
redacted.

## Envelope encryption

`EnvelopeEncryptor` gives you app-level column encryption with AES-256-GCM
and an embedded key version, so a rotation swaps `active_encryption_key`
without a bulk re-encrypt — old rows keep decrypting under their historical
version.

```python
from austial import Injectable
from austial.security import EnvelopeEncryptor

@Injectable()
class VaultService:
    def __init__(self, enc: EnvelopeEncryptor):
        self._enc = enc

    def store(self, account_number: str) -> str:
        return self._enc.encrypt(account_number)

    def read(self, ciphertext: str) -> str:
        return self._enc.decrypt(ciphertext)
```

The wire format is base64 of `b"v1:<version>:" || nonce (12 B) ||
ciphertext || tag (16 B)`. Any of the following raises
`EncryptionKeyError`:

- A key that is not exactly 32 bytes.
- An `active_encryption_key` that is missing from `encryption_keys`.
- A ciphertext whose embedded version is not in the current key map.
- A ciphertext whose GCM tag does not authenticate (tampered).

Rotation flow:

1. Add the new version to `encryption_keys`.
2. Set `active_encryption_key` to the new version.
3. Deploy. New writes use the new key; existing rows keep decrypting under
   their historical version.
4. Optionally re-encrypt lazily on read-then-write, or run a background job
   to migrate rows in bulk.

Never remove a historical key that any row still references.

## OWASP mapping

| Primitive | ASVS | API Top-10 |
| --- | --- | --- |
| `SecureHeadersMiddleware` | V9 (transport), V14 (config) | API8 (security misconfiguration) |
| `RateLimitGuard` | V11 (rate limits) | API4 (unrestricted resource consumption) |
| `IdempotencyKeyGuard` | V11 (business logic) | API6 (unrestricted access to business flows) |
| `CsrfInterceptor` | V13 (API/web) | API8 |
| `PIIRedactor` | V8 (data protection) | — |
| `EnvelopeEncryptor` | V6 (cryptography), V8 (data protection) | — |

The framework never claims a checkbox on its own — you still need a WAF, TLS
termination, secret rotation, and a monitoring pipeline — but these
primitives keep the classes of bug they cover from surfacing in application
code.

---

Source: https://docs.austial.com/techniques/audit-log

# Audit log

`austial.audit` ships a WORM (write-once, read-many) audit log: an
append-only `AuditLogEntry` persisted through `austial.orm`, an injectable
`AuditLogService`, and an `@Audited(...)` interceptor that records a
success/failure event around a controller handler.

The framework guarantees three things:

1. Recorded events are immutable in memory (`AuditEvent` is a frozen
   dataclass) and append-only in the database (`AuditLogEntry` is
   `@Entity(append_only=True)`, so the entity manager rejects any
   `update()`/`delete()`).
2. Retention-driven eviction (`evict_older_than`) honours per-record
   `legal_hold` flags — a record under legal hold is never removed, even
   when it is past its retention cutoff.
3. Default retention is 1,825 days (5 years), which matches the minimum
   record-retention rule in the IFSCA AML/CFT Guidelines (rule 10.1).
   Product code overrides it upward for stricter jurisdictions; the
   framework never silently trims below the default.

Product code supplies the storage and the SIEM forwarding — the framework
does not decide whether events go to Postgres, S3 Object Lock, Splunk HEC,
or all three at once.

## Setup

```python
from austial import Module
from austial.audit import AuditConfig, AuditModule

@Module(
    imports=[
        AuditModule.for_root(
            AuditConfig(retention_days=1825),  # IFSCA AML rule 10.1
        ),
    ],
)
class AppModule:
    pass
```

`AuditModule` is registered `@Global()` by default so `AuditLogService`
and `AuditedInterceptor` are injectable everywhere without a re-import.

`AuditModule.for_root(...)` accepts a `store=` override — pass an
`OrmAuditStore(repo)` in production, or a SIEM-backed store you write. The
default is `InMemoryAuditStore`, which is fine for tests and unsafe for
production because it does not survive a process restart.

## Recording an event

```python
from austial import Injectable
from austial.audit import AuditEvent, AuditLogService

@Injectable()
class KycService:
    def __init__(self, audit: AuditLogService):
        self._audit = audit

    async def deny(self, case_id: str, reason: str, actor: str) -> None:
        # ... do the work ...
        await self._audit.record(
            AuditEvent(
                action="kyc.deny",
                resource_type="Case",
                resource_id=case_id,
                actor=actor,
                outcome="success",
                metadata={"reason": reason},
            )
        )
```

`AuditEvent` is a frozen dataclass — `id` is a server-generated UUIDv4
and `timestamp` defaults to `datetime.now(UTC)`. Once you have handed the
event to `record`, no code path can mutate it.

The `AuditLogEntry.timestamp` column is `TIMESTAMP WITH TIME ZONE`, matching the timezone-aware
`AuditEvent.timestamp`; generate your migration from the entity so the column is created that way.

Do not put PII, tokens, or secrets in `metadata`. The audit log is
long-lived by design; anything that lands in it lives for the full
retention window and often ends up in an external SIEM.

## `@Audited` on a controller handler

Most audit coverage does not need to run inside services — it belongs on
the request boundary, where the actor, tenant, and request id are already
known. Use the interceptor for that.

```python
from austial import Controller, Post, UseInterceptors
from austial.audit import Audited, AuditedInterceptor

@Controller("/funds")
@UseInterceptors(AuditedInterceptor)
class FundsController:
    @Post()
    @Audited(
        "funds.create",
        resource_type="Fund",
        resource_id_from="result.id",
    )
    def create(self, dto):
        # ... returns {"id": "F42", "name": ...}
        ...

    @Delete(":id")
    @Audited(
        "funds.delete",
        resource_type="Fund",
        resource_id_from="path:id",
    )
    def delete(self, id: str):
        ...
```

`resource_id_from` resolves the resource identifier:

- `"path:<name>"` reads `request.path_params[<name>]`.
- `"result.<key>"` (dotted for nested lookups) reaches into the response
  after the handler has returned successfully.
- `None` (the default) falls back to
  `request.state.audit_resource_id` if product code has set it.

The interceptor reads `request.state.user`, `request.state.tenant` (when
`austial.tenancy` is wired), and `request.state.request_id` (typically
set by `LoggingMiddleware`), so the actor/tenant/correlation-id chain is
consistent with the rest of the request pipeline.

On success it records `outcome="success"` after the handler returns; on
exception it records `outcome="failure"`, puts the exception class name
and message in `metadata`, then re-raises so the framework's exception
filter runs unchanged.

Pass `capture_result=True` when the response is small and auditable
(e.g. a KYC decision object). Do not enable it on responses containing
customer PII, tokens, or full payment payloads.

The captured value is run through the same `jsonable_encoder` FastAPI uses for response bodies
before it is stored, so a handler returning a pydantic response DTO (the normal case) or a value
containing a `datetime` is encoded to plain JSON-safe primitives -- not stored as the raw object,
which would fail with a driver-level serialization error against `OrmAuditStore`'s `metadata_json`
(a real `JSON` column) once persisted, even though it round-trips silently through the in-memory
store's plain Python dict.

## Persisting the log

The default store keeps events in memory. In production, wire the
ORM-backed store against the `AuditLogEntry` entity:

```python
from austial.audit import AuditLogEntry, AuditModule, OrmAuditStore
from austial.orm import InjectRepository, OrmModule, Repository

@Module(
    imports=[
        OrmModule.for_root(...),
        OrmModule.for_feature([AuditLogEntry]),
    ],
)
class PersistenceModule:
    pass

@Injectable()
class _AuditStoreFactory:
    def __init__(self, repo: Repository[AuditLogEntry] = InjectRepository(AuditLogEntry)):
        self.store = OrmAuditStore(repo)
```

`OrmAuditStore.record()` writes with `Repository.insert()` (the event's UUID is the primary key, so
`save()` would be an UPDATE and the append-only guard would reject it).

`AuditLogEntry` is declared `@Entity("audit_log", append_only=True)` —
`EntityManager` raises `AppendOnlyViolationError` on any `update()`,
`delete()`, or `remove()` against it. In production, revoke `UPDATE` and
`DELETE` on the underlying table at the database role level too; the
framework guard is a defence in depth, not a substitute for
least-privilege.

## Querying

`AuditLogService.query(...)` and the underlying `AuditStore.query(...)` accept
`subject`/`actor` (aliases for the same filter), `action`, `tenant_id`, `since`,
`until`, and `limit`. Multi-tenant products exposing a query endpoint (e.g. a
platform-operator console) must always pass `tenant_id` when the caller is
scoped to one tenant — omitting it returns events across every tenant, since
`tenant_id=None` means "no filter", not "current tenant":

```python
# a tenant's own compliance officer: only their tenant's events
await audit.query(tenant_id=current_tenant_id)

# a cross-tenant operator role: an explicit, caller-chosen tenant
await audit.query(tenant_id=requested_tenant_id)
```

## Retention and legal hold

`AuditLogService.evict_older_than(cutoff)` removes events older than
`cutoff` from the store. `evict_expired()` is a convenience wrapper that
uses `AuditConfig.retention_days`:

```python
# nightly job, wired via austial.scheduling or a cron in your infra
async def prune(audit: AuditLogService) -> None:
    evicted = await audit.evict_expired()
    ...
```

Both methods honour `legal_hold`. In the in-memory store,
`set_legal_hold(event_id, hold=True)` flags a specific event; in the ORM
store, the `AuditLogEntry.legal_hold` column carries the flag and admin
processes toggle it out of band (`AuditLogEntry` itself is append-only,
so the flip lives outside the ORM path — typically a `pg_amend` role or a
manual SQL migration).

The `OrmAuditStore.evict_older_than` implementation returns `0` — an
append-only ORM entity cannot be deleted through the ORM. Production
deployments that need retention-driven eviction handle it at the database
layer, usually per-partition `DROP TABLE` from a job that first checks
for active legal holds.

## Errors

- `AuditError` — base class (500 by default).
- `AuditStoreUnavailableError` — the store is unreachable (503).
- `LegalHoldViolationError` — strict-mode eviction refused to remove a
  held record (409). Not raised by the default flow, which silently
  preserves held rows.

All extend `HttpException`, so `AllExceptionsFilter` maps them without a
translation layer.

## Compliance posture

- **OWASP ASVS V7 (Logging and Error Handling)** — V7.1 (log security
  events), V7.2 (structured, tamper-resistant), V7.3 (no secrets in
  logs). The framework satisfies the append-only + structured-event
  guarantees; keeping secrets out is on the caller — do not put tokens,
  PANs, or full customer records in `metadata`.
- **ISO/IEC 27001 Annex A.5.24 (Information security incident planning
  and preparation)** — audit trails are the evidentiary basis for
  incident triage; the WORM guarantee here is what makes them
  admissible.
- **ISO/IEC 27001 Annex A.8.15 (Logging)** — retained for the required
  period (default 5 years), protected against tampering (append-only
  entity, DB role separation), and reviewable via
  `AuditLogService.query(...)`.
- **IFSCA AML/CFT Guidelines rule 10.1** — records retained for a
  minimum of five years, matched by the default retention window.

---

Source: https://docs.austial.com/techniques/background-jobs

# Background jobs

`austial.jobs` runs work off the request path. It provides a `JobQueue`
(enqueue / cancel / status), a `Worker` (background loop with retries and
exponential backoff), `@Cron` for scheduled jobs, `@Job` for enqueueable
ones, a pluggable `JobBackend` (in-memory reference impl ships out of the
box), and a jobs-scoped `IdempotencyStore` for deduplicating enqueue calls.

The design mirrors `@nestjs/schedule` + BullMQ: the framework provides the
scheduler, queue and worker; product code declares handlers on `@Injectable`
classes. Redis / Arq / Dramatiq wiring is deferred -- the in-memory backend
is a drop-in default for tests and single-node deployments. Product code
swaps it via `JobsModule.for_root(config, backend=MyBackend())`.

## Setup

```python
from austial import Module
from austial.jobs import JobsConfig, JobsModule

@Module(imports=[JobsModule.for_root(JobsConfig())])
class AppModule:
    pass
```

`JobsModule` is registered `@Global()` by default, so `JobQueue`,
`JobRegistry` and `Worker` are injectable everywhere. `JobsConfig` has
sensible defaults (`concurrency=4`, `default_max_attempts=3`,
`default_backoff_seconds=30`, `poll_interval_seconds=1.0`) -- override per
environment.

## Declaring handlers

```python
from austial import Injectable
from austial.jobs import Cron, Job

@Injectable()
class Payments:
    @Job("send-remittance", max_attempts=5, backoff_seconds=60)
    async def send(self, *, remittance_id: str) -> None:
        # Do the work. Raise on failure -- the worker retries.
        ...

    @Cron("@daily")
    async def daily_reconciliation(self) -> None:
        ...

    @Cron("*/5 * * * *", name="poll-partner-feed", jitter_seconds=15)
    async def poll(self) -> None:
        ...
```

A single method may carry both decorators (a cron that can also be triggered
manually). The registry looks each handler up by name at dispatch time --
rename a method freely, don't rename its `@Job("...")` name once it's
enqueuable in prod.

### Schedule strings

`@Cron` accepts:

- Standard 5-field cron: `"*/5 * * * *"`, `"0 9 * * 1-5"`, ...
- Shorthand: `"@hourly"`, `"@daily"`, `"@weekly"`, `"@monthly"`, `"@yearly"`.
- Fixed offset: `"@every 30s"`, `"@every 5m"`, `"@every 2h"`.

The parser is hand-rolled -- no third-party dependency. It supports `*`,
`N`, `A-B`, `*/N` and comma-separated combos per field, which covers the
common cases in a fintech backend. A product that needs the exotic corners
of cron (`L`, `W`, `#`) can swap in croniter and rebuild `Schedule` in its
own module.

### Discovery

Handlers register with the `JobRegistry`. Two paths:

```python
# Explicit -- clearest at boot time.
registry.register_job(spec, payments.send)

# Implicit -- walks an instance's methods and picks up @Job/@Cron metadata.
registry.discover(payments)
```

Product bootstrap code calls `discover(instance)` on each injectable that
carries job handlers, right after the DI container instantiates it.

## Enqueueing

Inject `JobQueue` and call `enqueue(...)`:

```python
from datetime import datetime, timedelta

from austial import Injectable
from austial.jobs import JobQueue

@Injectable()
class RemittanceService:
    def __init__(self, queue: JobQueue):
        self._queue = queue

    async def submit(self, remittance_id: str) -> str:
        return await self._queue.enqueue(
            "send-remittance",
            payload={"remittance_id": remittance_id},
            idempotency_key=f"remittance:{remittance_id}",
        )

    async def submit_later(self, remittance_id: str) -> str:
        return await self._queue.enqueue(
            "send-remittance",
            payload={"remittance_id": remittance_id},
            run_at=datetime.utcnow() + timedelta(minutes=5),
        )
```

`enqueue` returns the job id. `run_at` schedules the first attempt in the
future. `idempotency_key` deduplicates against a jobs-scoped store: a second
call with the same key returns the id of the first without enqueuing a
duplicate. The store is independent from `austial.security`'s idempotency
store -- that one caches HTTP responses for retry replay; this one prevents
double-enqueue of the same background job.

`cancel(job_id)` transitions a `pending` job to `cancelled`; it returns
`False` if the job has already started running or finished. `status(job_id)`
returns the current `JobStatus` (`pending`, `running`, `succeeded`,
`failed`, `cancelled`) or `None` if the id is unknown.

## Running the worker

The `Worker` is an injectable, but in most deployments you run it as a
separate process using the bundled CLI:

```bash
austial-worker run src.app_module:AppModule
```

The CLI resolves the module, wires the in-memory backend, discovers every
`@Job`/`@Cron` handler on the module's providers, and enters the poll-and-
execute loop until it receives `SIGINT` / `SIGTERM`.

The CLI also has one-off commands for shell ops:

```bash
austial-worker enqueue send-remittance --payload '{"remittance_id":"r_123"}'
austial-worker status <job_id>
```

Embedding the worker in the same process (dev, tests) is one line:

```python

from austial.jobs import Worker

worker: Worker = ...  # from the DI container

asyncio.create_task(worker.run())
```

## Idempotency

Two independent forms of idempotency:

1. **Enqueue-side.** `idempotency_key` on `enqueue(...)` deduplicates
   dispatch: two callers who both want to send remittance `r_123` will
   agree on one job id. Suitable for user-triggered actions (submit
   button, webhook receiver, retry after a network glitch).

2. **Handler-side.** The `idempotent=True` flag on `@Job` is a hint that
   the handler is safe to retry -- important when the worker retries a
   partial failure. Product code guarantees this with unique keys on the
   underlying tables (e.g. an `INSERT ... ON CONFLICT DO NOTHING` on a
   business-key column, or a "already sent?" check at the top of the
   handler).

The framework won't stop a non-idempotent handler from being retried on
failure. `idempotent=False` is a documentation aid; a genuinely non-
retryable handler should also set `max_attempts=1`.

## Error handling and retries

An exception in a `@Job` handler is caught by the worker. If the attempt
count is still below `max_attempts`, the job is scheduled for another try
at `now + backoff * 2^(attempts-1)` seconds (exponential backoff). Once
retries are exhausted, the record is marked `failed` and its `error` field
carries the `repr()` of the last exception.

```python
@Job("charge-card", max_attempts=5, backoff_seconds=15)
async def charge(self, *, order_id: str) -> None:
    # first retry after 15s, then 30s, 60s, 120s (exp base=15)
    ...
```

The base backoff can be set globally on `JobsConfig.default_backoff_seconds`
and per-handler on `@Job(backoff_seconds=...)`. Both are honoured; the
per-handler value wins.

## Custom backends

`JobBackend` is a `Protocol` with five async methods (`enqueue`, `pop_due`,
`mark`, `cancel`, `get`). A Redis-backed implementation would use `LPUSH`
+ `BRPOPLPUSH` on a sorted set keyed by `next_attempt_at`; a Postgres one
would use `SELECT ... FOR UPDATE SKIP LOCKED` inside a transaction. Swap
it in at module import:

```python
from austial.jobs import JobsModule
from my_app.jobs import PostgresJobBackend

JobsModule.for_root(
    JobsConfig(concurrency=32),
    backend=PostgresJobBackend(engine),
)
```

The rest of the module (queue, worker, registry, decorators) is unchanged.

---

Source: https://docs.austial.com/techniques/file-storage

# File storage

`austial.storage` gives every request a uniform way to read and write blobs
— user uploads, generated PDFs, exports, KYC documents. It ships a driver
contract, two backends (local filesystem for dev, S3 for prod), a service
that layers path-safety, size limits, content-type inference and pluggable
virus scanning on top of the driver, and a `SignedUrlService` that mints
short-lived HMAC-signed download URLs.

The framework does not define the HTTP surface. Product code writes a
`StorageController` that accepts uploads and streams signed downloads —
`StorageService` and `SignedUrlService` are injected into it.

## Setup

```python
from pathlib import Path

from austial import Module
from austial.storage import StorageConfig, StorageModule

@Module(
    imports=[
        StorageModule.for_root(
            StorageConfig(
                signed_url_secret="a-real-secret-at-least-32-characters",
                driver="local",
                local_base_dir=Path("./var/storage"),
            ),
        ),
    ],
)
class AppModule:
    pass
```

`StorageConfig` accepts:

- `signed_url_secret` — HMAC-SHA256 signing key (at least 32 characters;
  `StorageConfig` fails at construction otherwise, per OWASP ASVS V6).
- `driver` — `"local"` (default) or `"s3"`.
- `local_base_dir` — required when `driver="local"`; a directory the process
  can read and write.
- `s3_bucket`, `s3_region`, `s3_endpoint_url` — required when `driver="s3"`.
- `max_object_size_bytes` — refuses payloads above this size at the
  `StorageService.put` boundary (default 100 MiB).
- `allowed_content_types` — optional allow-list; empty tuple means "any".

`StorageModule` is registered `@Global()` by default so `StorageService` and
`SignedUrlService` are injectable everywhere.

## Local vs S3

The default `LocalDriver` writes under `local_base_dir/<key>` and keeps
per-object metadata in an adjacent `.<name>.meta.json`. It resolves every
key against the base directory and refuses paths that escape it, on top of
the key-safety check performed by `StorageService`.

The `S3Driver` imports `boto3` lazily. If `boto3` isn't installed the
constructor raises with a message pointing at the install command; the
class itself still imports so tests and type-checkers work in a lean dev
environment. When you're ready for S3:

```python
StorageModule.for_root(
    StorageConfig(
        signed_url_secret="...",
        driver="s3",
        s3_bucket="swadely-uploads",
        s3_region="ap-south-1",
    ),
)
```

Product code should inject `StorageService`, never the driver directly —
that keeps size limits, path-safety, content-type inference and virus
scanning centralised.

## Put, get, list

```python
from austial import Injectable
from austial.storage import StorageService

@Injectable()
class DocumentsService:
    def __init__(self, storage: StorageService):
        self._storage = storage

    async def save(self, tenant: str, name: str, body: bytes) -> str:
        stored = await self._storage.put(
            f"{tenant}/{name}",
            body,
            metadata={"tenant": tenant},
        )
        return stored.key  # persist alongside the domain record

    async def read(self, key: str) -> bytes:
        obj = await self._storage.get(key)
        return obj.data or b""

    async def list_for_tenant(self, tenant: str) -> list[str]:
        return [obj.key async for obj in self._storage.list(f"{tenant}/")]
```

`put` returns a `StoredObject` with `key`, `size`, `content_type`, `etag`,
`last_modified` and `metadata`. `get` additionally populates `data`. `stat`
returns everything except `data`. `delete` is idempotent.

Every object key is checked before it reaches the driver: `../foo`,
`/etc/passwd`, `foo\x00bar`, empty strings and Windows drive letters raise
`PathTraversalError` (subclass of `BadRequestException`, so
`AllExceptionsFilter` renders `400`). Content-type is inferred from the
suffix if not passed. Payloads above `max_object_size_bytes` raise
`ObjectTooLargeError`.

## Signed URLs for public-safe downloads

`SignedUrlService` mints URL *fragments* — the framework doesn't know your
public base URL. The product controller prepends it:

```python
from datetime import timedelta

from austial import Controller, Get, Param, Query
from austial.storage import InvalidSignedUrlError, SignedUrlService, StorageService

@Controller("/files")
class FilesController:
    def __init__(self, storage: StorageService, signed: SignedUrlService):
        self._storage = storage
        self._signed = signed

    @Get("/:key/link")
    def issue(self, key: str) -> dict[str, str]:
        fragment = self._signed.issue(key, ttl=timedelta(minutes=5))
        return {"url": f"/files/{key}{fragment}"}

    @Get("/:key")
    async def download(self, key: str, exp: int, m: str, sig: str):
        # rebuild the query string exactly as issued
        fragment = f"?exp={exp}&m={m}&sig={sig}"
        self._signed.verify(fragment, expected_key=key, method="GET")
        obj = await self._storage.get(key)
        return obj.data
```

`verify` raises `InvalidSignedUrlError` (subclass of `ForbiddenException`,
so the framework renders `403`) on:

- an expired `exp`;
- a tampered `sig`;
- a wrong `method` (a `GET` URL replayed as `PUT`);
- a mismatched key (a URL for `a.pdf` used against `b.pdf`).

The signature covers `method\nkey\nexp` under HMAC-SHA256, truncated to 24
bytes and URL-safe-base64 encoded. That's enough to authenticate a
short-lived download without leaking a cookie or an auth header into a
public URL.

## Virus scanning

The framework does not ship a real scanner — ClamAV integration and
third-party APIs are product concerns. It does ship the contract:

```python
from austial.storage import ScanResult, VirusScanHook

class ClamAVHook(VirusScanHook):
    async def scan(self, key: str, data: bytes) -> ScanResult:
        # call clamd, VirusTotal, S3 GuardDuty, whatever
        return ScanResult(clean=True, signature=None)
```

Plug it in at module setup:

```python
StorageModule.for_root(
    StorageConfig(...),
    virus_scan=ClamAVHook(),
)
```

`StorageService.put` awaits the hook before writing. An infected payload
raises `MalwareDetectedError` (subclass of `BadRequestException`) with the
scanner's signature name. `NoopVirusScan` is the default and always
reports clean — an explicit choice so a product that forgets to plug in a
real scanner has to opt out in code, not by accident.

## Standards

- **ISO/IEC 27001 A.8.24 (Use of cryptography)** — signed URLs use
  HMAC-SHA256 with a ≥32-character shared secret; verification is
  constant-time via `hmac.compare_digest`.
- **OWASP ASVS V12 (File and Resources)** — key-safety rejects path
  traversal, absolute paths, null bytes and control characters; size
  limits are enforced at the service boundary; the virus-scan hook is a
  first-class hook rather than an afterthought.
- **OWASP ASVS V6 (Cryptographic Storage)** — signing secrets have to be
  ≥128 bits of entropy (enforced as ≥32 characters at config load).

## Errors

| Error | HTTP | When |
| --- | --- | --- |
| `PathTraversalError` | 400 | key contains `..`, absolute path, null byte |
| `ObjectTooLargeError` | 400 | payload above `max_object_size_bytes` |
| `MalwareDetectedError` | 400 | `VirusScanHook` reported `clean=False` |
| `ObjectNotFoundError` | 404 | `get` or `stat` on a missing key |
| `InvalidSignedUrlError` | 403 | signed URL expired, tampered, or method-mismatched |

Each subclasses the matching `austial.common.exceptions` type so
`AllExceptionsFilter` maps them to the right status without any product
translation layer.

---

Source: https://docs.austial.com/techniques/i18n

# Internationalization

`austial.i18n` translates application strings on the server. It mirrors the
frontend convention we already use: **one JSON file per module per locale**,
merged into a `<module>.<key>` lookup namespace. The intended use is
application-level code — controller responses, error messages that reach the
end user, i18n-aware label prefixes. Framework internals (log lines, startup
banners, developer-facing exceptions) stay in English.

## Layout

```
locales/
    en/
        health.json     # {"check": {"database_label": "Database"}}
        app.json        # {"welcome": "Hello, {name}!"}
    fr/
        health.json
        app.json
```

The loader walks `locales/<locale>/<module>.json` and merges every file into
a `{module: tree}` dict for that locale. The first segment of a translation
key names the module (JSON filename); the rest is a dotted path into that
file's tree:

```python
service.t("health.check.database_label")   # -> "Database"
```

Locales are auto-discovered from the top-level directory names.

## Setup

```python
from pathlib import Path

from austial import Module
from austial.i18n import I18nConfig, I18nModule

@Module(
    imports=[
        I18nModule.for_root(
            I18nConfig(
                locales_dir=Path("i18n/locales"),
                default_locale="en",
                fallback_chain={"en-IN": ["en"], "fr-CA": ["fr"]},
            )
        ),
    ]
)
class AppModule:
    pass
```

`I18nModule` is registered `@Global()` by default, so `I18nService` and
`I18nContext` are injectable everywhere without re-importing.

## Translating in code

```python
from austial import Injectable
from austial.i18n import I18nService

@Injectable()
class HealthController:
    def __init__(self, i18n: I18nService):
        self._i18n = i18n

    def readable_status(self, kind: str) -> str:
        # Looks up "<module>.<subkey>", using the request's locale.
        return self._i18n.t(f"health.check.{kind}_label")
```

Pass `locale=` to override the request's locale (batch jobs, notifications
sent in the recipient's preferred language, tests):

```python
self._i18n.t("app.welcome", locale="fr", name="Alice")
```

`I18nService.has("health.check.queue_label")` returns `True` when the key
resolves to a string in the effective locale (walking the fallback chain
and the default locale). `available_locales()` returns the set of locales
the loader found on disk. `add_locale(locale, translations)` registers or
replaces a locale in-memory — for tests and hot-reload flows.

## Interpolation

The default placeholder syntax is `{name}`:

```json
{"welcome": "Hello, {name}!"}
```

```python
i18n.t("app.welcome", name="Alice")   # -> "Hello, Alice!"
```

Switch to `{{name}}` (i18next-style) by setting `interpolation="double_brace"`
on `I18nConfig`.

A missing variable is substituted with `""` and a warning is logged — in
`strict=True` mode it raises `MissingTranslationError` instead.

## Fallback chain

Lookup order for `t("health.check.database_label", locale="fr-CA")`:

1. `fr-CA` translations.
2. Each locale listed in `I18nConfig.fallback_chain["fr-CA"]` (in order).
3. The `default_locale`.
4. If still nothing: return the key itself (or raise
   `MissingTranslationError` when `strict=True`).

The default locale is always tried last; you do not need to list it.

## Locale middleware

`LocaleMiddleware` decides which locale a request is in, and pushes it into
`I18nContext` for the duration of that request. Precedence (highest first):

1. A dedicated fallback header (default `X-Locale`), useful when the frontend
   already knows the user's preference from their stored profile.
2. `Accept-Language`, parsed with `q`-weighting. The first tag that has
   translations loaded wins; the language subtag is tried when the exact
   region tag is not deployed (`en-IN` → `en`).
3. The configured `default_locale`.

Register it like any other Nest middleware — in your `AppModule.configure()`:

```python
from austial.common.middleware import MiddlewareConsumer
from austial.i18n import LocaleMiddleware

@Module(imports=[I18nModule.for_root(config)])
class AppModule:
    def configure(self, consumer: MiddlewareConsumer) -> None:
        consumer.apply(LocaleMiddleware).for_routes("*")
```

Inside a handler:

```python
from austial import Injectable
from austial.i18n import I18nContext, I18nService

@Injectable()
class InvoicesController:
    def __init__(self, i18n: I18nService, ctx: I18nContext):
        self._i18n = i18n
        self._ctx = ctx

    def show(self):
        # ctx.current_locale() -> the request's locale, resolved by middleware.
        return {"locale": self._ctx.current_locale(), "greeting": self._i18n.t("app.welcome", name="Alice")}
```

## Strict mode

`I18nConfig(strict=True)` turns any missing key or missing interpolation
variable into a `MissingTranslationError` (500). Useful in CI and staging;
production usually wants the default lenient mode where a missing key
degrades to a visible-but-usable label.

## Malformed files

The loader logs a warning (via `austial.common.logger.Logger`) when a JSON
file fails to parse, or when a file's top-level value is not an object, then
continues loading the rest. A bad locale file never crashes the app on boot;
it shows up loudly in the startup logs.

## What this package is not for

Framework internals — the log lines, exception messages you see as a
developer, the `[Austial]` startup banner — stay in English. `austial.i18n`
is for application code (product backends, landing page copy served from the
API, notifications sent to the end user). This mirrors the workspace rule in
`CLAUDE.md`: no hardcoded user-facing strings in app code; all messages,
labels and prefixes live in i18n JSON locale files, keyed by module.

---

Source: https://docs.austial.com/techniques/health-checks

# Health Checks (Terminus)

`austial.terminus` is a Terminus-style health-check module — an aggregator
that runs a set of health indicator checks in parallel intent and returns a
single, consistently-shaped result.

## `HealthCheckService`

```python
from austial.terminus import HealthCheckService, MemoryHealthIndicator

@Injectable()
class HealthService:
    def __init__(
        self,
        health_check_service: HealthCheckService,
        memory: MemoryHealthIndicator,
    ):
        self.health_check_service = health_check_service
        self.memory = memory

    async def check(self):
        return await self.health_check_service.check([
            lambda: self.memory.check_heap("memory_heap", 300 * 1024 * 1024),
        ])
```

```python
@Controller("health")
class HealthController:
    def __init__(self, health_service: HealthService):
        self.health_service = health_service

    @Get()
    async def check(self):
        return await self.health_service.check()
```

## Response shape

```json
{
  "status": "ok",
  "info": {
    "memory_heap": { "status": "up" }
  },
  "error": {},
  "details": {
    "memory_heap": { "status": "up" }
  }
}
```

- `status` is `"ok"` if every indicator passed, `"error"` if any failed.
- `info` collects passing indicators' results; `error` collects failing
  ones; `details` always contains the union of both.

## Built-in indicators

### `MemoryHealthIndicator`

```python
async def check_heap(self, key: str, threshold_bytes: int) -> HealthIndicatorResult: ...
```

Reads process RSS via `resource.getrusage(RUSAGE_SELF).ru_maxrss` and
raises `HealthCheckError` if it exceeds `threshold_bytes`.

### `DatabaseHealthIndicator`

```python
from austial.orm import DataSource
from austial.terminus import DatabaseHealthIndicator

@Injectable()
class DatabaseHealthIndicator(HealthIndicator):
    def __init__(self, data_source: DataSource):
        self._data_source = data_source

    async def ping_check(self, key: str) -> HealthIndicatorResult: ...
```

Runs `SELECT 1` through the injected `DataSource`'s driver (see
[Database](/techniques/database)) and raises `HealthCheckError` if the
connection fails — plug it straight into `check([...])` alongside memory:

```python
result = await self.health_check_service.check([
    lambda: self.memory.check_heap("memory_heap", 300 * 1024 * 1024),
    lambda: self.database.ping_check("database"),
])
```

## Writing a custom indicator

```python
from austial.terminus import HealthCheckError, HealthIndicator, HealthIndicatorResult

class DiskHealthIndicator(HealthIndicator):
    async def check_disk(self, key: str, path: str, threshold_percent: float) -> HealthIndicatorResult:
        usage = shutil.disk_usage(path)
        used_percent = usage.used / usage.total * 100
        is_healthy = used_percent < threshold_percent
        result = self.get_status(key, is_healthy, {"used_percent": round(used_percent, 1)})
        if not is_healthy:
            raise HealthCheckError("Disk usage too high", result)
        return result
```

`HealthIndicator.get_status(key, is_healthy, data)` builds the
`{key: {"status": "up" | "down", **data}}` shape both the built-in
indicators and `HealthCheckService.check()` expect — return it directly
when healthy, or attach it to a raised `HealthCheckError` when not.

## Guarding a health endpoint

Public health checks are common, but so are gated ones (internal
dashboards, deploy tooling). Combine with a [guard](/overview/guards) the
same way `austial new` scaffolds it out of the box:

```python
@Get("protected")
@UseGuards(ApiKeyGuard)
async def protected(self):
    return await self.health_service.check()
```

---

Source: https://docs.austial.com/techniques/testing

# Testing

`austial.testing` provides a way to build a real, fully-wired application
(real DI, real routing) from an ad-hoc set of controllers/providers,
without needing your full `AppModule` or a running server.

## Unit tests — resolving a provider directly

```python

from austial.testing import Test

from .cats_module import CatsModule
from .cats_service import CatsService

@pytest.mark.asyncio
async def test_creates_and_lists_cats():
    module = await Test.create_testing_module(imports=[CatsModule]).compile()
    service = module.get(CatsService)

    service.create(CreateCatDto(name="Whiskers"))

    assert len(service.find_all()) == 1
```

`Test.create_testing_module(imports=, controllers=, providers=)` synthesizes
an anonymous root module from whatever you pass, scans it into a fresh DI
container, and `.compile()`s it into a `TestingModule`. `module.get(Token)`
resolves anything registered in that graph — the same singleton-by-default
behavior as a real running app (resolve the same token twice, get the same
instance back).

## E2E tests — hitting real HTTP routes

```python

from httpx import ASGITransport, AsyncClient

from austial.testing import Test

from .cats_controller import CatsController
from .cats_service import CatsService

@pytest.mark.asyncio
async def test_full_request_lifecycle():
    module = await Test.create_testing_module(
        controllers=[CatsController], providers=[CatsService]
    ).compile()
    app = module.create_austial_application()
    await app.init()

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://testserver") as client:
        create_response = await client.post("/cats", json={"name": "Whiskers"})
        assert create_response.status_code == 201

        list_response = await client.get("/cats")
        assert len(list_response.json()) == 1
```

`module.create_austial_application()` builds a real `AustialApplication`
(no middleware bindings) around the same `FastAPI` app your production code
would run through — routing, guards, pipes, interceptors, and exception
filters all execute exactly as they would in production.
`await app.init()` builds the routes (idempotent — safe even if you never
call `listen()`), and `httpx.ASGITransport` lets `AsyncClient` talk to it
in-process, with **no real network socket** involved.

## Defining throwaway fixtures inline

Because `create_testing_module` accepts any controller/provider classes
directly, framework-level tests (and your own integration tests) don't need
a full sample app to exercise the request lifecycle — define a
`WidgetsController`/`WidgetsService`/`ApiKeyGuard` right in the test file
if that's all a given test needs:

```python
from pydantic import BaseModel

from austial import Body, Controller, Get, Injectable, NotFoundException, Post
from austial.testing import Test

class Widget(BaseModel):
    id: int
    name: str

class CreateWidgetDto(BaseModel):
    name: str

@Injectable()
class WidgetsService:
    def __init__(self):
        self._items: list[Widget] = []
        self._next_id = 1

    def create(self, dto: CreateWidgetDto) -> Widget:
        widget = Widget(id=self._next_id, name=dto.name)
        self._next_id += 1
        self._items.append(widget)
        return widget

@Controller("widgets")
class WidgetsController:
    def __init__(self, service: WidgetsService):
        self.service = service

    @Post()
    async def create(self, dto: CreateWidgetDto = Body()) -> Widget:
        return self.service.create(dto)
```

## Asserting the error body shape

Since [exception filters](/overview/exception-filters) always produce the
same `statusCode`/`message`/`error`/`timestamp`/`path` shape, testing error
paths is just JSON assertions:

```python
response = await client.get("/widgets/999")
assert response.status_code == 404
body = response.json()
assert body["statusCode"] == 404
assert "timestamp" in body
assert body["path"] == "/widgets/999"
```

## Naming convention

Test files use a `*.spec.ts`-style convention translated to Python:
`*_spec.py` (e.g. `cats_service_spec.py`, `app_e2e_spec.py`). A project
scaffolded by `austial new` configures `pytest` to discover this pattern
automatically via `[tool.pytest.ini_options]` in `pyproject.toml`.

---

Source: https://docs.austial.com/ai/overview

# Building with AI agents

Austial looks like NestJS, runs on FastAPI, and is written in Python. That is
exactly the combination that trips up an AI coding tool: it pattern-matches
to one of the three, writes `Depends()` where DI should be, or invents a
decorator that does not exist. This section gives every tool the context it
needs to build **correctly** and to **finish production-ready**, not just to
produce something that runs on a laptop.

It has four layers. Use as many as you need.

| Layer | What it is | Page |
|---|---|---|
| **Instructions** | One `AGENTS.md` in your repo: commands, layout, rules, gotchas | [AGENTS.md template](/ai/agents-md) |
| **Tool wiring** | Symlinks so Claude Code, Copilot, Gemini, Windsurf, Cline, Aider and Zed all read that same file | [Tool setup](/ai/setup) |
| **Reference** | Imports, decorators, request pipeline and patterns in a form an agent can scan | [Framework reference](/ai/reference) |
| **Gates** | What "production-ready" means for an Austial service, as a checklist an agent can run through | [Production readiness](/ai/production) |

[Playbooks](/ai/playbooks) tie them together with task recipes and prompts.

## Five-minute start

New projects are set up for you. `austial new` writes `AGENTS.md` and the
symlinks for every supported tool:

```bash
austial new my-app && cd my-app
```

Open the project in your AI tool and fill in the **Project facts** block at
the top of `AGENTS.md` first. It is the one part of the file only you can
write, and it is what lets the agent make the right call between, say, SQLite
and Postgres or JWT and API keys. Pass `--skip-agents` to opt out.

For an **existing** project, add the same setup by hand:

```bash
curl -fsSL https://docs.austial.com/AGENTS.md -o AGENTS.md
curl -fsSL https://docs.austial.com/setup-agents.sh -o setup-agents.sh
sh setup-agents.sh
```

## Why a single file with symlinks

- **No drift.** One file means the rules Claude Code follows are the rules
  Copilot follows.
- **Reviewable.** A change to agent behaviour is a normal diff in a normal
  pull request.
- **Tool-agnostic.** Switching tools, or a teammate using a different one,
  costs nothing.

## What the agent gets right, and what it still needs you for

With the template in place an agent reliably gets framework mechanics right:
module registration, DI, the request pipeline, error shapes, repository
usage, and the verification loop (`ruff`, `mypy`, `pytest`).

It cannot know your business rules, your threat model, or your deployment
target. Those belong in the **Project facts** block and in code review.
Treat the [production checklist](/ai/production) as a review aid for you, not
a substitute for one.

## Keeping it current

`AGENTS.md` and `setup-agents.sh` are generated from these docs on every site
build, along with `/llms.txt` and `/llms-full.txt`. Re-download them when you
upgrade Austial. In CI, `sh setup-agents.sh --check` fails the build if a
symlink goes missing.

---

Source: https://docs.austial.com/ai/setup

# Tool setup

Every AI coding tool looks for its own instructions file. Maintaining nine
copies of the same rules guarantees they drift. The setup below keeps **one**
file, `AGENTS.md`, and points every tool at it with symlinks.

```
my-app/
├── AGENTS.md                              # the only file you edit
├── CLAUDE.md                -> AGENTS.md  # Claude Code
├── GEMINI.md                -> AGENTS.md  # Gemini CLI
├── CONVENTIONS.md           -> AGENTS.md  # Aider
├── .windsurfrules           -> AGENTS.md  # Windsurf
├── .clinerules              -> AGENTS.md  # Cline
├── .rules                   -> AGENTS.md  # Zed
└── .github/
    └── copilot-instructions.md -> ../AGENTS.md   # GitHub Copilot
```

Several tools (Codex, Cursor, Copilot's coding agent, Amp and others) read
`AGENTS.md` directly, so they need no link at all. Tool conventions change
quickly, so treat the filenames above as the current best practice and check
your tool's own docs if one stops picking the file up.

## New projects

`austial new` already does all of this: it writes `AGENTS.md` and creates every
symlink below (falling back to copies where the filesystem cannot create
symlinks). Nothing to run.

## Existing projects

Run this from the root of your Austial project:

```bash
curl -fsSL https://docs.austial.com/setup-agents.sh -o setup-agents.sh
sh setup-agents.sh
```

The script downloads the [AGENTS.md template](/ai/agents-md) if you do not
have one, then creates every symlink in the diagram. It is safe to re-run: it
replaces links it made earlier and never overwrites a real file.

| Flag | Effect |
|---|---|
| `--copy` | Copy `AGENTS.md` instead of symlinking (for filesystems without symlinks) |
| `--force` | Replace existing real files at the link paths |
| `--check` | Change nothing. Exit non-zero if any link is missing, stale or a drifted copy |

Prefer to read the script before running it? It is reproduced in full
[below](#the-setup-script).

## Doing it by hand

**macOS / Linux**

```bash
ln -s AGENTS.md CLAUDE.md
ln -s AGENTS.md GEMINI.md
ln -s AGENTS.md CONVENTIONS.md
ln -s AGENTS.md .windsurfrules
ln -s AGENTS.md .clinerules
ln -s AGENTS.md .rules
mkdir -p .github
ln -s ../AGENTS.md .github/copilot-instructions.md
```

A symlink target is resolved **relative to the link's own directory**,
which is why the Copilot link points at `../AGENTS.md`.

**Windows**

Symlinks on Windows need Developer Mode (or an elevated shell) and Git
configured to create them:

```powershell
git config core.symlinks true
```

Without that, run `sh setup-agents.sh --copy` from Git Bash or WSL, and
add `sh setup-agents.sh --check` to CI so a copy that drifts from
`AGENTS.md` fails the build.

Commit the links. Git stores a symlink as a tiny file holding the target
path, so teammates on macOS and Linux get working links on clone.

## Per-tool notes

| Tool | Reads | Notes |
|---|---|---|
| Claude Code | `CLAUDE.md` | Symlink works. Alternatively make `CLAUDE.md` a one-liner, `@AGENTS.md`, if you want Claude-only additions below it. |
| OpenAI Codex CLI | `AGENTS.md` | Native. |
| Cursor | `AGENTS.md` | Native. Use `.cursor/rules/*.mdc` only for glob-scoped extras. |
| GitHub Copilot | `.github/copilot-instructions.md` | Symlink. The coding agent also reads `AGENTS.md`. |
| Gemini CLI | `GEMINI.md` | Symlink, or set the context filename to `AGENTS.md` in Gemini's settings. |
| Windsurf | `.windsurfrules` | Symlink. |
| Cline | `.clinerules` | Symlink. |
| Aider | `CONVENTIONS.md` | Symlink, then pass `--read CONVENTIONS.md` or add `read: CONVENTIONS.md` to `.aider.conf.yml`. |
| Zed | `.rules` | Symlink. |
| Anything else | `AGENTS.md` | Paste or `@`-mention the file at the start of the session. |

### Keep tool-specific extras out of the shared file

If one tool needs something the others should not see (a Claude Code
permissions allow-list, a Cursor glob rule), put it in that tool's own config
directory. Never fork the shared instructions. For Claude Code, for example,
a project `.claude/settings.json` that lets the agent run the verification
loop without prompting every time:

```json
{
  "permissions": {
    "allow": [
      "Bash(uv run pytest:*)",
      "Bash(uv run ruff:*)",
      "Bash(uv run mypy:*)",
      "Bash(uv run austial:*)"
    ]
  }
}
```

## Give the agent the docs, not just the rules

`AGENTS.md` tells an agent how to behave. It cannot hold the whole API. The
docs site publishes two machine-readable files, regenerated on every build:

| File | Use it for |
|---|---|
| [`/llms.txt`](https://docs.austial.com/llms.txt) | A short index: page titles, one-line summaries, links. Good default for tools that fetch pages on demand. |
| [`/llms-full.txt`](https://docs.austial.com/llms-full.txt) | Every page concatenated as plain text. Add it as an indexed docs source (for example Cursor's `@Docs`), or have an agent fetch it once at the start of a task. |

A reliable pattern is to start any non-trivial task with:

```
Read https://docs.austial.com/llms.txt, then fetch the pages relevant to
this task before writing code.
```

## Verify the setup

1. Open the project in your tool and ask: *"What command do I run to verify
   my changes in this project?"* It should answer with the ruff, mypy and
   pytest loop from `AGENTS.md`.
2. Ask it to add a `cats` feature. It should reach for
   `austial g resource cats`, not hand-write a FastAPI router.
3. In CI, run `sh setup-agents.sh --check` so a deleted or stale link is
   caught.

## The setup script

```sh
#!/usr/bin/env sh
# Link every AI coding tool's instructions file to a single AGENTS.md.
#
#   sh setup-agents.sh            create symlinks (downloads AGENTS.md if absent)
#   sh setup-agents.sh --copy     copy instead of symlink (no symlink support)
#   sh setup-agents.sh --force    replace real files sitting at a link path
#   sh setup-agents.sh --check    verify only, change nothing (for CI)
set -eu

SOURCE="AGENTS.md"
SOURCE_URL="https://docs.austial.com/AGENTS.md"
MODE="link"
FORCE=0
CHECK=0

for arg in "$@"; do
  case "$arg" in
    --copy) MODE="copy" ;;
    --force) FORCE=1 ;;
    --check) CHECK=1 ;;
    -h|--help) sed -n '2,8p' "$0"; exit 0 ;;
    *) echo "unknown option: $arg" >&2; exit 2 ;;
  esac
done

# path:relative-link-target
TARGETS="
CLAUDE.md:AGENTS.md
GEMINI.md:AGENTS.md
CONVENTIONS.md:AGENTS.md
.windsurfrules:AGENTS.md
.clinerules:AGENTS.md
.rules:AGENTS.md
.github/copilot-instructions.md:../AGENTS.md
"

status=0

if [ ! -f "$SOURCE" ]; then
  if [ "$CHECK" -eq 1 ]; then
    echo "missing: $SOURCE" >&2
    exit 1
  fi
  echo "downloading $SOURCE_URL"
  curl -fsSL "$SOURCE_URL" -o "$SOURCE"
fi

for entry in $TARGETS; do
  path="${entry%%:*}"
  target="${entry#*:}"

  if [ "$CHECK" -eq 1 ]; then
    if [ -L "$path" ]; then
      [ "$(readlink "$path")" = "$target" ] || { echo "stale link: $path"; status=1; }
    elif [ -f "$path" ]; then
      cmp -s "$path" "$SOURCE" || { echo "drifted copy: $path"; status=1; }
    else
      echo "missing: $path"; status=1
    fi
    continue
  fi

  mkdir -p "$(dirname "$path")"

  if [ -L "$path" ]; then
    rm "$path"
  elif [ -e "$path" ]; then
    if [ "$FORCE" -eq 1 ] || cmp -s "$path" "$SOURCE"; then
      rm "$path"
    else
      echo "skipped $path (exists and differs; use --force to replace)"
      continue
    fi
  fi

  if [ "$MODE" = "copy" ]; then
    cp "$SOURCE" "$path"
    echo "copied  $path"
  else
    ln -s "$target" "$path"
    echo "linked  $path -> $target"
  fi
done

exit "$status"
```

---

Source: https://docs.austial.com/ai/agents-md

# AGENTS.md template

`AGENTS.md` is the one file every AI coding tool should read before it
touches your project. This page holds the canonical Austial template. Copy
it into the root of your app, fill in the **Project facts** block, and
[link it to every tool with symlinks](/ai/setup).

```bash
curl -fsSL https://docs.austial.com/AGENTS.md -o AGENTS.md
```

The file below is generated from this page at build time, so the download
and the page never drift apart.

````markdown
# AGENTS.md

This is an **Austial** application: a NestJS-style, batteries-included web
framework for Python, built on FastAPI. Austial is not FastAPI, Django, or
NestJS. It looks like all three, and guessing from them produces code that
does not run. When you are unsure of an API, read the docs instead of
inventing one:

- Docs: https://docs.austial.com
- Whole doc set as one file for LLMs: https://docs.austial.com/llms-full.txt
- Page index: https://docs.austial.com/llms.txt

## Project facts (fill these in)

- App name / purpose: <what this service does>
- Database: <postgres | sqlite> via `austial.orm`
- Auth: <JWT via austial.auth | API key | none>
- Multi-tenant: <yes | no>
- Deployed as: <uvicorn src.main:app on ... >

## Commands

```bash
uv sync --extra dev          # install, including dev tools
cp .env.example .env         # first time only
uv run austial serve         # dev server on :8000 with reload
uv run austial g resource <name>   # scaffold a full CRUD feature module
uv run ruff check . --fix && uv run ruff format .
uv run mypy src tests
uv run pytest -q
```

**Definition of done:** `ruff check`, `ruff format --check`, `mypy src tests`
and `pytest` all pass. Never report work as finished without running them.

## Layout

```
src/
├── main.py              # AustialFactory.create(AppModule) + global pipes/filters
├── app_module.py        # root @Module: imports every feature module
└── modules/<snake_name>/
    ├── <name>_module.py
    ├── <name>_controller.py
    ├── <name>_service.py
    ├── entities/<name>_entity.py
    └── dto/create_<name>_dto.py, update_<name>_dto.py
tests/unit/, tests/e2e/      # files named *_spec.py
```

- One folder per feature. File names use underscore suffixes
  (`cats_controller.py`), never dots.
- Imports are absolute from the project root: `from src.modules.cats...`.
- Prefer `uv run austial g resource <name>` over hand-writing a feature. It
  also patches `app_module.py`.

## The mental model

Request pipeline, in order: **guards -> pipes -> interceptors -> handler ->
exception filters**. Middleware runs before all of it.

1. **Everything must be registered.** A controller goes in a module's
   `controllers=[...]`, a provider in `providers=[...]`, and the module in
   an `imports=[...]` chain that reaches `AppModule`. A missing registration
   shows up as a missing route or a DI error, not an import error.
2. **Dependency injection is by `__init__` type hint.** Providers are
   singletons (`Scope.DEFAULT`) unless marked `Scope.TRANSIENT`. There is no
   request scope. Never instantiate a service yourself with `Foo()`.
3. **Repositories** are injected as
   `repo: Repository[Cat] = InjectRepository(Cat)`, and the owning module
   needs `OrmModule.for_feature([Cat])`. The root needs one
   `OrmModule.for_root(...)`.
4. **Request data** uses marker defaults: `Body()`, `Param()`, `Query()`,
   `Headers()`. DTOs are plain pydantic `BaseModel`s. Route paths use `:id`
   and `:key*` (catch-all that also matches `/`), not `{id}`.
5. **Errors:** raise Austial exceptions (`NotFoundException`,
   `BadRequestException`, `ConflictException`, ...). Do not hand-build error
   responses. The global `AllExceptionsFilter` renders
   `statusCode/message/error/timestamp/path`.
6. **Guards are the security boundary.** Every non-public route must be
   covered by `AuthGuard` (and `RolesGuard` / `TenantGuard` where relevant).
   Mark intentionally open routes with `@Public()`.

## Import map

| You need | Import from |
|---|---|
| `Module Controller Injectable Get Post Put Patch Delete Body Param Query Headers UseGuards UseInterceptors UsePipes UseFilters HttpCode Header Scope Global Inject`, all `*Exception` classes, `CanActivate ExecutionContext NestInterceptor NestMiddleware PipeTransform ExceptionFilter Catch ValidationPipe AustialFactory Reflector` | `austial` |
| `ConfigModule ConfigService` | `austial.config` |
| `OrmModule DataSource Entity Column PrimaryGeneratedColumn Repository InjectRepository FindOptions In NotIn IsNull NotNull ManyToOne OneToMany` | `austial.orm` |
| `AuthModule AuthConfig AuthGuard AuthService CurrentUser Public JwtStrategy Argon2Hasher TotpStrategy` | `austial.auth` |
| `AuthzModule RolesGuard Roles Permissions AuthorizeObject PermissionService` | `austial.authz` |
| `TenancyModule TenantGuard TenantContext RequireTenant Tenanted` | `austial.tenancy` |
| `SecurityModule SecurityConfig RateLimit RateLimitGuard IdempotencyKeyGuard CsrfInterceptor SecureHeadersMiddleware PIIRedactor EnvelopeEncryptor` | `austial.security` |
| `AuditModule AuditConfig AuditLogService AuditEvent Audited OrmAuditStore` | `austial.audit` |
| `JobsModule JobsConfig JobQueue Job Cron Worker` | `austial.jobs` |
| `StorageModule StorageConfig StorageService SignedUrlService` | `austial.storage` |
| `MigrationsModule MigrationsConfig MigrationRunner` | `austial.migrations` |
| `I18nModule I18nConfig I18nService LocaleMiddleware` | `austial.i18n` |
| `HealthCheckService MemoryHealthIndicator DatabaseHealthIndicator` | `austial.terminus` |
| `Test` (testing module) | `austial.testing` |

If a symbol is not in this table, check
https://docs.austial.com/llms-full.txt before using it.

## Do

- Write async handlers and services. The ORM is async end to end.
- Return typed values or DTOs from handlers, and validate input with DTOs.
- Filter every query on a `Tenanted` entity by `tenant_id`, and set it
  explicitly on insert from `TenantContext.require()`.
- Use `repository.insert(...)` when the primary key is assigned by the
  application (UUIDs). `save()` would issue an UPDATE.
- Use `IsNull()` / `NotNull()` to filter on NULL. `{"col": None}` compiles to
  `= NULL` and never matches.
- Use `append_only=True` entities for audit and ledger data.
- Write a test for every new route (`austial.testing.Test`, `httpx`
  `ASGITransport`) and assert error bodies, not just 200s.
- Add a new Alembic revision for every schema change. Read the generated
  file before committing it.

## Do not

- Do not use FastAPI `Depends()` for things Austial's DI already provides.
- Do not import SQLAlchemy in application code, or build SQL with f-strings.
  Use `Repository`, `FindOptions` and `QueryBuilder` with bound params.
- Do not log or audit-log PII, tokens or secrets. Pass records through
  `PIIRedactor` first.
- Do not edit a migration that has been applied. Add a new one.
- Do not hardcode secrets. Read them with `ConfigService.get_or_throw(...)`.
- Do not rely on `synchronize=True` outside local development and tests.
- Do not ship the in-memory defaults (see below) to production.

## Not production-safe by default

These ship as in-memory or local reference implementations. Replace each with
a persistent or shared one before running more than one worker or surviving a
restart:

| Component | Default | Production need |
|---|---|---|
| Refresh tokens | `InMemoryRefreshTokenStore` | ORM/Redis-backed `RefreshTokenStore` |
| Rate limiting | `InMemoryRateLimitStore` | shared store, plus `trusted_proxy_hops` behind a proxy |
| Idempotency keys | `InMemoryIdempotencyStore` | persistent store |
| Audit log | `InMemoryAuditStore` | `OrmAuditStore` or a SIEM store |
| Background jobs | in-memory `JobBackend` (process-local) | persistent `JobBackend` |
| File storage | `driver="local"` | `driver="s3"` |
| Schema | `synchronize=True` | Alembic migrations (`austial-migrate`) |

## Production gates

Before calling a feature production-ready, confirm:

1. All checks in "Definition of done" pass.
2. Secrets come from the environment and `jwt_secret` /
   `signed_url_secret` are 32+ characters.
3. `DataSource.initialize()` runs in a `startup` event handler, because
   `uvicorn src.main:app` never calls `bootstrap()`.
4. `GET /health` is wired to `DatabaseHealthIndicator`.
5. Global `ValidationPipe` and `AllExceptionsFilter` are registered, and
   `AuthGuard` is global or on every non-public controller.
6. Full checklist: https://docs.austial.com/ai/production
````

---

Source: https://docs.austial.com/ai/reference

# Framework reference for agents

A dense, scannable summary of Austial's surface area, organised by the
question an agent actually has while coding. Each row links to the full page.
When this summary and a full page disagree, the full page wins.

## Request pipeline

```
middleware -> guards -> pipes -> interceptors (wrapping) -> handler -> exception filters
```

Guards run first, so unauthorised requests never pay for validation or reach
handler code. Exceptions raised anywhere in the pipeline, including FastAPI's
own validation errors, flow through the exception filters.

## Building blocks

| Concept | Declare with | Registered in | Page |
|---|---|---|---|
| Module | `@Module(imports=, controllers=, providers=, exports=)` | an `imports=[...]` chain reaching `AppModule` | [Modules](/overview/modules) |
| Controller | `@Controller("prefix")` + `@Get/@Post/@Put/@Patch/@Delete/@Options/@Head` | a module's `controllers` | [Controllers](/overview/controllers) |
| Provider | `@Injectable()` | a module's `providers` | [Providers](/overview/providers) |
| Guard | subclass `CanActivate`, `can_activate(context) -> bool` | `@UseGuards(...)` or `app.use_global_guards(...)` | [Guards](/overview/guards) |
| Pipe | subclass `PipeTransform` | `@UsePipes(...)` or `app.use_global_pipes(...)` | [Pipes](/overview/pipes) |
| Interceptor | subclass `NestInterceptor`, `intercept(context, call_next)` | `@UseInterceptors(...)` | [Interceptors](/overview/interceptors) |
| Exception filter | subclass `ExceptionFilter` + `@Catch(...)` | `@UseFilters(...)` or `app.use_global_filters(...)` | [Exception filters](/overview/exception-filters) |
| Middleware | subclass `NestMiddleware` | `configure(consumer)` on a module | [Middleware](/overview/middleware) |
| Custom provider | `{"provide": TOKEN, "useClass"/"useValue"/"useFactory": ..., "inject": [...]}` | a module's `providers` | [Custom providers](/fundamentals/custom-providers) |
| Dynamic module | static method returning `DynamicModule` | `imports=[X.for_root(...)]` | [Dynamic modules](/fundamentals/dynamic-modules) |

## Handler parameters

Python has no parameter decorators, so request data is bound with marker
defaults and type hints:

| You want | Write |
|---|---|
| JSON body validated by a DTO | `dto: CreateCatDto = Body()` |
| Path parameter | `id: int` (matches `:id` in the route) or `key: str = Param("key")` |
| Catch-all path (matches `/`) | route `files/:key*`, then `key: str = Param("key")` |
| Query string value | `limit: int = Query(default=10)` |
| Header | `x_request_id: str \| None = Headers("x-request-id")` |
| Authenticated principal | `user=CurrentUser()` (`austial.auth`) |
| Tenant context | inject `TenantContext` in the constructor |
| Status code / header | `@HttpCode(204)`, `@Header("Cache-Control", "no-store")` |

## Errors

Raise these from anywhere; never build the response yourself. Every one takes
`(message: str | dict | None = None)`.

| Exception | Status | Exception | Status |
|---|---|---|---|
| `BadRequestException` | 400 | `ConflictException` | 409 |
| `UnauthorizedException` | 401 | `UnprocessableEntityException` | 422 |
| `ForbiddenException` | 403 | `TooManyRequestsException` | 429 |
| `NotFoundException` | 404 | `InternalServerErrorException` | 500 |
| `MethodNotAllowedException` | 405 | `ServiceUnavailableException` | 503 |

Response shape: `{"statusCode", "message", "error", "timestamp", "path"}`.

## Data access (`austial.orm`)

| Task | API |
|---|---|
| Define a table | `@Entity()` class with `PrimaryGeneratedColumn()`, `Column()`, relation markers |
| Wire the connection | `OrmModule.for_root(type_="postgres" \| "sqlite", url=..., entities=[...])` or `for_root_async(use_factory=..., inject=[ConfigService], entities=[...])` |
| Wire a feature's entities | `OrmModule.for_feature([Cat])` in the feature module |
| Get a repository | `repo: Repository[Cat] = InjectRepository(Cat)` |
| Read | `find`, `find_one`, `find_by`, `find_one_by`, `count` |
| Write | `save` (insert when PK unset, else update), `insert` (always insert), `update(criteria, partial)`, `delete(criteria)`, `remove(entity)` |
| Filter | `FindOptions(where=..., relations=[...], order_by=..., skip=, take=)`; wrap values in `In([...])`, `NotIn([...])`, `IsNull()`, `NotNull()` |
| Aggregate | `repo.sum(column, criteria=, group_by=)`, `repo.aggregate("SUM" \| "AVG" \| "MIN" \| "MAX", ...)` |
| Complex query | `repo.create_query_builder("alias")` with `.where(sql, params)`, `.where_eq`, `.where_in`, joins, `.get_many()` |
| Transaction | `await data_source.transaction(async_callback)` |
| Immutable rows | `@Entity(append_only=True)` (audit, ledger) |
| Initialise | `await app.get(DataSource).initialize()` at startup |

Relations (`OneToOne`, `OneToMany`, `ManyToOne`, `ManyToMany`) take a class or
a `lambda: Class` target, and pass `join_column=JoinColumn(...)` as a keyword
argument. Full detail: [Database](/techniques/database).

## Feature packages

| Need | Package | Wire with | Page |
|---|---|---|---|
| Environment config | `austial.config` | `ConfigModule.for_root()`, then `ConfigService.get / get_or_throw` | [Configuration](/techniques/configuration) |
| Login, JWT, MFA, refresh tokens | `austial.auth` | `AuthModule.for_root(AuthConfig(jwt_secret=..., issuer=...))`, `AuthGuard`, `@Public()` | [Authentication](/techniques/authentication) |
| Roles and permissions | `austial.authz` | `AuthzModule.for_root()`, `RolesGuard`, `@Roles`, `@Permissions`, `@AuthorizeObject` | [Authorization](/techniques/authorization) |
| Tenant isolation | `austial.tenancy` | `TenancyModule.for_root()`, `TenantGuard`, `@RequireTenant()`, `Tenanted` mixin | [Multi-tenancy](/techniques/multi-tenancy) |
| Headers, rate limit, idempotency, CSRF, PII, encryption | `austial.security` | `SecurityModule.for_root(SecurityConfig(...))` | [Security](/techniques/security) |
| Who did what | `austial.audit` | `AuditModule.for_root(AuditConfig(...), store=OrmAuditStore(repo))`, `@Audited` | [Audit log](/techniques/audit-log) |
| Background work | `austial.jobs` | `JobsModule.for_root(JobsConfig())`, `@Job`, `@Cron`, `JobQueue.enqueue` | [Background jobs](/techniques/background-jobs) |
| Files | `austial.storage` | `StorageModule.for_root(StorageConfig(...))`, `StorageService`, `SignedUrlService` | [File storage](/techniques/file-storage) |
| Schema changes | `austial.migrations` | `MigrationsModule.for_root(...)`, `austial-migrate` CLI | [Migrations](/techniques/migrations) |
| Translations | `austial.i18n` | `I18nModule.for_root(I18nConfig(...))`, `LocaleMiddleware` | [i18n](/techniques/i18n) |
| Liveness / readiness | `austial.terminus` | `HealthCheckService` plus indicators | [Health checks](/techniques/health-checks) |
| Tests | `austial.testing` | `Test.create_testing_module(...).compile()` | [Testing](/techniques/testing) |

## Guard ordering

When stacking guards, list them in the order they must run:

```python
@UseGuards(AuthGuard, TenantGuard, RolesGuard)
```

`TenantGuard` and `RolesGuard` read the principal that `AuthGuard`
publishes on `request.state`, so `AuthGuard` must come first. Rate limiting
and idempotency guards usually go before `AuthGuard` so abusive traffic is
rejected cheaply.

## CLI

```bash
austial new <name> [--skip-install] [--skip-git] [--link PATH]
austial g module|controller|service|resource <name>
austial serve [--host] [--port] [--reload/--no-reload]
austial-migrate init | revision -m "msg" --autogenerate | upgrade head | downgrade -1 | current | history
austial-worker run src.app_module:AppModule
```

## Known limits

Do not build on features that do not exist yet:

- No `Scope.REQUEST`. Only singleton (`DEFAULT`) and `TRANSIENT`.
- Module `imports` / `exports` are not enforced as a visibility boundary.
  Write them correctly anyway.
- `CallHandler.handle()` is a plain async continuation, not an RxJS
  Observable.
- `DataSource.initialize()` is a manual step. There is no async module-init
  hook.
- Only PostgreSQL (`asyncpg`) and SQLite (`aiosqlite`) drivers exist.
- The `Tenanted` mixin's `tenant_id` is invisible to mypy's synthesised
  `__init__`. Declare the column on the entity directly if strict typing
  matters more than reuse.

## Gotchas that cost agents the most time

| Symptom | Cause | Fix |
|---|---|---|
| Route returns 404 / not in `/docs` | Controller or module not registered | Add to `controllers=[...]` and import the module from `AppModule` |
| `Cannot resolve` DI error | Provider missing from any module's `providers` | Register it; check the `__init__` type hints are real classes |
| `InjectRepository` fails to resolve | Entity not registered | `OrmModule.for_feature([Entity])` in the owning module, and the entity in `for_root(entities=[...])` |
| `save()` on a UUID-keyed row updates nothing | `save()` treats a set PK as an UPDATE | Use `insert()` |
| `where={"col": None}` matches nothing | Compiles to `= NULL` | Use `IsNull()` |
| App works with `python src/main.py`, tables missing under uvicorn | `bootstrap()` never runs under `uvicorn src.main:app` | Initialise `DataSource` in a `startup` event handler |
| Rate limit keyed on the proxy's IP | `X-Forwarded-For` ignored by default | Set `SecurityConfig.trusted_proxy_hops` |
| `CurrentUser()` is `None` on a protected route | No `AuthGuard` ran for that route | Add the guard, or register it globally |

---

Source: https://docs.austial.com/ai/production

# Production readiness

"It runs locally" is not done. This is the checklist an agent (or a reviewer)
should walk before calling an Austial service production-ready. Each item
names the failure it prevents and links to the page with the detail.

An agent can verify most items by reading code. Items marked **(you)** need a
human decision or access to your infrastructure.

## 1. Build gates

All must pass, with no `# type: ignore` or `noqa` added just to get green:

```bash
uv run ruff check .
uv run ruff format --check .
uv run mypy src tests
uv run pytest -q
```

- [ ] Every new route has an e2e test through `Test.create_testing_module` and
      `httpx.ASGITransport` ([Testing](/techniques/testing)).
- [ ] Error paths are asserted (`statusCode`, `message`, `path`), not just
      the happy 200.
- [ ] Tests for security-relevant routes cover the **denied** case: no
      token, wrong role, wrong tenant.

## 2. Startup and process model

- [ ] `src/main.py` registers `ValidationPipe()` and `AllExceptionsFilter()`
      globally so every error has one shape.
- [ ] `DataSource.initialize()` runs in a `startup` event handler. Under
      `uvicorn src.main:app` or `austial serve`, `bootstrap()` is never
      called ([Database](/techniques/database), "Production deployments").
- [ ] `synchronize=True` is **off**. The schema comes from migrations.
- [ ] CORS is configured deliberately. `app.enable_cors()` with no arguments
      allows **every origin, method and header, with credentials**. That is a
      development default (and what the scaffold ships). Pass explicit
      `allow_origins=[...]` **(you)**.
- [ ] The production command is explicit, for example
      `uvicorn src.main:app --host 0.0.0.0 --port 8000`, with `--reload` off.

## 3. Configuration and secrets

- [ ] Every secret is read with `ConfigService.get_or_throw(...)` so a missing
      value fails at startup instead of falling back to a default.
- [ ] `.env` is git-ignored. `.env.example` lists every variable with no real
      values.
- [ ] `AuthConfig.jwt_secret` and `StorageConfig.signed_url_secret` are 32+
      characters. The framework refuses shorter ones.
- [ ] No secret, token or key appears in logs, audit `metadata`, or error
      messages.
- [ ] Encryption keys in `SecurityConfig.encryption_keys` are loaded from a
      secret manager, and no historical key still referenced by data is
      ever removed **(you)**.

## 4. Replace the in-memory defaults

These are reference implementations. They are correct, but process-local or
non-durable. With more than one worker or one restart, they silently lose
state:

| Component | Default | Replace with | Page |
|---|---|---|---|
| Refresh tokens | `InMemoryRefreshTokenStore` | persistent `RefreshTokenStore` (pass a class so it gets DI) | [Authentication](/techniques/authentication#refresh-tokens) |
| Rate limiting | `InMemoryRateLimitStore` | shared store via `rate_limit_store=` | [Security](/techniques/security#rate-limiting) |
| Idempotency keys | `InMemoryIdempotencyStore` | persistent store | [Security](/techniques/security#idempotency) |
| Audit log | `InMemoryAuditStore` | `OrmAuditStore(repo)` or a SIEM-backed store | [Audit log](/techniques/audit-log#persisting-the-log) |
| Background jobs | in-memory `JobBackend` | persistent backend via `backend=` | [Background jobs](/techniques/background-jobs) |
| File storage | `driver="local"` | `driver="s3"` | [File storage](/techniques/file-storage) |

The jobs case deserves emphasis: `austial-worker run` is a **separate
process**, and an in-memory backend is local to the process that created it.
A web process and a worker process do not share a queue until you supply a
persistent backend.

## 5. Authentication, authorisation, tenancy

- [ ] `AuthGuard` is registered globally (`app.use_global_guards(...)`) or on
      every non-public controller. Intentionally open routes carry
      `@Public()` and nothing else is open by accident.
- [ ] Guard order is correct: `AuthGuard` before `TenantGuard` / `RolesGuard`.
- [ ] Every mutating handler carries `@Roles`, `@Permissions` or `@Public`. For
      a hard guarantee subclass `RolesGuard` with `deny_undeclared = True`
      ([Authorization](/techniques/authorization#deny-by-default)).
- [ ] Object-level access (ownership, tenant match) uses `@AuthorizeObject`.
      A role check alone does not stop one user reading another's record.
- [ ] Multi-tenant apps: `TenantGuard` subclass with `require_by_default =
      True`, and **every** query on a `Tenanted` entity filters by
      `tenant_id` ([Multi-tenancy](/techniques/multi-tenancy)).
- [ ] Password changes, MFA re-enrolment and admin actions call
      `logout_everywhere(subject)`.
- [ ] Auth failures return a generic 401. The specific sub-type is logged,
      never returned to the client.
- [ ] Response DTOs are narrow. Never return an entity that carries
      `password_hash`, secrets or internal columns.

## 6. Abuse and data protection

- [ ] `SecureHeadersMiddleware` is applied to `"*"`, and the CSP is tuned for
      your frontend **(you)**.
- [ ] Login, password reset, OTP and any expensive endpoint carry
      `@RateLimit(...)` behind `RateLimitGuard`.
- [ ] Behind a load balancer or CDN, `trusted_proxy_hops` matches the number
      of proxies **you** operate. Too low rate-limits the proxy; too high lets
      clients forge their IP **(you)**.
- [ ] Payment-like or otherwise non-repeatable `POST`s use
      `IdempotencyKeyGuard`.
- [ ] Cookie-authenticated mutating routes use `CsrfInterceptor`.
- [ ] Sensitive columns use `EnvelopeEncryptor`. Logged or audited records pass
      through `PIIRedactor`.
- [ ] Uploads set `allowed_content_types` and `max_object_size_bytes`, and use
      signed URLs for downloads.

## 7. Data and migrations

- [ ] Every schema change has an Alembic revision, and the generated file was
      read and corrected (renames, enums, server defaults, backfills).
- [ ] Applied revisions are never edited. Fix forward with a new one.
- [ ] Deploys run `austial-migrate upgrade head` **before** the new code takes
      traffic, and a rollback plan exists **(you)**.
- [ ] Multi-step writes use `data_source.transaction(...)`.
- [ ] Audit and ledger tables use `@Entity(append_only=True)`, and, if
      regulation requires it, a database-level `REVOKE UPDATE, DELETE`. The ORM
      guard does not stop raw SQL.
- [ ] Queries you wrote by hand use bound parameters, never string
      formatting. Reserved-word table or column names are quoted in raw
      fragments.

## 8. Observability and operations

- [ ] `GET /health` returns real state: `DatabaseHealthIndicator` wired in,
      and the endpoint reachable by your orchestrator ([Health checks](/techniques/health-checks)).
- [ ] Sensitive health details, if any, sit behind a guard.
- [ ] Audit events are recorded for security-relevant actions (login, role
      change, data export, deletion) ([Audit log](/techniques/audit-log)).
- [ ] Background jobs have `max_attempts`, backoff, and an `idempotency_key`
      at enqueue so a retry cannot do the work twice.
- [ ] Logs, metrics and alerting exist and someone owns them **(you)**.

## How an agent should use this page

Give the agent this instruction at the end of a feature:

```
Review the diff against https://docs.austial.com/ai/production.
For each section, list what applies to this change, whether it is
satisfied (with file and line), and what is missing. Do not mark an item
done without citing code or a test. Fix what is in scope and list the rest.
```

Items marked **(you)** should come back as questions, not as assumptions.

---

Source: https://docs.austial.com/ai/playbooks

# Playbooks

Task recipes for agents, and the prompts to start them. Each recipe is the
shortest path that ends with something that passes the
[production gates](/ai/production), not just something that runs.

## The loop every task follows

```
1. Read      AGENTS.md, then the docs page for the feature you are touching
2. Plan      list the files you will create or change, and the tests you will add
3. Build     smallest change that works; scaffold with `austial g` when it fits
4. Verify    ruff check, ruff format --check, mypy src tests, pytest -q
5. Review    walk the diff against /ai/production; list what does not apply
```

Step 4 is not optional. If a check fails, fix the cause instead of silencing
it.

## Starter prompts

Paste one of these as the first message of a session.

**New feature**

```
Read AGENTS.md and https://docs.austial.com/llms.txt. Add a "invoices"
feature: list, create, get, and cancel, scoped per tenant, with an audit
event on cancel. Use `austial g resource` as the starting point. Write e2e
tests including the denied cases. Run the full verification loop and show me
the output before you say you are done.
```

**Review before merge**

```
Review the current diff against https://docs.austial.com/ai/production. For
each section list what applies, whether it is satisfied (cite file and line),
and what is missing. Do not mark anything satisfied without evidence.
```

**Harden an existing service**

```
Audit this service against https://docs.austial.com/ai/production section
by section. Report findings first, ordered by severity, and wait for my
go-ahead before changing anything.
```

## Recipe: add a CRUD feature

```bash
uv run austial g resource invoices
```

This creates the module, controller, service, entity and DTOs, and patches
`app_module.py`. Then, by hand:

1. Add real columns to `entities/invoices_entity.py`, and real fields to both
   DTOs. The generated `name` column is a placeholder.
2. Register the entity in `OrmModule.for_root(entities=[...])` at the root.
3. Generate and read a migration:
   `austial-migrate revision -m "add invoices" --autogenerate`.
4. Put guards on the controller: `@UseGuards(AuthGuard, TenantGuard, RolesGuard)`.
5. Write tests, then run the verification loop.

## Recipe: protect a route

```python
from austial import Controller, Get, UseGuards
from austial.auth import AuthGuard, CurrentUser
from austial.authz import Roles, RolesGuard

@Controller("reports")
@UseGuards(AuthGuard, RolesGuard)
class ReportsController:
    @Get()
    @Roles("admin", "ops")
    async def list(self, user=CurrentUser()):
        ...
```

Then test three cases: no token (401), valid token without the role (403),
valid token with the role (200).

## Recipe: add a tenant-scoped entity

```python
from austial.orm import Column, Entity, PrimaryGeneratedColumn
from austial.tenancy import Tenanted

@Entity("invoices")
class Invoice(Tenanted):
    id: int = PrimaryGeneratedColumn()
    number: str = Column(nullable=False)
```

In the service, take the tenant from `TenantContext.require()`, set
`tenant_id` on insert, and filter **every** read, update and delete by it.
Test that tenant A cannot read, update or delete tenant B's row.

## Recipe: add a background job

```python
from austial import Injectable
from austial.jobs import Job

@Injectable()
class RemittanceJobs:
    @Job("send-remittance", max_attempts=5, backoff_seconds=60)
    async def send(self, *, remittance_id: str) -> None:
        ...
```

Register `JobsModule.for_root(JobsConfig())` and the provider class, enqueue
with `JobQueue.enqueue("send-remittance", payload=..., idempotency_key=...)`.
Make the handler safe to run twice, because retries are the point. Before
production, replace the in-memory backend (see [step 4](/ai/production#4-replace-the-in-memory-defaults)).

## Recipe: change the schema safely

1. Edit the entity.
2. `austial-migrate revision -m "describe the change" --autogenerate`.
3. **Read the generated revision.** Fix renames (drop + create becomes
   `op.alter_column`), enums, server defaults. Add a backfill if existing rows
   need values.
4. `austial-migrate upgrade head` against a scratch database, then
   `downgrade -1` and `upgrade head` again to prove it reverses.
5. Never edit a revision that has shipped. Add a new one.

## Recipe: debug a DI or routing failure

| Symptom | Check first |
|---|---|
| Route missing | Controller in `controllers=[...]`? Module imported by `AppModule`? |
| `Cannot resolve ...` | Provider in some module's `providers`? Constructor hints are real classes, not strings? |
| Repository injection fails | `OrmModule.for_feature([Entity])` in the module, entity in `for_root(entities=...)` |
| Guard never runs | Applied with `@UseGuards` or `use_global_guards`? Guard class registered as a provider if it has constructor dependencies? |
| `CurrentUser()` is `None` | `AuthGuard` did not run for that route |

Reproduce it in a test with `Test.create_testing_module(...)` before changing
code. A failing test that names the missing registration is faster than
guessing.

## Anti-patterns agents fall into

| Looks right | Why it is wrong here | Do this |
|---|---|---|
| `router = APIRouter()` and `@router.get(...)` | Bypasses modules, DI, guards and the filter chain | `@Controller` + `@Get` inside a module |
| `Depends(get_db)` and a SQLAlchemy `Session` | The ORM is async and DI-managed; SQLAlchemy is not part of the app API | `Repository` via `InjectRepository` |
| `raise HTTPException(404)` | Skips the consistent error body | `raise NotFoundException("...")` |
| `{id}` in a route string | Austial rewrites `:id`; mixing styles confuses readers | `@Get(":id")` |
| Calling `Service()` in a handler | Singleton state and dependencies are skipped | Constructor-inject it |
| `where={"deleted_at": None}` | `= NULL` never matches | `IsNull()` |
| Catching `Exception` and returning `{"error": ...}` | Hides failures from the filter chain and the audit trail | Let it propagate, or raise a specific exception |
| `synchronize=True` in production | Skips review of schema changes | Alembic migrations |

---

Source: https://docs.austial.com/cli/overview

# CLI Overview

The `austial` CLI scaffolds brand-new projects and generates artifacts into
an existing one.

## Installation

```bash
uv tool install austial
# or: pipx install austial / pip install austial
```

## Commands at a glance

| Command | Purpose |
|---|---|
| `austial new <name>` | Scaffold a brand-new project |
| `austial generate <schematic> <name>` (alias `austial g`) | Add an artifact to an existing project |
| `austial serve` | Run the app with auto-reload |

## Schematics

`austial generate`/`austial g` supports four schematics:

| Schematic | Generates |
|---|---|
| `module` | A bare `@Module()` |
| `controller` | A bare `@Controller()` with a `find_all` stub |
| `service` | A bare `@Injectable()` |
| `resource` | Module + full CRUD controller + service + DTOs + entity, patched into `app_module.py` |

See [Usage](/cli/usage) for `new`/`serve` flags, and
[Generating Resources](/cli/generating) for exactly what each schematic
produces on disk.

---

Source: https://docs.austial.com/cli/usage

# CLI Usage

## `austial new`

```bash
austial new <name> [--directory/-d PATH] [--skip-install] [--skip-git] [--skip-agents] [--link PATH]
```

| Flag | Effect |
|---|---|
| `--directory`, `-d` | Scaffold into a specific path instead of `./<name>` |
| `--skip-install` | Don't run `uv sync` after scaffolding |
| `--skip-git` | Don't run `git init` after scaffolding |
| `--skip-agents` | Don't write `AGENTS.md` or the per-tool symlinks (see [AI Agents](/ai/setup)) |
| `--link PATH` | Add an editable `[tool.uv.sources]` entry pointing `austial` at a local framework checkout, instead of resolving it from PyPI |

```bash
austial new my-app
```

generates a full project (see [First Steps](/first-steps) for the
resulting file tree), then — unless skipped — runs `git init -q` and
`uv sync`.

Use `--link` when you're developing the framework itself and want a
throwaway project that resolves `austial` from your local clone instead of
the published package:

```bash
git clone https://github.com/austial/austial-py
cd austial-py && uv sync
uv run austial new ../my-app --link .
```

This writes into the generated `pyproject.toml`:

```toml
[tool.uv.sources]
austial = { path = "/path/to/austial-py", editable = true }
```

so `uv sync` inside `my-app` resolves `austial` from your checkout instead
of PyPI — edits to the framework are picked up immediately without
reinstalling.

## `austial generate` / `austial g`

```bash
austial generate <module|controller|service|resource> <name>
austial g <module|controller|service|resource> <name>       # short alias
```

```bash
austial generate module cats        # src/modules/cats/cats_module.py
austial generate controller cats    # src/modules/cats/cats_controller.py
austial generate service cats       # src/modules/cats/cats_service.py
austial generate resource cats      # full CRUD module
austial g resource cats             # same, short alias
```

Must be run from inside an existing project (or a subdirectory of one) --
the CLI walks up looking for `src/app_module.py` to find the project root,
raising an error if none is found. See
[Generating Resources](/cli/generating) for what each schematic actually
writes to disk, including how `resource` patches `app_module.py`
automatically.

## `austial serve`

```bash
austial serve [--host 0.0.0.0] [--port 8000] [--reload/--no-reload]
```

Equivalent to:

```bash
uv run uvicorn src.main:app --reload
```

`--reload` is **on by default** — pass `--no-reload` to disable it for a
production-like local run.

---

Source: https://docs.austial.com/cli/generating

# Generating Resources

Every schematic writes into `src/modules/<snake_case_name>/`, following the
same one-folder-per-feature convention the rest of the framework uses (see
[Modules](/overview/modules)).

## `module` / `controller` / `service`

The three bare schematics are intentionally minimal — a starting point, not
a full feature:

```bash
austial generate module cats
```
```python
# src/modules/cats/cats_module.py
@Module()
class CatsModule:
    pass
```

```bash
austial generate controller cats
```
```python
# src/modules/cats/cats_controller.py
@Controller("cats")
class CatsController:
    @Get()
    async def find_all(self):
        return []
```

```bash
austial generate service cats
```
```python
# src/modules/cats/cats_service.py
@Injectable()
class CatsService:
    pass
```

Running `module` for a name also patches the nearest `app_module.py`,
inserting both the import line and the class into the `imports=[...]`
array — see [App module patching](#app-module-patching) below.

## `resource` — full CRUD

```bash
austial generate resource cats
```

generates a complete feature module:

```
src/modules/cats/
├── __init__.py
├── cats_module.py
├── cats_controller.py
├── cats_service.py
├── entities/cats_entity.py
└── dto/
    ├── create_cats_dto.py
    └── update_cats_dto.py
```

`entities/cats_entity.py` — an `@Entity()`-decorated domain entity (see
[Database](/techniques/database) for the full `austial.orm` API):

```python
from austial.orm import Column, Entity, PrimaryGeneratedColumn

@Entity()
class Cat:
    id: int = PrimaryGeneratedColumn()
    name: str = Column()
```

`dto/create_cats_dto.py` / `dto/update_cats_dto.py`:

```python
class CreateCatDto(BaseModel):
    name: str

class UpdateCatDto(BaseModel):
    name: str | None = None
```

`cats_service.py` — CRUD methods backed by an injected `Repository` (this
resolves as soon as an `OrmModule.for_root(...)` is configured somewhere in
the app — see [Database](/techniques/database)):

```python
@Injectable()
class CatService:
    def __init__(self, repository: Repository[Cat] = InjectRepository(Cat)):
        self.repository = repository

    async def create(self, dto: CreateCatDto) -> Cat:
        return await self.repository.save(Cat(**dto.model_dump()))

    async def find_all(self) -> list[Cat]:
        return await self.repository.find()

    async def find_one(self, id: int) -> Cat:
        item = await self.repository.find_one_by({"id": id})
        if item is None:
            raise NotFoundException(f"Cat #{id} not found")
        return item

    async def update(self, id: int, dto: UpdateCatDto) -> Cat:
        await self.find_one(id)
        await self.repository.update(id, dto.model_dump(exclude_unset=True))
        return await self.find_one(id)

    async def remove(self, id: int) -> None:
        await self.find_one(id)
        await self.repository.delete(id)
```

`cats_controller.py` — full CRUD routes, wired to the service above:

```python
@Controller("cats")
class CatController:
    def __init__(self, service: CatService):
        self.service = service

    @Post()
    async def create(self, dto: CreateCatDto = Body()) -> Cat:
        return await self.service.create(dto)

    @Get()
    async def find_all(self) -> list[Cat]:
        return await self.service.find_all()

    @Get(":id")
    async def find_one(self, id: int) -> Cat:
        return await self.service.find_one(id)

    @Patch(":id")
    async def update(self, id: int, dto: UpdateCatDto = Body()) -> Cat:
        return await self.service.update(id, dto)

    @Delete(":id")
    async def remove(self, id: int) -> dict:
        await self.service.remove(id)
        return {"deleted": True}
```

`cats_module.py` — registers the entity with `OrmModule.for_feature(...)`
so `InjectRepository(Cat)` above has something to resolve:

```python
@Module(
    imports=[OrmModule.for_feature([Cat])],
    controllers=[CatController],
    providers=[CatService],
)
class CatModule:
    pass
```

<div style={{ opacity: 0.8, fontSize: "0.9em" }}>
Note: the generator uses the name you type directly for class naming
(via `to_pascal_case`) — it doesn't singularize/pluralize automatically.
`austial generate resource cats` produces `CatController`/`CatService`/`Cat`
because that's what a straightforward pascal-case of "cats" minus its
trailing `s` looks like in the generator's naming convention; check the
generated names before committing.
</div>

## App module patching

`module` and `resource` both call the same text-based patcher
(`register_module`) against your project's `src/app_module.py`:

1. Inserts `from src.modules.cats.cats_module import CatModule` into the
   existing same-package import block, re-sorted alphabetically.
2. Appends `CatModule` into the `imports=[...]` array literal on the
   `@Module(...)` decorator.

It's idempotent — running the same generator twice, or against an
already-imported module, is a no-op rather than a duplicate entry. If your
`app_module.py` has been hand-edited into a shape the patcher can't
recognize (no clear `imports=[...]` array to append to), it leaves the file
untouched rather than guessing — add the import/array entry by hand in
that case.

## File-already-exists guard

Every generator refuses to overwrite an existing file by default — it
skips and reports it — delete or rename the existing file first if you
intend to regenerate it.