Peewee is a small, expressive ORM for SQLite, PostgreSQL, and MySQL. pip install peewee. Model: from peewee import *; db = SqliteDatabase("app.db"). class User(Model): name = CharField(); class Meta: database = db. Create: db.create_tables([User]). Insert: User.create(name="Alice"). User(name="Bob").save(). Select: User.select(). .where(User.name == "Alice"). .order_by(User.name). .limit(10). Get: User.get(User.id == 1). User.get_or_none(User.name == "Alice"). Update: User.update(name="Bob").where(User.id == 1).execute(). u = User.get(id=1); u.name = "Bob"; u.save(). Delete: User.delete().where(...).execute(). u.delete_instance(). Foreign key: author = ForeignKeyField(User, backref="posts"). Post.select().join(User). Bulk: User.bulk_create([User(name="A"), User(name="B")]). Transaction: with db.atomic():. @db.atomic(). Prefetch: User.select().prefetch(Post). Fields: CharField, IntegerField, FloatField, BooleanField, DateTimeField, TextField, BlobField, AutoField, JSONField(pg). null=True default= unique=True index=True. Pool: from playhouse.pool import PooledSqliteDatabase. Migrate: from playhouse.migrate import *; migrator = SqliteMigrator(db). Claude Code generates Peewee models, query builders, and database migration pipelines.
CLAUDE.md for Peewee
## Peewee Stack
- Version: peewee >= 3.17 | pip install peewee
- DB: SqliteDatabase("app.db") | PostgresqlDatabase(db, user, pw) | MySQLDatabase
- Model: class User(Model): name = CharField() | class Meta: database = db
- CRUD: User.create() | .get() | .get_or_none() | .save() | .delete_instance()
- Query: User.select().where(User.active==True).order_by(-User.created).limit(10)
- Join: Post.select().join(User).where(User.name=="Alice")
- Atomic: with db.atomic(): ... | @db.atomic() decoratorPeewee ORM Pipeline
# app/models.py — Peewee models, CRUD helpers, and query patterns
from __future__ import annotations
import datetime
from typing import Any
from peewee import (
AutoField,
BooleanField,
CharField,
DateTimeField,
FloatField,
ForeignKeyField,
IntegerField,
Model,
SqliteDatabase,
TextField,
)
# ─────────────────────────────────────────────────────────────────────────────
# Database setup
# ─────────────────────────────────────────────────────────────────────────────
db = SqliteDatabase(
"app.db",
pragmas={
"journal_mode": "wal",
"foreign_keys": 1,
"cache_size": -64 * 1024, # 64 MB
},
)
class BaseModel(Model):
"""Base model: all subclasses share the same database."""
class Meta:
database = db
# ─────────────────────────────────────────────────────────────────────────────
# Models
# ─────────────────────────────────────────────────────────────────────────────
class User(BaseModel):
id = AutoField()
name = CharField(max_length=200)
email = CharField(max_length=200, unique=True)
active = BooleanField(default=True)
score = FloatField(default=0.0)
created_at = DateTimeField(default=datetime.datetime.utcnow)
class Meta:
table_name = "users"
indexes = ((("email",), True),) # unique index
class Tag(BaseModel):
id = AutoField()
name = CharField(max_length=100, unique=True)
class Post(BaseModel):
id = AutoField()
author = ForeignKeyField(User, backref="posts", on_delete="CASCADE")
title = CharField(max_length=500)
body = TextField()
published = BooleanField(default=False)
view_count = IntegerField(default=0)
created_at = DateTimeField(default=datetime.datetime.utcnow)
class Meta:
table_name = "posts"
indexes = ((("author", "title"), False),)
class PostTag(BaseModel):
"""Many-to-many join table."""
post = ForeignKeyField(Post, backref="post_tags", on_delete="CASCADE")
tag = ForeignKeyField(Tag, backref="post_tags", on_delete="CASCADE")
class Meta:
table_name = "post_tags"
primary_key = False
indexes = ((("post", "tag"), True),)
ALL_MODELS = [User, Tag, Post, PostTag]
# ─────────────────────────────────────────────────────────────────────────────
# 1. Database initialisation
# ─────────────────────────────────────────────────────────────────────────────
def init_db(create_tables: bool = True) -> None:
"""Connect and optionally create all tables."""
db.connect(reuse_if_open=True)
if create_tables:
db.create_tables(ALL_MODELS, safe=True)
def close_db() -> None:
if not db.is_closed():
db.close()
# ─────────────────────────────────────────────────────────────────────────────
# 2. CRUD helpers
# ─────────────────────────────────────────────────────────────────────────────
def create_user(name: str, email: str, score: float = 0.0) -> User:
"""Create and return a new User."""
return User.create(name=name, email=email, score=score)
def get_user(user_id: int) -> User | None:
"""Fetch a user by ID; None if not found."""
return User.get_or_none(User.id == user_id)
def get_user_by_email(email: str) -> User | None:
return User.get_or_none(User.email == email)
def update_user(user: User, **fields: Any) -> User:
"""Update user fields and save."""
for key, value in fields.items():
setattr(user, key, value)
user.save()
return user
def deactivate_user(user: User) -> None:
User.update(active=False).where(User.id == user.id).execute()
def create_post(author: User, title: str, body: str, published: bool = False) -> Post:
return Post.create(author=author, title=title, body=body, published=published)
# ─────────────────────────────────────────────────────────────────────────────
# 3. Query patterns
# ─────────────────────────────────────────────────────────────────────────────
def active_users(limit: int = 50) -> list[User]:
return list(
User.select()
.where(User.active == True)
.order_by(User.name)
.limit(limit)
)
def top_users_by_score(n: int = 10) -> list[User]:
return list(
User.select()
.where(User.active == True)
.order_by(-User.score)
.limit(n)
)
def published_posts(page: int = 1, per_page: int = 20) -> list[Post]:
return list(
Post.select(Post, User)
.join(User)
.where(Post.published == True)
.order_by(-Post.created_at)
.paginate(page, per_page)
)
def posts_by_user(user: User) -> list[Post]:
"""Return all posts by a user, newest first."""
return list(user.posts.order_by(-Post.created_at))
def search_posts(query: str, limit: int = 20) -> list[Post]:
"""Full text search on title and body."""
return list(
Post.select()
.where((Post.title.contains(query)) | (Post.body.contains(query)))
.where(Post.published == True)
.limit(limit)
)
def user_post_count() -> list[dict[str, Any]]:
"""Return each user's post count as a list of dicts."""
from peewee import fn
query = (
User.select(User, fn.COUNT(Post.id).alias("post_count"))
.join(Post, join_type="LEFT OUTER")
.group_by(User.id)
.order_by(-fn.COUNT(Post.id))
)
return [{"user": u.name, "count": u.post_count} for u in query]
# ─────────────────────────────────────────────────────────────────────────────
# 4. Bulk operations
# ─────────────────────────────────────────────────────────────────────────────
def bulk_create_users(users_data: list[dict[str, Any]], chunk_size: int = 100) -> int:
"""Bulk insert users. Returns number created."""
users = [User(**d) for d in users_data]
with db.atomic():
User.bulk_create(users, batch_size=chunk_size)
return len(users)
def bulk_update_scores(updates: list[dict[str, Any]]) -> int:
"""
Bulk-update user scores.
updates: [{"id": 1, "score": 100.0}, ...]
"""
with db.atomic():
User.bulk_update(
[User(id=u["id"], score=u["score"]) for u in updates],
fields=[User.score],
batch_size=100,
)
return len(updates)
# ─────────────────────────────────────────────────────────────────────────────
# 5. Many-to-many tag helpers
# ─────────────────────────────────────────────────────────────────────────────
def get_or_create_tag(name: str) -> Tag:
tag, _ = Tag.get_or_create(name=name.lower().strip())
return tag
def tag_post(post: Post, tag_names: list[str]) -> None:
"""Attach tags to a post (idempotent)."""
with db.atomic():
for name in tag_names:
tag = get_or_create_tag(name)
PostTag.get_or_create(post=post, tag=tag)
def posts_by_tag(tag_name: str, limit: int = 20) -> list[Post]:
tag = Tag.get_or_none(Tag.name == tag_name.lower())
if tag is None:
return []
return list(
Post.select()
.join(PostTag)
.join(Tag)
.where(Tag.name == tag_name.lower(), Post.published == True)
.limit(limit)
)
# ─────────────────────────────────────────────────────────────────────────────
# 6. Flask / FastAPI integration
# ─────────────────────────────────────────────────────────────────────────────
def flask_db_hooks(app) -> None:
"""
Register Peewee connect/close hooks for a Flask app.
Opens a connection before each request and closes after.
"""
@app.before_request
def _open():
db.connect(reuse_if_open=True)
@app.teardown_request
def _close(exc):
if not db.is_closed():
db.close()
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
import os
DB_FILE = "/tmp/peewee_demo.db"
if os.path.exists(DB_FILE):
os.remove(DB_FILE)
db.init(DB_FILE)
init_db()
print("=== Create users ===")
with db.atomic():
alice = create_user("Alice", "[email protected]", score=98.5)
bob = create_user("Bob", "[email protected]", score=87.0)
carol = create_user("Carol", "[email protected]", score=92.0)
print(f" Created: {alice.name}, {bob.name}, {carol.name}")
print("\n=== Queries ===")
top = top_users_by_score(n=3)
for u in top:
print(f" {u.name:10} score={u.score}")
print("\n=== Posts ===")
with db.atomic():
p1 = create_post(alice, "Hello Python", "Python is great!", published=True)
p2 = create_post(alice, "Async Tips", "Use asyncio!", published=True)
p3 = create_post(bob, "Draft post", "WIP", published=False)
tag_post(p1, ["python", "tutorial"])
tag_post(p2, ["python", "async"])
published = published_posts()
print(f" Published: {[p.title for p in published]}")
print("\n=== Join query (post count per user) ===")
for row in user_post_count():
print(f" {row['user']:10}: {row['count']} posts")
print("\n=== Tag search ===")
python_posts = posts_by_tag("python")
print(f" Posts tagged 'python': {[p.title for p in python_posts]}")
print("\n=== Search ===")
results = search_posts("Python")
print(f" Search 'Python': {[p.title for p in results]}")
print("\n=== Bulk create ===")
bulk_data = [{"name": f"User{i}", "email": f"u{i}@x.com"} for i in range(10)]
n = bulk_create_users(bulk_data)
print(f" Bulk created {n} users | Total: {User.select().count()}")
close_db()
os.remove(DB_FILE)For the SQLAlchemy alternative — SQLAlchemy is more powerful (supports async, complex relationships, full migration with Alembic, and all major databases including async engines), but has a steeper learning curve; Peewee’s concise model declaration (name = CharField()) and chainable query API (.where().order_by().limit()) make it faster to get a working data layer for small-to-medium projects, especially with SQLite; use Peewee when you want an ORM without the complexity and boilerplate of SQLAlchemy core + ORM. For the Django ORM alternative — Django’s ORM is excellent but only works within a Django project; Peewee is framework-agnostic and works with Flask, FastAPI, CLI tools, and standalone scripts — one class User(Model) is all you need to start querying. The Claude Skills 360 bundle includes Peewee skill sets covering SqliteDatabase/PostgresqlDatabase/MySQLDatabase setup, Model field types (CharField, ForeignKeyField, DateTimeField, etc.), CRUD (create/get_or_none/save/delete_instance), query chaining (where/order_by/limit/paginate), JOINs and prefetch, db.atomic() transactions, bulk_create() and bulk_update(), fn.COUNT/fn.SUM aggregate queries, many-to-many patterns with junction models, flask_db_hooks() connection management, and PostgreSQL JSONField. Start with the free tier to try Peewee ORM code generation.