REST API Reference

RestEndpoint is the single building block of every LightAPI application. A subclass simultaneously acts as the SQLAlchemy ORM model, the Pydantic v2 schema, and the HTTP endpoint handler.

`RestEndpoint`

from lightapi import RestEndpoint, Field

Declaring Fields

from typing import Optional
from decimal import Decimal
from lightapi import RestEndpoint, Field

class ProductEndpoint(RestEndpoint):
    name:        str            = Field(min_length=1, max_length=200)
    price:       Decimal        = Field(ge=0, decimal_places=2)
    category:    str            = Field(min_length=1)
    description: Optional[str] = None
    in_stock:    bool           = Field(default=True)
    supplier_id: int            = Field(foreign_key="suppliers.id")

Python type → SQLAlchemy column mapping:

Python annotation	Column type	Nullable
`str`	`VARCHAR`	No
`Optional[str]`	`VARCHAR`	Yes
`int`	`INTEGER`	No
`Optional[int]`	`INTEGER`	Yes
`float`	`FLOAT`	No
`bool`	`BOOLEAN`	No
`datetime`	`DATETIME`	No
`Decimal`	`NUMERIC(scale=N)`	No
`UUID`	`UUID`	No

LightAPI-specific Field() kwargs (passed in json_schema_extra):

Kwarg	Effect
`foreign_key="table.col"`	Adds `ForeignKey` constraint
`unique=True`	Adds `UNIQUE` constraint
`index=True`	Adds a database index
`exclude=True`	Field is excluded from DB and schema entirely
`decimal_places=N`	A `Decimal` with more than N decimal places is rejected (422); the column type stays `NUMERIC`

Auto-injected Columns

Every RestEndpoint subclass automatically receives these columns:

Column	Type	Value
`id`	`Integer` PK	autoincrement
`created_at`	`DateTime`	`utcnow` on insert
`updated_at`	`DateTime`	`utcnow` on insert and update
`version`	`Integer`	`1` on insert; incremented on `PUT`/`PATCH`

These are excluded from create/update input schemas but always included in responses.

`Meta` Inner Class

class MyEndpoint(RestEndpoint):
    class Meta:
        authentication = Authentication(backend=..., permission=...)
        filtering      = Filtering(backends=[...], fields=[...], search=[...], ordering=[...])
        pagination     = Pagination(style="page_number"|"cursor", page_size=20)
        serializer     = Serializer(fields=[...]) | Serializer(read=[...], write=[...])
        cache          = Cache(ttl=60)
        reflect        = False | True | "partial"
        table          = "custom_table_name"
        table_name     = "custom_table_name"   # alias for table

Sync CRUD Methods

Override these methods to customise sync behaviour. Used when a sync engine is passed (or when the override is def, not async def, even on an async engine app):

Method	Signature	Default behaviour
`queryset`	`(self, request)`	`select(cls._model_class)`
`list`	`(self, request)`	Paginated `SELECT *`
`retrieve`	`(self, request, pk)`	`SELECT WHERE id=pk`
`create`	`(self, data)`	`INSERT RETURNING`
`update`	`(self, data, pk, partial)`	Optimistic-lock `UPDATE`
`destroy`	`(self, request, pk)`	`DELETE WHERE id=pk`

PATCH and Optional fields

When partial=True (a PATCH request), only fields present in the request body are updated. Sending an explicit null value for a nullable (Optional[...]) field clears it to NULL in the database:

# subtitle is Optional[str] — sending null clears the column
PATCH /products/42
{"subtitle": null, "version": 3}
# → 200 {"id": 42, "subtitle": null, ...}

Sending null for a non-nullable (required) field is silently ignored — the field retains its current database value.

Async CRUD Helpers

When an AsyncEngine is passed to LightApi, LightAPI dispatches through these internal async methods. You can call them directly from async def overrides:

Helper	Return	Description
`await self._list_async(request)`	`JSONResponse`	Paginated list with filter/ordering
`await self._retrieve_async(request, pk)`	`JSONResponse` / 404	Single row
`await self._create_async(data: dict)`	`JSONResponse(201)`	INSERT + flush + refresh
`await self._update_async(data, pk, partial=False)`	`JSONResponse` / 409 / 404	Optimistic-lock UPDATE
`await self._destroy_async(request, pk)`	`Response(204)` / 404	DELETE
`self.background(fn, args, *kwargs)`	`None`	Schedule a post-response task

Example: Async Override with Background Task

import json
from starlette.requests import Request
from starlette.responses import Response
from lightapi import RestEndpoint, Field

async def on_create(item_id: int) -> None:
    ...   # send notification, write audit log

class OrderEndpoint(RestEndpoint):
    amount: float = Field(ge=0)

    async def post(self, request: Request) -> Response:
        data = json.loads(await request.body())
        resp = await self._create_async(data)
        if resp.status_code == 201:
            self.background(on_create, json.loads(resp.body)["id"])
        return resp

HTTP Method Overrides

Define get, post, put, patch, or delete as either sync or async — LightAPI detects and dispatches accordingly:

class MyEndpoint(RestEndpoint):
    name: str = Field(min_length=1)

    # Sync override
    def get(self, request):
        return self.list(request)

    # Async override — detected via asyncio.iscoroutinefunction
    async def post(self, request):
        import json
        return await self._create_async(json.loads(await request.body()))

HttpMethod Mixins

Restrict which HTTP verbs are available:

from lightapi import RestEndpoint, HttpMethod, Field

class ReadOnlyEndpoint(RestEndpoint, HttpMethod.GET):
    name: str = Field(min_length=1)

class CreateOnlyEndpoint(RestEndpoint, HttpMethod.POST):
    name: str = Field(min_length=1)

class FullCRUDEndpoint(
    RestEndpoint,
    HttpMethod.GET, HttpMethod.POST,
    HttpMethod.PUT, HttpMethod.PATCH, HttpMethod.DELETE,
):
    name: str = Field(min_length=1)

Unregistered methods return 405 Method Not Allowed with an Allow header listing the registered verbs.

Class Attributes

Attribute	Type	Description
`_model_class`	`type`	SQLAlchemy-mapped class
`_meta`	`dict`	Parsed `Meta` configuration
`_allowed_methods`	`set[str]`	Registered HTTP verbs
`__schema_create__`	Pydantic model	Input schema for POST/PUT/PATCH
`__schema_read__`	Pydantic model	Output schema for GET responses
`_background`	`BackgroundTasks \\| None`	Starlette BackgroundTasks (set per-request)
`_current_request`	`Request \\| None`	Current request (set per-request)

Error Responses

Scenario	Status	Body
Validation failure	`422`	`{"detail": [...pydantic errors...]}`
Not found	`404`	`{"detail": "not found"}`
Optimistic lock conflict	`409`	`{"detail": "version conflict"}`
Constraint violation (unique, foreign key)	`409`	`{"detail": "constraint violation"}`
Auth failure (JWT)	`401`	`{"detail": "Authentication credentials invalid."}` with `WWW-Authenticate: Bearer`
Auth failure (Basic)	`401`	`{"error": "authentication failed"}` with `WWW-Authenticate: Basic realm="Restricted Area"`
Permission denied	`403`	`{"detail": "Insufficient permissions"}`
Method not registered	`405`	`{"detail": "Method Not Allowed. Allowed: GET, POST"}`

Reflection

Set Meta.reflect = True to map an endpoint to an existing database table instead of creating a new one. LightAPI inspects the table schema at startup and builds the Pydantic schemas from the live column definitions.

class LegacyOrderEndpoint(RestEndpoint):
    class Meta:
        reflect = True
        table = "orders"   # exact table name in the database

When reflect is True:

The table must already exist — LightAPI does not call CREATE TABLE.
Fields declared on the class are ignored; columns come from the reflected table.
All auto-injected columns (id, created_at, updated_at, version) are optional — if they are not present in the real table they are simply absent from the schema.

In YAML configuration:

endpoints:
  - route: /orders
    reflect: true
    meta:
      table: orders

Session Helpers

Exported from lightapi for use outside of endpoint methods:

from lightapi import get_sync_session, get_async_session

`get_sync_session(engine)`

Context manager; commits on clean exit, rolls back on exception.

from sqlalchemy import create_engine, select
from lightapi import get_sync_session

engine = create_engine("sqlite:///app.db")
with get_sync_session(engine) as session:
    rows = session.execute(select(MyModel)).scalars().all()

`get_async_session(engine)`

Async context manager; await commit on clean exit, await rollback on exception.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from lightapi import get_async_session

engine = create_async_engine("postgresql+asyncpg://...")
async with get_async_session(engine) as session:
    rows = (await session.execute(select(MyModel))).scalars().all()