Models & Database¶
P8s uses SQLModel (SQLAlchemy + Pydantic) for database models.
Base Model¶
All models should extend p8s.Model:
from p8s import Model
from sqlmodel import Field
class Product(Model, table=True):
name: str = Field(index=True)
description: str = ""
price: float = 0.0
is_active: bool = True
Inherited Features¶
The base Model class provides:
| Field | Type | Description |
|---|---|---|
id |
UUID |
Auto-generated primary key |
created_at |
datetime |
Auto-set on creation |
updated_at |
datetime |
Auto-updated on save |
deleted_at |
datetime |
Soft delete timestamp |
Database Configuration¶
Configure in .env:
# SQLite (default)
DATABASE_URL=sqlite+aiosqlite:///p8s.db
# PostgreSQL
DATABASE_URL=postgresql+asyncpg://user:password@localhost/dbname
# PostgreSQL with pgvector (for AI features)
DATABASE_URL=postgresql+asyncpg://user:password@localhost/dbname?server_settings=vector.enable=on
Field Reference¶
P8s provides Django-style field helpers in p8s.db.fields that wrap SQLModel fields.
Basic Types¶
CharField(max_length=255): String field.TextField(): Unlimited text field.BooleanField(default=False): Boolean field.IntegerField(default=0): Integer field.FloatField(default=0.0): Float field.DecimalField(max_digits=10, decimal_places=2): Decimal field.
Date & Time¶
DateField(auto_now=False, auto_now_add=False): Date field.DateTimeField(auto_now=False, auto_now_add=False): DateTime field.
Structured & Validated¶
JSONField(default=None): JSON storage (requires PostgreSQL/SQLite JSON support).EmailField(): String field validated as email.URLField(): String field for URLs.ColorField(format="hex"): Stores color values, renders color picker in Admin.
Relationships¶
One-to-Many¶
from uuid import UUID
class Author(Model, table=True):
name: str
class Book(Model, table=True):
title: str
author_id: UUID = ForeignKey("author.id")
One-to-One¶
from p8s.db.fields import OneToOneField
class UserProfile(Model, table=True):
bio: str
user_id: UUID = OneToOneField("user.id")
Many-to-Many¶
from sqlmodel import Relationship
class BookTag(Model, table=True):
book_id: UUID = Field(foreign_key="book.id", primary_key=True)
tag_id: UUID = Field(foreign_key="tag.id", primary_key=True)
class Tag(Model, table=True):
name: str
books: list["Book"] = Relationship(back_populates="tags", link_model=BookTag)
class Book(Model, table=True):
title: str
tags: list[Tag] = Relationship(back_populates="books", link_model=BookTag)
CRUD Operations¶
Create¶
from p8s.db.session import get_session
async with get_session() as session:
product = Product(name="Widget", price=9.99)
session.add(product)
await session.commit()
Read¶
from sqlmodel import select
async with get_session() as session:
# Get by ID
result = await session.get(Product, product_id)
# Query all
result = await session.execute(select(Product))
products = result.scalars().all()
# Filter
result = await session.execute(
select(Product).where(Product.is_active == True)
)
Update¶
async with get_session() as session:
product = await session.get(Product, product_id)
product.price = 19.99
session.add(product)
await session.commit()
Delete¶
# Soft delete (recommended)
product.soft_delete()
session.add(product)
await session.commit()
# Hard delete
await session.delete(product)
await session.commit()
Soft Delete¶
All models support soft delete by default:
# Soft delete
product.soft_delete()
# Restore
product.restore()
# Query includes soft-deleted items
select(Product).where(Product.deleted_at.is_(None))
Migrations¶
Create tables¶
With Alembic (advanced)¶
Best Practices¶
- Always use UUIDs (handled by base Model)
- Use soft deletes for audit trails
- Index frequently queried fields
- Use
table=Truefor database-backed models
Standalone Scripts & Seeding¶
For scripts that run outside the API context (e.g., seeding data, cron jobs, data analysis), use the setup_context utility. This ensures the database and settings are correctly initialized.
import asyncio
from p8s.core.context import setup_context
from p8s.db.session import SessionManager
from backend.models import Product
async def main():
async with setup_context():
async with SessionManager() as session:
# Full ORM access
product = Product(name="Script created", price=10.0)
session.add(product)
await session.commit()
if __name__ == "__main__":
asyncio.run(main())
Use p8s seed to execute these scripts easily.