Entities

Entities are Nitro's core abstraction. They combine data validation (Pydantic), persistence (SQLAlchemy), and business logic in a single class. Instead of scattering logic across services, controllers, and models, you consolidate everything into rich domain objects that manage themselves.

Defining an Entity

An entity is a Python class that extends Entity with table=True for database backing:

from nitro import Entity

class User(Entity, table=True):
    name: str
    email: str
    role: str = "member"

    def promote(self):
        """Business logic lives in entity methods."""
        self.role = "admin"
        self.save()

    def deactivate(self):
        self.role = "inactive"
        self.save()

This gives you:

Pydantic validation — name and email are required strings, role defaults to "member"
SQLAlchemy persistence — a user table is created with these columns
Active Record methods — save(), get(), delete(), all(), where(), filter(), search()
String primary key — every entity has an id: str field (defined in the base class)

Database Initialization

Before using entities, initialize the database to create tables:

from nitro import Entity

class User(Entity, table=True):
    name: str
    email: str
    role: str = "member"

# Create all tables — safe to call multiple times
User.repository().init_db()

CRUD Operations

save()

Creates or updates a record. Returns True on success.

from nitro import Entity, uniq

class Product(Entity, table=True):
    title: str
    price: float
    stock: int = 0

Product.repository().init_db()

# Create
product = Product(id=uniq(), title="Laptop", price=999.99, stock=10)
product.save()

# Update
product.price = 899.99
product.save()

save() does an upsert — if a record with the same id exists, it updates; otherwise it creates.

get()

Retrieves an entity by ID. Returns None if not found.

from nitro import Entity

class Product(Entity, table=True):
    title: str
    price: float
    stock: int = 0

product = Product.get("prod-123")
if product:
    print(f"{product.title}: ${product.price}")

find() is an alias for get() — use whichever you prefer.

delete()

Removes the entity from the database. Returns True if deleted, False if not found.

product = Product.get("prod-123")
if product:
    product.delete()

The in-memory object still exists after deletion — only the database record is removed.

exists()

Checks existence without loading the entity. More efficient than get() when you only need a boolean.

if Product.exists("prod-123"):
    print("Product exists")

Querying

Nitro provides four query methods. Use the right one for each situation:

Method	Use when...	Returns
`all()`	You need every record	`List[Entity]`
`where()`	You need SQL expressions (`<`, `>`, `LIKE`, etc.)	`List[Entity]`
`filter()`	You need exact field matches with sorting/pagination	`List[Entity]` or `List[dict]`
`search()`	You need full-text search across all string fields	`List[Entity]` or `List[dict]`
`find_by()`	You need the first record matching exact field values	`Entity` or `None`

all()

Returns every entity of this type. Use with caution on large tables.

from nitro import Entity

class Task(Entity, table=True):
    title: str
    status: str = "pending"
    assignee: str = ""

Task.repository().init_db()

all_tasks = Task.all()
for task in all_tasks:
    print(f"{task.title}: {task.status}")

where() — SQL Expressions

The most flexible query method. Use it for comparisons, ranges, and complex conditions.

from nitro import Entity

class Product(Entity, table=True):
    title: str
    price: float
    stock: int = 0

Product.repository().init_db()

# Comparison operators
cheap = Product.where(Product.price < 20.0)

# Multiple conditions (AND)
available_cheap = Product.where(
    Product.price < 20.0,
    Product.stock > 0
)

# Ordering and pagination
top_expensive = Product.where(
    Product.stock > 0,
    order_by=Product.price.desc(),
    limit=10
)

# Pagination
page_2 = Product.where(
    Product.stock > 0,
    order_by=Product.price,
    limit=10,
    offset=10
)

# order_by also accepts a string
by_title = Product.where(Product.stock > 0, order_by="title")

filter() — Exact Field Matches

Use filter() when you know the exact field values. It validates field names and supports partial string matching, IN queries, and date ranges.

from nitro import Entity

class Task(Entity, table=True):
    title: str
    status: str = "pending"
    assignee: str = ""

Task.repository().init_db()

# Exact match
pending = Task.filter(status="pending")

# With sorting and pagination
recent = Task.filter(
    status="pending",
    sorting_field="title",
    sort_direction="desc",
    limit=20,
    offset=0
)

# Partial string matching
johns_tasks = Task.filter(assignee="John", exact_match=False)
# Matches "John Doe", "Johnny", etc.

# IN query — pass a list
active = Task.filter(status=["pending", "in_progress"])

# As dictionaries (useful for JSON responses)
tasks_json = Task.filter(status="pending", as_dict=True)

# Select specific fields
summaries = Task.filter(
    status="pending",
    fields=["id", "title", "assignee"],
    as_dict=True
)

search() — Full-Text Search

Searches across all string fields using case-insensitive partial matching (SQL ILIKE). No setup required.

from nitro import Entity

class Product(Entity, table=True):
    title: str
    description: str
    sku: str
    price: float

Product.repository().init_db()

# Searches title, description, and sku
results = Product.search(search_value="laptop")

# With sorting and pagination
top_matches = Product.search(
    search_value="laptop",
    sorting_field="price",
    sort_direction="asc",
    limit=10
)

find_by() — First Match

Returns the first entity matching exact field values, or None.

from nitro import Entity

class User(Entity, table=True):
    name: str
    email: str
    role: str = "member"

User.repository().init_db()

admin = User.find_by(role="admin")
if admin:
    print(f"Admin: {admin.name}")

# Multiple fields
user = User.find_by(name="Alice", role="member")

Signals Integration

Every entity has a signals property that creates a reactive Signals object from its current field values. This integrates with Datastar for real-time UI updates.

from nitro import Entity, Div, Signals

class Counter(Entity, table=True):
    count: int = 0

counter = Counter.get("c1") or Counter(id="c1", count=0)

# counter.signals returns Signals(id="c1", count=0)
component = Div(
    f"Count: {counter.count}",
    signals=counter.signals
)

See the Signals documentation for details on reactive UI patterns.

Repository Configuration

Default: SQLite

By default, all entities use a singleton SQLModelRepository backed by SQLite:

# Default database file
sqlite:///nitro.db

Changing the Database

Set the NITRO_DB_URL environment variable or add it to .env:

# PostgreSQL
NITRO_DB_URL=postgresql://user:password@localhost/myapp

# MySQL
NITRO_DB_URL=mysql+pymysql://user:password@localhost/myapp

Connection Pooling (Production)

from nitro.domain.repository.sql import SQLModelRepository

repo = SQLModelRepository(
    url="postgresql://user:password@localhost/myapp",
    pool_size=10,
    max_overflow=20,
    pool_timeout=30.0
)
repo.init_db()

Pool settings are ignored for SQLite (it uses SingletonThreadPool).

In-Memory Repository

For testing or ephemeral data, use MemoryRepository directly:

from nitro.domain.repository.memory import MemoryRepository

repo = MemoryRepository()

# Save with optional TTL (auto-expires after 300 seconds)
repo.save(session_data, ttl=300)

# Retrieve
data = repo.find(SessionData, "sess-123")

# Cleanup expired entries
expired_count = repo.cleanup_expired_sync()

MemoryRepository is useful for unit tests (no database required) and temporary data like verification codes or session caches. It uses the singleton pattern — all callers share the same in-memory store.