Piccolo ORM

Background: FastAPI with typer and hatchling. One is not smart enough to run SQLAlchemy + alembic. Piccollo to the help ;)

FastAPI structure

before adding Piccolo

graph TD
    root[📁 root]
    root --> src[📁 src]
    
	root --> dockerfile[Dockerfile]
    root --> docker-compose.yml[docker-compose.yml]
    root --> pyproject.toml[pyproject.toml]
    root --> .env[.env]
    
    src --> app[📁 app]
    src --> init1["\_\_init__.py"]
    
    app --> main.py[main.py]
    app --> init3["\_\_init__.py"]
    

Install

uv add piccolo[postgresql]

Create files

piccolo_conf.py

# piccolo_conf.py in root dir
from piccolo.conf.apps import AppRegistry
from piccolo.engine.postgres import PostgresEngine
 
import os
from dotenv import load_dotenv
 
load_dotenv()
 
 
DB = PostgresEngine(
	config={
		"database": os.environ.get("POSTGRES_DB"),
		"user": os.environ.get("POSTGRES_USER"),
		"password": os.environ.get("POSTGRES_PASSWORD"),
		"host": os.environ.get("POSTGRES_HOST"),
		"port": os.environ.get("POSTGRES_PORT"),
	}
)
 
APP_REGISTRY = AppRegistry(
		apps=["app.piccolo_app"]
	)

src/app/piccolo_app.py

# src/app/piccolo_app.py
import os
from piccolo.conf.apps import AppConfig
from .db.tables import Users
 
CURRENT_DIRECTORY = os.path.dirname(os.path.abspath(__file__))
 
APP_CONFIG = AppConfig(
	app_name="app",
	migrations_folder_path=os.path.join(CURRENT_DIRECTORY, "db", "piccolo_migrations"),
	table_classes=[Users],
	migration_dependencies=[],
	commands=[]
	)

src/app/db/tables.py

create __init__.py to make src/app/db/ a module.

# src/app/db/tables.py
from piccolo.table import Table
from piccolo.columns import Varchar, UUID, ForeignKey
#from piccolo.utils.pydantic import create_pydantic_model
 
import uuid
from enum import Enum
  
 
class Users(Table):
	class UserType(str, Enum):
		user = 'u'
		manager = 'm'
		admin = 'a'
 
	id = UUID(primary_key=True, default=uuid.uuid4)
	email = Varchar(length=255, unique=True, null=False, required=True)
	password = Varchar(length=255, unique=True, null=False, required=True)
	user_type = Varchar(choices=UserType)
 
 
class Hierarchy(Table):
	id = UUID(primary_key=True, default=uuid.uuid4)
	manager = ForeignKey(references=Users)
	subordinate = ForeignKey(references=Users)

[optional] update pyproject.toml

If you’re using build tools, I know hatchling only, update build settings:

[project]
name = "app-name"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "dotenv>=0.9.9",
    "fastapi>=0.128.0",
    "passlib[argon2]>=1.7.4",
    "piccolo[playground,playground-sqlite,postgres,sqlite]>=1.30.0",
    "typer>=0.21.1",
    "uvicorn>=0.40.0",
]
 
# hatchling part, build
[tool.hatch.build.targets.wheel]
packages = ["src/app"]
 
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
 
# shortcut
[project.scripts]
app = "app.cli:cli"
 

Migrate

test if it’s there’s no obvious issues: uv run piccolo --diagnose

Create migrations: uv run piccolo migrations new app --auto

Run migrations: uv run piccolo migrations forwards app

New file structure

graph TD
    root[📁 root]
    root --> src[📁 src]
    
	root --> dockerfile[Dockerfile]
    root --> docker-compose.yml[docker-compose.yml]
    root --> pyproject.toml[pyproject.toml]
    root --> .env[.env]
    root --> piccolo_conf.py[piccolo_conf.py]
    
    src --> app[📁 app]
    src --> init1["\_\_init__.py"]
    app --> db[📁 db]
    app --> init3["\_\_init__.py"]
    
    db --> init2["\_\_init__.py"]
    db --> tables.py
    db --> piccolo_migrations[📁 piccolo_migrations]
    
    app --> main.py[main.py]
    app --> piccolo_app.py[piccolo_app.py]
    

Use it

#main.py
 
#........
#........
#........
#........
from fastapi import FastAPI
# import db stuff
from piccolo.engine import engine_finder
from app.db.tables import Users
 
 
# force python check root (../../) for piccolo_conf.py
# must be a better way
current_dir = os.path.dirname(os.path.abspath(__file__))
root_dir = os.path.dirname(os.path.dirname(current_dir))
if root_dir not in sys.path:
    sys.path.insert(0, root_dir)
 
# Initialize database as context manager
@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup, open db conn
    db = engine_finder()
    if not db:
        raise RuntimeError("No Database engine found. Check piccolo_conf.py")
    await db.start_connection_pool()
    try:
        yield
    finally:
        await db.close_connection_pool()
 
app = FastAPI(lifespan=lifespan)
 
@app.get("/users")
async def get_users_mock():
    return await Users.select().order_by(Users.id)
 
#........
#........
#........
#........