lenApi

lenApi is a small ASGI framework experiment inspired by FastAPI. Pydantic v2 validates request bodies, typed path/query parameters, and response models. Routers group endpoints, while SQLAlchemy provides the ORM and request-scoped sessions.

from pydantic import BaseModel, ConfigDict
from lenapi import LenApi, Router

app = LenApi()
api = Router(prefix="/api")

class GreetingInput(BaseModel):
    name: str

class GreetingOutput(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    message: str

@api.post("/hello", response_model=GreetingOutput)
async def hello(data: GreetingInput):
    return {"message": f"Hello, {data.name}!"}

app.include_router(api)

Use lenApi's standard-library logger in application code. logger is the default lenapi logger; get_logger() returns it or a namespaced child logger:

import logging
from lenapi import get_logger, logger

logging.basicConfig(level=logging.INFO)
logger.info("Application started")

auth_logger = get_logger("auth")
auth_logger.warning("Login attempt rejected")

The logger follows Python's normal logging configuration, so developers can set levels, handlers, and formats with the logging module.

WebSockets

Register WebSocket handlers with @app.websocket(...) and use the connection helpers from lenapi.service:

from lenapi import LenApi
from lenapi.service import WebSocket, WebSocketDisconnect

app = LenApi()

@app.websocket("/ws/{room_id}")
async def room_socket(websocket: WebSocket, room_id: int):
    await websocket.accept()
    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_json({"room_id": room_id, "message": message})
    except WebSocketDisconnect:
        pass

WebSocket supports text, bytes, and JSON messages, explicit acceptance, and close codes. WebSocket paths can also be registered on a Router and mounted with include_router() or add_router().

Use the optional Faker service to generate realistic sample data. From the lenApi repository root, install it with python -m pip install -e '.[faker]':

from lenapi.service import Faker

fake = Faker("en_NG", seed=42)
print(fake.name())
print(fake.email())

The service delegates Faker providers such as name(), email(), and address() to a Faker instance. The seed makes an instance's generated sequence repeatable.

OpenAI and Ollama services

Install the optional HTTP client with python -m pip install -e '.[ai]'. Both clients are async, return the provider's JSON response, and support async context management so connections close cleanly:

from lenapi.service import OpenAI, Ollama

async with OpenAI(model="gpt-4.1-mini") as openai:
    answer = await openai.responses("Explain LenApi in one sentence.")

async with Ollama(model="llama3.2") as ollama:
    answer = await ollama.chat([
        {"role": "user", "content": "Explain LenApi in one sentence."},
    ])

OpenAI reads OPENAI_API_KEY from the environment unless api_key= is passed directly. It supports the Responses API, chat completions, and embeddings. Ollama defaults to http://127.0.0.1:11434 and supports chat, prompt generation, and embeddings; set base_url= to use another Ollama host. Ollama must be running and the requested model must be available there. These calls are non-streaming and raise HTTPX errors for unsuccessful provider responses. The OpenAI Responses API is the recommended starting point for new projects; chat completions are included for compatibility. (OpenAI API, Ollama chat API, Ollama generate API)

Payment provider plugins

Payment integrations are optional plugins with both synchronous and asynchronous APIs. Install their shared HTTPX transport with python -m pip install -e '.[payments]'. LenApi includes hosted-checkout adapters for Nomba, Flutterwave, Stripe, and Paystack; select only the provider your application uses. Credentials are read from environment variables by default, or can be passed explicitly:

from lenapi import Payments

with Payments("paystack") as payment:  # PAYSTACK_SECRET_KEY
    checkout = payment.create_checkout(
        amount="2500.00", currency="NGN", email="customer@example.com",
        reference="order-2026-0001", success_url="https://example.com/payment/return",
    )
    print(checkout["data"]["authorization_url"])
    verified = payment.verify("order-2026-0001")

For async def routes, use async with and the a-prefixed methods:

async with Payments("paystack") as payment:
    checkout = await payment.acreate_checkout(
        amount="2500.00", currency="NGN", email="customer@example.com",
        reference="order-2026-0001", success_url="https://example.com/payment/return",
    )
    verified = await payment.averify("order-2026-0001")

Provider environment variables: NOMBA_CLIENT_ID, NOMBA_CLIENT_SECRET, NOMBA_ACCOUNT_ID; FLW_SECRET_KEY; STRIPE_SECRET_KEY; and PAYSTACK_SECRET_KEY. For webhooks, configure FLW_WEBHOOK_SECRET_HASH and STRIPE_WEBHOOK_SECRET where relevant. Paystack and Stripe adapters expose verify_webhook(raw_body, signature); Flutterwave accepts its verif-hash header as signature. Nomba's adapter obtains and refreshes its client-credentials token for checkout and verification.

Adapters use the provider's hosted checkout, so LenApi does not collect or store card numbers or CVVs. create_checkout() returns the provider response, and verify() queries the provider; your application must verify transaction status, amount, currency, and reference before fulfilling an order, and record its own order/payment state. Treat webhook deliveries as untrusted until signature validation and transaction verification succeed. Provider APIs have different capabilities and field requirements, so adapter-specific options are passed through to the provider. See the Nomba checkout API, Flutterwave verification and webhooks, Stripe Checkout Sessions, and Paystack transaction API.

You can also define a router without a prefix and mount it with the Django Ninja style add_router(prefix, router) call. Prefixes with or without trailing slashes are normalized:

from lenapi import LenApi, Router

app = LenApi()
users = Router()

@users.get("/")
def list_users():
    return []

app.add_router("/users/", users)  # GET /users/

Add an IP-based cooldown to any route with api_flag=True. The client IP may call that route once per cooldown window; further requests receive 429 Too Many Requests and a Retry-After header:

@app.post("/webhooks/receive", api_flag=True, cool_time=300)
def receive_webhook(payload: dict):
    return {"received": True}

cool_time is in seconds and defaults to 300. IP cooldowns use the shared security backend so they work across server workers. Choose SECURITY_STORAGE = "database" or "redis" in project settings.py. Database mode stores state in the internal lenapi_security_store table, which is included in migrations; run python lenora.py migration after enabling it. Redis mode uses REDIS_URL and requires pip install 'lenapi[redis]' plus a running Redis server. The generated settings default to database storage and include REDIS_URL = "redis://127.0.0.1:6379/0" for switching to Redis. Set SECURITY_STORAGE = "memory" only for single-process development.

Model helpers include model.EmailField(), model.IPField(), model.PhoneField(), and model.TextField():

class Contact(model.Model):
    email = model.EmailField(unique=True)
    ip_address = model.IPField(nullable=True)
    phone = model.PhoneField(nullable=True)
    notes = model.TextField(nullable=True)

These declare database column types; validate request input with a Pydantic schema as well.

Model field helpers accept SQLAlchemy's default= option for per-field defaults, such as count = model.IntegerField(default=0) or status = model.StringField(default="new"). model.BooleanField() defaults to True; set default=False when a flag should start disabled. This default applies to every boolean field unless that field overrides it.

All model field declarations default to nullable=True and blank=True; pass nullable=False or blank=False to opt out. blank is recorded in the SQLAlchemy column's info metadata for form or schema tooling. Request-body requiredness is still controlled by the Pydantic request schema, so add a Pydantic default or optional type there when the API should accept an omitted or empty value.

Add a database index to any model field with index=True, for example email = model.EmailField(index=True). Apply the schema change with python lenora.py migration; indexes are useful for fields that are frequently filtered or searched.

Numeric helpers include model.BigIntField() (also named BigIntegerField()), the existing model.FloatField(), and model.DecimalField(max_digits=10, decimal_places=2). Decimal values use fixed precision and are represented as Python Decimal values:

class Invoice(model.Model):
    quantity = model.BigIntField()
    weight = model.FloatField()
    amount = model.DecimalField(max_digits=12, decimal_places=2)

model.UUIDField() stores a native UUID and generates a UUID4 by default. model.ImageField(upload_to="avatars") stores a relative file path in the database; the upload bytes are saved through Request.form() and UploadFile.save_image(). The generated project setting MEDIA_ROOT defaults to BASE_DIR / "media".

The repository root contains the lenApi library. Create an application scaffold and run it separately:

python -m pip install -e '.[server]'
python lenora.py startproject my_api
cd my_api
python lenora.py migration
python lenora.py runserver

The server listens on 127.0.0.1:8009 by default. The generated application contains app/models.py, settings.py, main.py, test.py, a project-local CLI launcher, alembic.ini, and a migrations/ folder. The source tree keeps working applications under app/; they are separate from the installable lenapi package.

settings.py centralizes the API title/version, docs and OpenAPI URLs, host, port, database URL, token settings, and shared security storage. app = LenApi() loads these project settings automatically, including the title displayed by Swagger. Each generated project gets a unique high-entropy SECRET_KEY; TOKEN uses it to sign tokens and configures token lifetime and issuer. Keep the key private. Changing it invalidates tokens signed with the old key. Override the host, port, or SQLite URL with LENAPI_HOST, LENAPI_PORT, and LENAPI_DATABASE_URL environment variables.

settings.py centralizes the API title/version, docs and OpenAPI URLs, host, port, database URL, token settings, and shared security storage. app = LenApi() loads these project settings automatically, including the title displayed by Swagger. Each generated project gets a unique high-entropy SECRET_KEY; TOKEN uses it to sign tokens and configures token lifetime and issuer. Keep the key private. Changing it invalidates tokens signed with the old key. Override the host, port, or SQLite URL with LENAPI_HOST, LENAPI_PORT, and LENAPI_DATABASE_URL environment variables.

Automated tests

Install the test extra and run the library and example application tests from the repository root:

python -m pip install -e '.[test]'
python -m pytest

The generate_test, run_test, and run_test_output commands are application commands. Run them from an application directory, where main.py and settings.py live. run_test_output writes a CSV report under that application directory.

For the included Student API example, run those commands from app/student_api:

cd app/student_api
python lenora.py generate_test
python lenora.py run_test

Request bodies are limited to 10 MiB by default, and lenApi allows up to 30 seconds to receive the body. Configure MAX_REQUEST_BODY_SIZE (bytes) and REQUEST_BODY_TIMEOUT_SECONDS in settings.py; set either to None to disable that limit. Oversized bodies receive HTTP 413, and bodies that exceed the receive timeout receive HTTP 408. Direct Application(...) instances accept the equivalent max_request_body_size and request_body_timeout arguments.

The runserver command looks for app in the current application's main.py. It watches application files and reloads the server when code changes. The default port is 8009. Change PORT in application settings.py, set LENAPI_PORT, or pass --port to runserver, for example python lenora.py runserver --port 8010. Host and app can also be changed with --host and --app package.module:variable; disable reload with --no-reload. Installing the package also provides the lenora runserver command.

Async database access

Routes can be async def, and lenApi can inject a SQLAlchemy AsyncSession when configured with an AsyncDatabase. The standard lenApi installation includes SQLAlchemy's asyncio extra (including greenlet) and aiosqlite, so async SQLite is ready to use. For async PostgreSQL or MySQL, install the corresponding optional driver extra.

Then configure both URLs in settings.py: DATABASE_URL is the synchronous URL used by migrations, while ASYNC_DATABASE_URL is used for awaited database requests. They should point to the same database. For SQLite, use sqlite:///... and sqlite+aiosqlite:///... respectively.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from lenapi import LenApi
from app.models import Student

app = LenApi()

@app.get("/async-students")
async def list_students(db: AsyncSession):
    result = await db.scalars(select(Student))
    return result.all()

AsyncSession is opened for the request and closed afterwards. Async model helpers use that request session automatically; pass session=db when calling them outside a request:

student = await Student.acreate(person_name="Ada", course="Python")
student = await Student.aget(id=student.id)
student.course = "Databases"
await student.asave()

students = await Student.afilter(course="Databases").order("_id").all()
deleted_count = await Student.adelete(id=student.id)

Model.adelete(...) requires filters to prevent accidental full-table deletes. You can also call await instance.adelete() or await Model.afilter(...).delete(). The synchronous Model.get(), .filter(), and .save() methods remain available. PostgreSQL and MySQL drivers are available through the async-postgres and async-mysql extras, alongside their existing synchronous driver extras for migrations.

Middleware

Add ASGI middleware with app.add_middleware. Middleware is applied in registration order, with the first middleware added wrapping the rest:

from lenapi import CORSMiddleware, RequestIDMiddleware, TimingMiddleware

app.add_middleware(RequestIDMiddleware)
app.add_middleware(TimingMiddleware)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://frontend.example"],
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["authorization", "content-type"],
)

Custom middleware follows the ASGI callable contract: Middleware(app, **options) and await app(scope, receive, send). Request IDs and elapsed time are returned in X-Request-ID and X-Process-Time-Ms.

Database transactions

Use a database transaction context when multiple writes must commit or roll back together. Use the yielded SQLAlchemy session for every operation in that block:

with app.database.transaction() as db:
    db.add(first_record)
    db.add(second_record)

For async work, use async with app.async_database.transaction() as db: and await each SQLAlchemy operation. The context commits on success, rolls back if an exception escapes, and closes the session.

Cache service

app.cache is an async JSON-value cache. Set CACHE_BACKEND in settings.py to "memory" (default), "database", or "redis". Database-backed cache uses the shared lenapi_security_store table, so run python lenora.py migration first. Redis mode uses REDIS_URL and requires pip install '.[redis]'.

await app.cache.set("profile:42", {"name": "Ada"}, ttl=300)
profile = await app.cache.get("profile:42")
await app.cache.delete("profile:42")

Durable background jobs

BackgroundTasks remains useful for short in-process work after a response. For a worker queue, BackgroundQueue supports database, redis, or cache/memory backends. Database mode requires the queue table migration; Redis mode requires pip install '.[redis]' and Redis persistence configured for the durability you need. The cache/memory backend is process-local and ephemeral: queued jobs are lost when the process exits and cannot be shared with a separate worker process. Jobs on database and Redis backends are delivered at least once; write task handlers to be idempotent.

from lenapi import BackgroundQueue

# Database backend:
queue = BackgroundQueue(app.database)
# Redis backend: queue = BackgroundQueue(backend="redis", redis_url=settings.REDIS_URL)
# Process cache: queue = BackgroundQueue(backend="cache")

@queue.task("send_welcome")
async def send_welcome(email: str):
    ...

job_id = await queue.enqueue("send_welcome", "ada@example.com")

Run a worker in a separate process after applying migrations:

# worker.py
import asyncio
from main import app
from lenapi import BackgroundQueue

queue = BackgroundQueue(app.database)

@queue.task("send_welcome")
async def send_welcome(email: str):
    ...

asyncio.run(queue.run_worker())

The API process and worker must register the same task names and handlers. run_once() is available for one-job execution in tests or custom worker loops; status(job_id) returns the persisted job status.

Swagger UI is available at http://127.0.0.1:8009/docs/, and the generated OpenAPI document is at http://127.0.0.1:8009/openapi.json. These are enabled by default. Set API_TITLE, API_VERSION, DOCS_URL, and OPENAPI_URL in settings.py to configure the Swagger metadata and paths, or override them directly with LenApi(title="My API", version="1.0.0", docs_url="/docs/", openapi_url="/openapi.json").

The sample app exposes the model queries as HTTP routes: POST /classrooms creates a classroom, GET /classrooms lists them, and POST /students accepts an optional classroom_id. POST /students/batch creates multiple students, PUT /students/batch updates multiple students by ID, and DELETE /students/batch deletes multiple students by ID. Batch requests are all-or-nothing: unknown IDs, duplicate names, or a protected deletion reject the whole operation. PUT /students/{student_id} updates one student; it returns HTTP 404 if the ID does not exist. GET /students/{student_id}/classroom returns the related classroom. Other student routes list, search, and check existence. These routes appear in Swagger at /docs/.

Batch request examples:

{"students": [
  {"person_name": "Ada", "course": "Python"},
  {"person_name": "Grace", "course": "Databases"}
]}

Use the same students wrapper for batch updates, with an id on each item. Batch deletion accepts {"ids": [1, 2]}.

Dependencies

Use Depends to share logic across routes. A dependency can accept the same typed path/query parameters and Request object as a route, and may be synchronous or asynchronous:

from lenapi import Application, Depends

app = Application()

def current_user(user_id: int):
    # Replace this with real authentication logic.
    return {"id": user_id, "name": "Ada"}

@app.get("/profile")
def profile(user: dict = Depends(current_user)):
    return user

Dependencies are cached once per request, so if multiple parameters use the same dependency callable, it runs once.

JWT bearer authentication

JWTAuth signs and verifies HS256 access tokens and works as a route dependency. It validates the signature, expiration, optional not-before time, and configured issuer. To add credential and refresh endpoints, mount the router with your own credential checker. The checker must validate the password (using a password hash in a real app) and return a string user ID or (user_id, claims); invalid credentials return None:

def authenticate_user(username: str, password: str):
    user = find_user(username)
    if user is None or not verify_password(password, user.password):
        return None
    return str(user.id), {"role": user.role}

app.include_router(jwt_auth.router(authenticate_user))

This adds POST /auth/token with {"username": "...", "password": "..."} and POST /auth/refresh with {"refresh_token": "..."}. Both return an access_token and refresh_token; access tokens expire after TOKEN["LIFETIME_SECONDS"] and refresh tokens after TOKEN["REFRESH_LIFETIME_SECONDS"]. Refresh tokens are rejected by protected routes.

JWTAuth does not authenticate usernames or passwords itself; your callback owns that check. Keep a high-entropy secret private and use HTTPS in deployed applications:

Choose the application's default authentication in the project settings.py:

AUTHENTICATION = "basic_login"  # choose one: jwt, api_key, basic_login, opaque_bearer, or session
BASIC_AUTH_FIELD = "email"  # username, email, phone, or another User field

The HTTP Basic username slot is treated as the configured identifier, so a client can send auth=(email, password) or auth=(phone, password). For a custom identifier, add that field to User and migrate the database; then set BASIC_AUTH_FIELD to the model field name.

The generated main.py uses that setting to protect /auth/profile; Swagger shows the selected scheme. JWT mode also registers /auth/token and /auth/refresh. create_authenticator(settings, ...) builds the selected dependency; provide callbacks for database-backed credentials and stores:

from lenapi import Depends, create_authenticator
import settings

auth = create_authenticator(settings, authenticate=authenticate_user)

@app.get("/protected")
def protected(user=Depends(auth)):
    return {"authenticated": True}

if settings.AUTHENTICATION == "jwt":
    app.include_router(auth.router(authenticate_user))

The primary authentication setting chooses one method for the scaffold's default protected route: jwt, api_key, basic_login, opaque_bearer, or session. Session cookies, HMAC, and OIDC can also be attached as additional callable dependencies on selected routes. This lets an app use API keys generally and require a session cookie on a browser route, or require both on one route:

from lenapi import Depends, hmac_auth, oidc_auth, opaque_bearer_auth, session_auth

session = session_auth(store=app.security_store, cookie_name="session_id")
stored_token = opaque_bearer_auth(store=app.security_store)
signed_request = hmac_auth(settings.HMAC_SECRET, nonce_store=app.security_store)
provider_token = oidc_auth(oidc_token_lookup)

@app.get("/browser/private")
def browser_private(api_identity=Depends(auth), session_identity=Depends(session)):
    return {"api_user": api_identity, "session_user": session_identity}

@app.get("/browser/session-only")
def browser_session_only(user=Depends(session)):
    return {"user": user}

@app.post("/webhooks/provider")
async def provider_webhook(identity=Depends(signed_request)):
    return {"accepted": True}

@app.get("/federated/private")
def federated_private(identity=Depends(provider_token)):
    return {"identity": identity}

SECURITY_STORAGE selects the shared store used by IP flags, sessions, opaque bearer tokens, and HMAC nonce replay checks. The store exposes async get(key), set(key, value, ttl=...), delete(key), and atomic claim(key, ttl) methods. For a login flow, create a random session ID, store the JSON-serialized identity as await app.security_store.set(f"session:{session_id}", json.dumps(identity), ttl=3600), and set it in a cookie with Response.set_cookie(..., httponly=True, secure=True, samesite="lax"). Opaque bearer tokens can be stored similarly under opaque_bearer:{token}. oidc_token_lookup(token) must validate the provider's issuer, audience, signature, and expiry using its supported OIDC library. lenApi supplies the dependency helpers; the app owns the login/callback and credential lifecycle. These extra dependencies can be combined with the primary auth dependency when a route requires both.

For HMAC, configure HMAC_SECRET with at least 32 bytes, then clients can generate the header signature with hmac_request_signature(secret, method, path, timestamp, body, nonce=nonce) and send it with X-LenAPI-Timestamp, X-LenAPI-Nonce, and X-LenAPI-Signature. Signatures older than the configured tolerance and repeated nonces are rejected. The replay cache is process-local; multi-worker deployments should replace it with a shared nonce store.

HMAC mode signs the request timestamp, method, path plus query, body digest, and a unique nonce. Configure HMAC_SECRET with at least 32 bytes, then clients can generate the header signature with hmac_request_signature(secret, method, path, timestamp, body, nonce=nonce) and send it with X-LenAPI-Timestamp and X-LenAPI-Nonce. Signatures older than the configured tolerance and repeated nonces are rejected. The replay cache is process-local; multi-worker deployments should replace it with a shared nonce store.

Authorization

Authentication confirms who called the API; authorization checks what they may do. Use the reusable dependencies for roles, permissions, and ownership checks. The auth dependency must return identity claims, such as a JWT role/permissions, or a session/API-key mapping with a user ID.

from lenapi import Depends, require_permission, require_role

admin_access = require_role("admin", auth=auth)
write_students = require_permission("students:write", auth=auth)

@app.delete("/admin/reports")
def delete_reports(identity=Depends(admin_access)):
    return {"deleted": True}

@app.post("/students")
def create_student(identity=Depends(write_students)):
    return {"created_by": identity["sub"]}

For object-level authorization, load the requested row and check that its owner matches the authenticated user's sub, user_id, or id. The dependency returns the row on success and returns 404 when the row is missing or belongs to someone else:

from lenapi import Depends, require_owner

document_access = require_owner(
    lambda document_id: Document.get(id=int(document_id)).get(),
    auth=auth,
    path_param="document_id",
    owner_field="user_id",
)

@app.get("/documents/{document_id}")
def read_document(document=Depends(document_access)):
    return document

CSRF protection for cookie-authenticated routes

Create a CSRFProtect instance with a secret of at least 32 bytes. Provide its signed token in a readable cookie, then require the same token in X-CSRF-Token on write requests. Apply it alongside authentication to routes that accept browser cookies:

from lenapi import CSRFProtect, Depends, JSONResponse

csrf = CSRFProtect(settings.SECRET_KEY, secure=True, samesite="lax")

@app.get("/auth/csrf")
def issue_csrf_token():
    response = JSONResponse({"message": "CSRF token issued"})
    csrf.set_cookie(response)
    return response

@app.post("/account/email")
def change_email(user=Depends(auth), _csrf=Depends(csrf)):
    return {"updated": True}

Safe methods do not require the header; state-changing methods do. SameSite is additional protection and does not replace this token check. Use HTTPS so the CSRF cookie can be Secure.

API_KEY is generated uniquely for each new project. The scaffold's Basic/JWT credential callback looks up User in the database and verifies its hashed password. Use make_password(password) to hash a password and check_password(password, stored_password) to verify it; the latter returns a boolean. Never store plaintext passwords. The older hash_password() and verify_password() names remain available as aliases. Use HTTPS for Basic credentials and bearer/API key requests outside local development.

When your registration or trusted user-provisioning flow creates a user, store the hash:

from lenapi import make_password, model

class User(model.Model):
    username = model.StringField(max_length=120, unique=True)
    email = model.StringField(max_length=254, unique=True, nullable=True)
    first_name = model.StringField(max_length=150, default="")
    last_name = model.StringField(max_length=150, default="")
    password = model.StringField(max_length=255)
    api_key = model.StringField(max_length=64, unique=True, nullable=True)
    phone_number = model.StringField(max_length=30, nullable=True)

# In a registration handler, after validating the submitted password:
user = User.create(
    username=username,
    password=make_password(password),
)

Add user attributes by declaring more fields on User, then create and apply a migration. To accept a new field during registration, add it to the Pydantic registration schema and pass it to User.create() in the registration handler. Keep sensitive fields such as passwords out of public response schemas.

When settings.AUTHENTICATION = "api_key", registration generates a unique, hyphen-free UUIDv5 key for the new user and returns it once as plain text in the registration response. The user.api_key column remains nullable for installations using another authentication mode; in API key mode new registered users get a key automatically. lenApi stores a SHA-256 digest in that column, so the original key cannot be recovered later. Send the returned key on authenticated requests with X-API-Key: <key>. The API key authenticator resolves the digest back to its user and returns that user's identity. Existing users need to be issued a key through a key-rotation/provisioning flow before they can use API-key authentication.

The scaffold also exposes POST /auth/register, which accepts username, email, first_name, last_name, password, and confirm_password. It validates that the passwords match, stores only the hashed password, and never stores confirm_password. The password encoding uses Django's pbkdf2_sha256 format, so Django's password hashers can verify it; this is one-way hashing, not reversible encryption. See Django password management for the encoded format and verification helpers.

Cookie-based browser sessions

lenApi routes can return JSON and set one or more cookies on the same response. The browser sends those cookies on later requests, which you can read with request.cookies:

from lenapi import Depends, HTTPException, JSONResponse, Request

@app.post("/auth/cookie-login")
def cookie_login(data: LoginRequest):
    user_id = authenticate_user(data.username, data.password)
    if user_id is None:
        raise HTTPException(401, "Invalid username or password")

    response = JSONResponse({"message": "Signed in"})
    response.set_cookie(
        "access_token",
        authenticator.create_access_token(user_id),
        max_age=1800,
        httponly=True,
        secure=True,  # Use HTTPS in deployed environments.
        samesite="lax",
        path="/",
    )
    return response

def current_user(request: Request):
    token = request.cookies.get("access_token")
    if not token:
        raise HTTPException(401, "Sign in required")
    try:
        return authenticator.decode_token(token)
    except ValueError as exc:
        raise HTTPException(401, "Invalid or expired session") from exc

@app.get("/account")
def account(user=Depends(current_user)):
    return {"user_id": user["sub"]}

@app.post("/auth/logout")
def logout():
    response = JSONResponse({"message": "Signed out"})
    response.delete_cookie("access_token", path="/")
    return response

Cookies are convenient for browser clients; native apps and server-to-server clients can keep using Authorization: Bearer .... For cookie authentication, use HTTPS, HttpOnly, and an appropriate SameSite setting, and add CSRF protection for state-changing browser requests. If your frontend is on another origin, configure CORS to allow credentials and use credentials: "include" in browser fetch calls.

import os

from lenapi import Depends, JWTAuth

jwt_auth = JWTAuth(
    secret_key=os.environ["LENAPI_JWT_SECRET"],
    token_lifetime_seconds=1800,
    refresh_token_lifetime_seconds=2592000,
    issuer="my-api",
)

# After checking the user's password in your login handler:
access_token = jwt_auth.create_access_token(
    str(user.id),
    claims={"role": user.role},
)

@app.get("/account")
def account(claims: dict = Depends(jwt_auth)):
    return {"user_id": claims["sub"], "role": claims.get("role")}

Send the token with Authorization: Bearer <access_token>. Missing, invalid, or expired tokens receive HTTP 401. Protected routes also get a Bearer authorization scheme in the generated OpenAPI document, so Swagger UI displays its Authorize control.

Use Body() with Any when an endpoint should accept and return arbitrary JSON without Pydantic validation:

from typing import Any
from lenapi import Body

@app.post("/webhooks/events")
async def receive_event(payload: Any = Body()):
    return payload

For image uploads, send multipart/form-data and read the UploadFile from Request.form(). save_image() accepts JPEG, PNG, GIF, WebP, or BMP only, checks the file signature against its content type, uses a generated safe filename, and returns the filename to store in the model:

from pathlib import Path
from lenapi import Request, UploadFile, model
import settings

class Photo(model.Model):
    image = model.ImageField(upload_to="photos")

@app.post("/photos")
async def upload_photo(request: Request):
    form = await request.form()
    image = form.get("image")
    if not isinstance(image, UploadFile):
        return {"error": "An image file is required"}
    upload_to = Photo.__table__.c.image.info["upload_to"]
    filename = await image.save_image(Path(settings.MEDIA_ROOT) / upload_to)
    photo = Photo(image=f"{upload_to}/{filename}")
    photo.save()
    return {"id": photo.id, "image": photo.image}

The stored path is relative. LenApi serves files from MEDIA_ROOT under MEDIA_URL (default /media/), so a saved path such as photos/<filename>.jpg is available at /media/photos/<filename>.jpg. Set both MEDIA_ROOT and MEDIA_URL in settings.py; media paths are restricted to files inside MEDIA_ROOT, and directory listings are not exposed. Request body size limits apply to multipart uploads too.

Webhooks and background work

Webhook senders retry deliveries, so verify the provider signature and deduplicate with its event ID. webhook_auth() verifies an HMAC-SHA256 signature over the exact raw body. If a timestamp header is configured, it verifies freshness and signs timestamp + b"." + body. Give the dependency the shared security store to acknowledge duplicate event IDs with HTTP 200 without running the route handler again:

from typing import Any
from lenapi import BackgroundTasks, Body, Depends, webhook_auth

webhook = webhook_auth(
    settings.WEBHOOK_SECRET,
    security_store=app.security_store,
    provider="payments",
    signature_header="x-webhook-signature",
    event_id_header="x-webhook-id",
    timestamp_header="x-webhook-timestamp",
)

def process_event(event_id, payload):
    # Process the event; open a fresh database session here if needed.
    ...

@app.post("/webhooks/payments")
async def payments_webhook(
    payload: Any = Body(),
    verified=Depends(webhook),
    background_tasks: BackgroundTasks = None,
):
    background_tasks.add_task(process_event, verified["event_id"], payload)
    return {"received": True}

BackgroundTasks.add_task(function, *args, **kwargs) runs work after lenApi sends the response. Synchronous functions run in a worker thread; async functions are awaited. This built-in runner is in-process and best for short follow-up work. A process crash can lose queued work, so use a durable queue for jobs that must survive restarts. The webhook dependency claims an event ID for seven days by default; configure deduplication_ttl if the provider retries for a different period. If the security store is database-backed, run python lenora.py migration first.

Databases

Application() uses SQLite by default and creates the lenapi.db file in the current working directory. Tables are managed by migrations, not created automatically at startup. Database accepts a SQLAlchemy URL when you want a different relational database. SQLAlchemy-backed models support SQLite, MySQL, and PostgreSQL. MongoDB is available through a separate optional Pydantic document API; it does not use SQL migrations or the SQLAlchemy Model base. Install optional drivers with pip install '.[mysql]', pip install '.[postgres]', or pip install '.[mongodb]'.

MongoDB documents

Install the official PyMongo driver extra with python -m pip install -e '.[mongodb]'. Create a MongoDatabase using a MongoDB URI and database name, then bind Pydantic document classes to it. The default collection name is the lower-case document class name; set collection_name to override it. MongoDB creates collections on first write, so there is no Alembic migration step for them.

from lenapi import LenApi, MongoDatabase, MongoDocument

mongo = MongoDatabase(
    "mongodb://127.0.0.1:27017",
    "school",
    serverSelectionTimeoutMS=5000,
)

class Student(MongoDocument):
    name: str
    course: str
    collection_name = "students"

Student.bind(mongo)
app = LenApi(mongodb=mongo)

# Synchronous CRUD
student = Student.create(name="Ifeanyi", course="Python")
found = Student.get(id=student.id)
students = Student.filter(course="Python")
student.course = "MongoDB"
student.save()
student.delete()

# Async CRUD
student = await Student.acreate(name="Ada", course="Python")
found = await Student.aget(id=student.id)
students = await Student.afilter(course="Python")
await student.asave()
await student.adelete()

MongoDatabase.ping() and await MongoDatabase.aping() check connectivity. Configure MONGODB_URI and MONGODB_DATABASE in settings.py to have LenApi() configure app.mongodb automatically. MongoDB uses the official PyMongo synchronous and AsyncMongoClient APIs; see the PyMongo driver guide.

Put application models in app/models.py and import that module from main.py. The migration command loads main:app, so it discovers those model classes before creating a migration. The library's Model base supplies an integer id:

from lenapi import model

class Classroom(model.Model):
    name = model.StringField(max_length=120)

class Student(model.Model):
    person_name = model.StringField(max_length=120, unique=True)
    course = model.StringField(max_length=120)
    classroom_id = model.ForeignKeyField("classroom.id", nullable=True, model=model.null)

class Course(model.Model):
    name = model.StringField(max_length=120)

class Teacher(model.Model):
    name = model.StringField(max_length=120)
    courses = model.ManyToManyField("Course")

class User(model.Model):
    username = model.StringField(max_length=120, unique=True)
    email = model.StringField(max_length=254, unique=True, nullable=True)
    first_name = model.StringField(max_length=150, default="")
    last_name = model.StringField(max_length=150, default="")
    password = model.StringField(max_length=255)

class Verification(model.Model):
    user_id = model.ForeignKeyField("user.id", model=model.cascade.delete)

class AuditLog(model.Model):
    user_id = model.ForeignKeyField("user.id", nullable=True, model=model.null)

class Invoice(model.Model):
    user_id = model.ForeignKeyField("user.id", model=model.protected)

Table names are inferred from model class names in lowercase (Student becomes student); no table-name declaration is needed. Import model from lenapi and access the model base class, fields, and relationship behaviors through it, such as model.Model, model.StringField, and model.ForeignKeyField.

From the project root, run one command to detect model changes, write a migration under migrations/versions/, and apply it to the configured database:

python lenora.py migration

After changing models, run the same command again; it creates and applies the next migration. Use --message "describe change" to name a migration. Existing unversioned SQLite databases that already match the models are adopted without changing their data. Inside a request, models provide these query helpers:

student = Student.create(person_name="Ifeanyi", course="Python")
students = Student.get.all()
specific_student = Student.get(person_name="Ifeanyi").get()
exists = Student.get(person_name="Ifeanyi").exists()
name = Student.get.unique("person_name").get(person_name="Ifeanyi")
summary = Student.get("person_name", "course").get(person_name="Ifeanyi")
matching_students = Student.filter(course="Python").values().order("_id").all()
searched_students = Student.search(person_name="ifea").all()
matching_name_or_course = Student.search_any("python", "person_name", "course").all()
first_page = Student.filter(course="Python").all(limit=20, offset=0)
second_page = Student.filter(course="Python").limit(20).offset(20).all()
latest_students = Student.get.all(limit=20, offset=40)
all_students_page = Student.all(limit=20, offset=60).all()

new_student = Student(person_name="Ada", course="Databases")
new_student.save()  # commits and refreshes the instance

Student.get(...) builds a single-row lookup; call .get() for the full model instance or .exists() for a boolean. Multiple matches raise a multiple-results error. Pass column names to Student.get(...) to return only those fields, either as a scalar or a dictionary. filter(...) accepts exact field matches and Django-style lookups such as course__icontains="py", id__gte=10, id__in=[1, 2], and classroom_id__isnull=True. Supported lookups include exact, iexact, ne, gt, gte, lt, lte, in, notin, contains, icontains, startswith, istartswith, endswith, iendswith, and isnull. search(...) applies case-insensitive substring matching across all provided string fields (all terms must match); search_any(query, *fields) searches one term across fields and returns rows matching any field. Use .values() to select all columns as dictionaries or .values("field", ...) for selected columns; chain .order("field") (prefix with - for descending) and then .all() to execute. Annotate a route argument as SQLAlchemy's Session when you need transaction control directly.

The sample API exposes these operations as GET /students/filter?course__icontains=py&id__gte=1&order_by=-id&limit=20&offset=0. Use comma-separated values for __in, for example GET /students/filter?id__in=1,2,3. Search using GET /students/search?q=python (searches person_name and course by default), or choose fields with GET /students/search?q=ada&fields=person_name,course.

Use Count, Sum, Min, Max, and Avg with Model.aggregate() or a filtered queryset. An ungrouped aggregate returns one dictionary; group_by returns a list of dictionaries, one per group:

from lenapi import model

class Purchase(model.Model):
    user_id = model.ForeignKeyField("user.id")
    amount = model.DecimalField(max_digits=12, decimal_places=2)

total_spent = Purchase.aggregate(total=model.Sum("amount"))
# {"total": Decimal("...")}

spending_by_user = Purchase.aggregate(
    group_by="user_id",
    total=model.Sum("amount"),
    purchases=model.Count(),
)
largest_purchase = Purchase.filter(user_id=1).aggregate(maximum=model.Max("amount"))
purchase_count = Purchase.count()

Use Purchase.aaggregate(...), Purchase.acount(), or await Purchase.afilter(...).aggregate(...) with async database access.

Delete a unique matching record with Student.get(person_name="Ifeanyi").delete(); it returns True if a row was removed. A retrieved instance also supports student.delete(); .delet() is provided as an alias. Student.filter(course="Python").delete() deletes matching rows and returns the count. Student.all().delete() drops the entire student table; a migration is required before querying it again.

Student.get.all() and Student.filter(...).all() return list-like collections. Collections support .delete() for their loaded rows, but not .save(); save an individual instance after changing it. Only Student.all().delete() drops the table.

Paginate list queries with .all(limit=..., offset=...), or chain .limit(...).offset(...).all(). Student.get.all(limit=20, offset=40) paginates model manager results, and Student.all(limit=20, offset=40).all() paginates a full-table query. Limits and offsets must be non-negative integers.

Update a record by assigning fields on a unique lookup result and saving it:

student = Student.get(person_name="Ifeanyi")
student.course = "Java"
student.save()

The model also provides common get/create/update/delete shortcuts. defaults contains the fields to write, while keyword arguments identify the row:

student = Student.get_or_404(person_name="Ifeanyi")

student = Student.create_update(
    defaults={"course": "Python"},
    person_name="Ifeanyi",
)

student = Student.update_or_404(
    defaults={"course": "Java"},
    person_name="Ifeanyi",
)

Student.delete_or_404(person_name="Ifeanyi")

student, created = Student.get_or_create(
    defaults={"course": "Python"},
    person_name="Ifeanyi",
)

get_or_404, update_or_404, and delete_or_404 raise HTTPException(404, ...) when no record matches, which lenApi returns as an HTTP 404 response. create_update creates a missing row or updates an existing one and returns the model instance. get_or_create returns (instance, created), where created is a boolean.

For a declared foreign-key column, .special("field_name") resolves its related model and exposes it under that same field name:

student = Student.get(person_name="Ifeanyi").special("classroom_id")
classroom_name = student.classroom_id.name

Here, the field name must be a ForeignKeyField targeting a registered lenApi model. The value at student.classroom_id is the related model instance; the project example defines this as classroom_id.

Choose what happens to dependent records when their referenced model is deleted with model:

• CASCADE: deleting a user also deletes its verification rows.
• SET_NULL: keep the dependent row and set its foreign key to NULL; declare nullable=True.
• PROTECT: prevent deletion while dependent rows still reference the record (the default).

Use exactly one policy value per field, such as model=cascade.delete, model=null, or model=protected. Python's or operator is not a list of alternatives: cascade.delete or null or protected always evaluates to its first truthy value. The earlier on_delete= keyword remains supported as an alias.

These policies are enforced by database foreign-key actions. SQLite foreign-key enforcement is enabled automatically. Model.delete() reports a protected deletion as HTTP 409; apply a migration after changing an existing foreign key's policy.

ManyToManyField creates an association table in model metadata, so python lenora.py migration includes it in migrations. The default table name combines the source table and relationship field (teacher_courses in the example). The relationship is a set-like collection; add or remove related model instances, then save the owning model to commit:

teacher = Teacher.get_or_404(id=1)
course = Course.get_or_404(id=1)
teacher.courses.add(course)
teacher.save()

teacher.courses.set([course])  # replace all related courses

courses = teacher.courses.all()
teacher.courses.remove(course)
teacher.save()

Pass table_name="teacher_courses" to ManyToManyField to choose the association table name. Use back_populates="teachers" to pair it with a reciprocal teachers = ManyToManyField("Teacher", table_name="teacher_courses", back_populates="courses") field on Course.

from lenapi import Application, Database

database = Database("sqlite:///app.db")
app = Application(database=database)

Swap the URL to use another database:

Database("mysql+pymysql://user:password@localhost/mydb")
Database("postgresql+psycopg://user:password@localhost/mydb")

A Session annotated route argument receives a session that lenApi closes after the request. Commit or roll back in the handler (or in your own service layer):

from sqlalchemy.orm import Session
from lenapi import Application, Database

app = Application(database=Database("sqlite:///app.db"))

@app.get("/health")
def health(db: Session):
    return {"database_ready": db.is_active}

The engine is disposed on ASGI lifespan shutdown. For response serialization from SQLAlchemy objects, define a Pydantic response schema with ConfigDict(from_attributes=True) and set it as response_model on the route.

Current shape

• ASGI HTTP request/response handling
• @app.get, @app.post, @app.put, @app.delete, and generic @app.route
• {parameter} path segments and Pydantic-validated path/query types
• async or sync route functions
• Request access to query parameters, raw body, and JSON body
• Pydantic request body models and response models
• mountable Router groups
• optional request-scoped SQLAlchemy Session injection
• python lenora.py runserver development command (default port 8009); python lenora.py migration generates and applies model migrations
• Swagger UI at /docs/ and generated OpenAPI 3.1 at /openapi.json
• JSON responses by default, with Response and JSONResponse for explicit control
• Request cookie parsing through request.cookies, plus Response.set_cookie() and Response.delete_cookie()

The installable library lives in lenapi/; runnable applications and their migrations are kept separately under app/. The standalone wheel and source download contain the framework, packaging metadata, documentation, and library tests, but omit application code and local databases.