What is Peewee and Why Migrate?
Peewee is a lightweight, expressive ORM for Python that bridges the gap between raw SQL and heavyweight frameworks. It provides a simple yet powerful API for database operations without the overhead of a full-stack framework. Migration from legacy frameworks such as Django ORM (used standalone), SQLAlchemy, or even raw SQL queries becomes necessary when you need a slimmer dependency footprint, faster startup times, or more direct control over your database interactions while still enjoying the benefits of an ORM.
Legacy frameworks often come with baggage—configuration layers, global state, complex session management, or implicit magic that makes debugging difficult. Peewee strips away the ceremony and gives you a clean, intuitive interface that feels like writing Python, not boilerplate.
The Legacy Landscape
Common scenarios that warrant migration include:
- Standalone Django ORM — Using Django's ORM outside of Django requires significant setup (settings, apps, and a complex initialization dance). Peewee eliminates all of that.
- SQLAlchemy Core/ORM — Powerful but notoriously steep learning curve. Peewee offers a gentler path with comparable functionality for most use cases.
- Raw SQL with psycopg2 or MySQLdb — Loses type safety, encourages string concatenation bugs, and lacks model-based abstractions. Peewee brings structure without sacrificing transparency.
- Flask-SQLAlchemy — Tightly coupled to Flask's app context; migrating to a non-Flask environment or a serverless function becomes painful. Peewee works anywhere with zero framework dependencies.
Why Peewee Matters
Peewee matters because it respects your time and your codebase. Key advantages include:
- Zero configuration startup — Define a database connection and start querying immediately
- Pythonic model definitions — Models feel like dataclasses, not magical metaclass constructs
- Explicit query building — Every query is inspectable, debuggable, and predictable
- Lightweight footprint — The entire library is a single module; deployment size matters for serverless and embedded contexts
- Rich feature set — Migrations, connection pooling, query logging, and advanced joins all available without third-party plugins
- Asynchronous support — Peewee offers async-compatible patterns for modern Python applications
Planning Your Migration
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →Before writing a single line of Peewee code, assess your current database schema and query patterns. Peewee supports SQLite, PostgreSQL, MySQL, and CockroachDB. If your legacy system uses one of these, migration is straightforward. The process typically follows these phases:
- Audit existing models and map them to Peewee Model classes
- Catalog all CRUD operations and their equivalent Peewee expressions
- Identify custom SQL that may need raw query fallback (Peewee handles this gracefully)
- Test migration in isolation with a copy of your database
- Incrementally replace legacy calls with Peewee equivalents
Step-by-Step Migration Guide
1. Installing Peewee
pip install peewee
For PostgreSQL support, also install the driver:
pip install psycopg2-binary # or psycopg2
For MySQL:
pip install pymysql
No additional dependencies are required for SQLite—it's built into Python.
2. Defining Your First Model
Here's how a typical Django ORM model maps to Peewee. Consider this Django model:
# Legacy Django ORM (standalone usage)
from django.db import models
class Author(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField(unique=True)
joined_date = models.DateTimeField(auto_now_add=True)
class Book(models.Model):
title = models.CharField(max_length=200)
author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name='books')
published_date = models.DateField(null=True, blank=True)
isbn = models.CharField(max_length=13, unique=True)
price = models.DecimalField(max_digits=6, decimal_places=2)
The equivalent Peewee model definition:
# Peewee equivalent
from peewee import *
from datetime import datetime
database = SqliteDatabase('library.db') # or PostgresqlDatabase, MySQLDatabase
class BaseModel(Model):
class Meta:
database = database
class Author(BaseModel):
name = CharField(max_length=100)
email = CharField(max_length=255, unique=True)
joined_date = DateTimeField(default=datetime.datetime.now)
class Book(BaseModel):
title = CharField(max_length=200)
author = ForeignKeyField(Author, backref='books', on_delete='CASCADE')
published_date = DateField(null=True)
isbn = CharField(max_length=13, unique=True)
price = DecimalField(max_digits=6, decimal_places=2)
Notice the clarity: Peewee uses plain Python classes. The Meta inner class holds database configuration. The backref parameter on ForeignKeyField replaces Django's related_name, and on_delete behavior is specified as a string.
3. Connecting and Creating Tables
Peewee connects lazily—no global connection pool to configure. You create tables explicitly:
# Connect and create tables
database.connect()
database.create_tables([Author, Book], safe=True)
# safe=True uses IF NOT EXISTS, preventing errors on re-run
This contrasts sharply with Django's manage.py migrate or SQLAlchemy's Base.metadata.create_all()—both of which require framework-level initialization. With Peewee, you're in full control from the first line.
4. CRUD Operations: The Full Translation Map
Let's walk through every common operation, showing the legacy Django ORM code and its Peewee counterpart side by side.
Create / Insert
# Django ORM
author = Author.objects.create(name='Jane Austen', email='jane@example.com')
book = Book.objects.create(
title='Pride and Prejudice',
author=author,
isbn='9780141439518',
price=9.99
)
# Peewee
author = Author.create(name='Jane Austen', email='jane@example.com')
book = Book.create(
title='Pride and Prejudice',
author=author,
isbn='9780141439518',
price=9.99
)
Peewee's .create() is a classmethod that returns the new instance. You can also instantiate and save:
author = Author(name='Jane Austen', email='jane@example.com')
author.save() # persists to database
print(author.id) # populated after save
Read / Select
# Django ORM — single record
author = Author.objects.get(id=1)
# Peewee — single record
author = Author.get_by_id(1)
# or using get() with expressions
author = Author.get(Author.id == 1)
# Django ORM — first match or None
author = Author.objects.filter(email='jane@example.com').first()
# Peewee — first match or None
author = Author.get_or_none(Author.email == 'jane@example.com')
# Django ORM — all records
authors = Author.objects.all()
# Peewee — all records
authors = Author.select()
# Django ORM — filtered queryset
books = Book.objects.filter(price__gt=10, published_date__isnull=False)
# Peewee — filtered query
books = Book.select().where(
Book.price > 10,
Book.published_date.is_null(False)
)
Peewee uses Python comparison operators (>, <, ==, !=) directly on field objects. There's no magic __gt or __isnull suffix syntax. This makes queries readable, IDE-friendly, and debuggable.
Update
# Django ORM
Book.objects.filter(author=author).update(price=12.99)
Author.objects.filter(id=1).update(name='Jane A.', email='jane.a@example.com')
# Peewee — bulk update
Book.update(price=12.99).where(Book.author == author).execute()
Author.update(name='Jane A.', email='jane.a@example.com').where(Author.id == 1).execute()
# Peewee — single instance update
author = Author.get_by_id(1)
author.name = 'Jane A.'
author.email = 'jane.a@example.com'
author.save() # only saves dirty fields
Note the explicit .execute() call on bulk updates. Peewee separates query construction from execution, giving you the chance to inspect the SQL before running it:
query = Book.update(price=12.99).where(Book.author == author)
print(query.sql()) # view the generated SQL
query.execute() # then execute
Delete
# Django ORM
Book.objects.filter(published_date=None).delete()
# Peewee — bulk delete
Book.delete().where(Book.published_date.is_null()).execute()
# Peewee — single instance delete
book = Book.get_by_id(42)
book.delete_instance() # also handles related objects if on_delete is set
The delete_instance() method respects on_delete rules you defined on foreign keys, cascading deletions when appropriate.
5. Relationships and Joins
Peewee's handling of relationships is one of its strongest features. Here's how to work with related records:
# Django ORM — accessing related objects
author = Author.objects.get(id=1)
books = author.books.all() # uses related_name 'books'
# Peewee — using backref
author = Author.get_by_id(1)
books = author.books # backref='books' creates a magic property
# Explicit join query in Django ORM
books_with_authors = Book.objects.select_related('author').all()
# Explicit join in Peewee
books_with_authors = Book.select().join(Author).execute()
# Filtering across joins — Django ORM
books = Book.objects.filter(author__name__startswith='Jane')
# Filtering across joins — Peewee
books = Book.select().join(Author).where(
Author.name.startswith('Jane')
)
For more complex joins, Peewee supports multiple join types:
# INNER JOIN with filter
query = (Book
.select()
.join(Author, on=Book.author == Author.id)
.where(Author.email.contains('example.com'))
.order_by(Book.published_date.desc())
.limit(10))
# LEFT JOIN example
query = (Author
.select()
.join(Book, JOIN_TYPE.LEFT_OUTER, on=Author.id == Book.author_id)
.where(Book.id.is_null()) # authors with no books
)
6. Handling Raw SQL and Legacy Queries
Not every legacy query translates cleanly to ORM methods. Peewee embraces this reality with first-class raw SQL support:
# Execute raw SQL and get model instances
authors = Author.raw('SELECT * FROM author WHERE name LIKE %s', 'Jane%')
# Raw SQL returning arbitrary results
cursor = database.execute_sql(
'SELECT author_id, COUNT(*) as book_count FROM book GROUP BY author_id'
)
results = cursor.fetchall()
# Using Peewee's SQL builder for complex cases
query = Author.select(
Author.name,
fn.COUNT(Book.id).alias('book_count')
).join(Book, JOIN_TYPE.LEFT_OUTER).group_by(Author.name)
This means you can migrate incrementally—move simple queries to Peewee first, leave complex raw SQL in place, and refactor them over time.
7. Migrating Schema Migrations
If your legacy system uses Django migrations or Alembic, you'll want to adopt Peewee's migration module:
from peewee import *
from playhouse.migrate import migrate, MySQLMigrator, PostgresqlMigrator, SqliteMigrator
# For SQLite
migrator = SqliteMigrator(database)
# Define operations
operations = [
migrator.add_column('book', 'rating', FloatField(default=0.0)),
migrator.add_index('book', ['author_id', 'published_date']),
migrator.rename_column('book', 'isbn', 'isbn13'),
migrator.drop_column('book', 'old_field'),
]
# Run migration
migrate(*operations)
Peewee migrations are Python scripts, not auto-generated files. This gives you precise control and makes CI/CD integration straightforward—just run the script against your target database.
8. Connection Management and Pooling
Legacy frameworks often manage connections implicitly, which can lead to leaks in long-running processes. Peewee offers explicit connection handling and optional pooling:
# Manual connection management
database.connect()
try:
# perform operations
authors = Author.select()
finally:
database.close()
# Connection pooling (PostgreSQL example)
from playhouse.pool import PooledPostgresqlDatabase
database = PooledPostgresqlDatabase(
'mydb',
host='localhost',
port=5432,
user='postgres',
password='secret',
max_connections=20,
stale_timeout=300,
)
# Pooled database handles acquire/release automatically
# when used with Peewee's context manager
with database:
authors = Author.select()
# connection automatically returned to pool on exit
For web applications or serverless functions, the context manager pattern eliminates connection leaks entirely.
Real-World Migration Example: A Complete Module
Here's a complete module that replaces a legacy Django ORM data access layer with Peewee. This is a practical, production-ready pattern:
# data_access.py — Peewee-based data access layer
from peewee import *
from playhouse.pool import PooledPostgresqlDatabase
from datetime import datetime
import os
# Configuration from environment (12-factor app style)
db_config = {
'host': os.environ.get('DB_HOST', 'localhost'),
'port': int(os.environ.get('DB_PORT', 5432)),
'user': os.environ.get('DB_USER', 'app_user'),
'password': os.environ.get('DB_PASSWORD', ''),
'database': os.environ.get('DB_NAME', 'library'),
}
database = PooledPostgresqlDatabase(
db_config['database'],
host=db_config['host'],
port=db_config['port'],
user=db_config['user'],
password=db_config['password'],
max_connections=10,
stale_timeout=600,
)
class BaseModel(Model):
class Meta:
database = database
class Author(BaseModel):
name = CharField(max_length=100)
email = CharField(max_length=255, unique=True)
joined_date = DateTimeField(default=datetime.utcnow)
bio = TextField(null=True)
class Book(BaseModel):
title = CharField(max_length=200)
author = ForeignKeyField(Author, backref='books', on_delete='CASCADE')
published_date = DateField(null=True)
isbn = CharField(max_length=13, unique=True)
price = DecimalField(max_digits=6, decimal_places=2)
created_at = DateTimeField(default=datetime.utcnow)
class Review(BaseModel):
book = ForeignKeyField(Book, backref='reviews', on_delete='CASCADE')
reviewer_name = CharField(max_length=100)
rating = IntegerField() # 1-5
comment = TextField(null=True)
created_at = DateTimeField(default=datetime.utcnow)
def initialize_database():
"""Create tables if they don't exist — safe for startup."""
with database:
database.create_tables([Author, Book, Review], safe=True)
def get_author_by_id(author_id: int) -> Author | None:
"""Fetch a single author by primary key."""
return Author.get_or_none(Author.id == author_id)
def get_authors_with_book_count():
"""Return authors with their book count using aggregation."""
query = (
Author
.select(
Author,
fn.COUNT(Book.id).alias('book_count')
)
.join(Book, JOIN_TYPE.LEFT_OUTER, on=Author.id == Book.author_id)
.group_by(Author.id)
.order_by(fn.COUNT(Book.id).desc())
)
return query.execute()
def search_books_by_title(keyword: str, limit: int = 20):
"""Search books with a case-insensitive title match."""
return (
Book
.select()
.where(Book.title.contains(keyword))
.order_by(Book.published_date.desc())
.limit(limit)
.execute()
)
def create_book_with_review(
title: str,
author_id: int,
isbn: str,
price: float,
reviewer_name: str,
rating: int,
comment: str | None = None
) -> Book:
"""Transactional creation of a book and its first review."""
with database.atomic() as txn:
book = Book.create(
title=title,
author=author_id,
isbn=isbn,
price=price,
published_date=datetime.utcnow().date()
)
Review.create(
book=book,
reviewer_name=reviewer_name,
rating=rating,
comment=comment
)
return book
def get_top_rated_books(min_rating: int = 4, limit: int = 10):
"""Find highly-rated books using a subquery."""
# Subquery: average rating per book
avg_rating = (
Review
.select(
Review.book_id,
fn.AVG(Review.rating).alias('avg_rating')
)
.group_by(Review.book_id)
.having(fn.AVG(Review.rating) >= min_rating)
)
# Use the subquery as a derived table
return (
Book
.select()
.join(avg_rating, on=Book.id == avg_rating.c.book_id)
.order_by(avg_rating.c.avg_rating.desc())
.limit(limit)
.execute()
)
This module demonstrates several important patterns: environment-based configuration, connection pooling, safe table creation, transactional operations with database.atomic(), aggregation queries, and subquery usage. It's self-contained and importable anywhere—no Django settings module, no Flask app context, no global registry.
Best Practices for Peewee Migration
Use Explicit Transactions
Never rely on autocommit alone for multi-step operations. Peewee's database.atomic() context manager ensures rollback on exception:
# Good: atomic transaction
with database.atomic():
author = Author.create(name='New Author', email='new@example.com')
Book.create(title='Debut Novel', author=author, isbn='1234567890', price=14.99)
# Without atomic, a failure in Book.create leaves an orphaned Author
Leverage Peewee's Field Types
Peewee offers specialized fields that map directly to database types. Use them to let the database enforce constraints:
# Use DecimalField for money, not FloatField
price = DecimalField(max_digits=8, decimal_places=2)
# Use IntegerField with choices for enums
status = IntegerField(choices=[(1, 'draft'), (2, 'published'), (3, 'archived')])
# Use BooleanField for true/false flags (maps to INTEGER 0/1 or BOOLEAN)
is_active = BooleanField(default=True)
# Use TimestampField for UTC timestamps with timezone awareness
created_at = DateTimeField(default=datetime.datetime.now)
Keep Database Configuration Separate
Define your database connection once and import it. Never scatter connection strings across modules:
# config/database.py
from playhouse.pool import PooledPostgresqlDatabase
import os
def get_database():
return PooledPostgresqlDatabase(
os.environ['DB_NAME'],
host=os.environ.get('DB_HOST', 'localhost'),
user=os.environ.get('DB_USER', 'app'),
password=os.environ.get('DB_PASSWORD', ''),
max_connections=10,
)
# models/base.py
from config.database import get_database
database = get_database()
class BaseModel(Model):
class Meta:
database = database
Use Connection Context Managers
For web requests or serverless invocations, wrap database access in a context manager:
# Flask view example
@app.route('/authors')
def list_authors():
with database:
authors = Author.select().order_by(Author.name)
return jsonify([{'id': a.id, 'name': a.name} for a in authors])
# FastAPI dependency example
async def get_db():
with database:
yield database
@app.get('/authors')
async def list_authors(db=Depends(get_db)):
authors = Author.select().order_by(Author.name)
return [{'id': a.id, 'name': a.name} for a in authors]
Log Queries During Development
Peewee's query logging reveals exactly what SQL is generated:
import logging
# Enable Peewee query logging
logger = logging.getLogger('peewee')
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())
# All queries will now appear in your console output
authors = Author.select().where(Author.name.contains('Jane'))
# Logs: SELECT "t1"."id", "t1"."name", ... FROM "author" AS "t1"
# WHERE ("t1"."name" LIKE '%Jane%')
Use this during migration to verify that your Peewee queries generate equivalent SQL to your legacy ORM.
Test Migrations Incrementally
Don't rewrite your entire data layer at once. Start with read operations, then writes, then complex joins:
# Phase 1: Replace reads with Peewee, keep legacy writes
def get_active_authors():
return Author.select().where(Author.is_active == True)
# Phase 2: Replace writes once reads are stable
def deactivate_author(author_id):
Author.update(is_active=False).where(Author.id == author_id).execute()
# Phase 3: Move complex queries
def get_author_stats():
return (
Author
.select(
Author.name,
fn.COUNT(Book.id).alias('total_books'),
fn.AVG(Review.rating).alias('avg_rating')
)
.join(Book, JOIN_TYPE.LEFT_OUTER)
.join(Review, JOIN_TYPE.LEFT_OUTER)
.group_by(Author.id)
.execute()
)
Common Pitfalls and How to Avoid Them
Pitfall: Forgetting to Call .execute() on Bulk Operations
Bulk updates and deletes require .execute() at the end. Forgetting it means the query is constructed but never run—no error, no effect:
# WRONG — query built but never executed
Book.update(price=0).where(Book.id == 100)
# CORRECT
Book.update(price=0).where(Book.id == 100).execute()
Pitfall: Using .get() Without Exception Handling
Peewee's .get() raises DoesNotExist if no row matches. Prefer .get_or_none() for optional lookups:
# Risky — will crash if author 999 doesn't exist
author = Author.get(Author.id == 999)
# Safe — returns None
author = Author.get_or_none(Author.id == 999)
if author is None:
# handle missing record
pass
Pitfall: Ignoring Connection State in Long-Running Processes
Database connections can time out. In daemons or background workers, explicitly manage connections:
# For periodic tasks, reconnect before each batch
def process_batch():
database.connect(reuse_if_open=True)
try:
items = QueueItem.select().limit(100)
for item in items:
# process item
item.delete_instance()
finally:
database.close()
Pitfall: Mixing Peewee Models from Different Databases
Every Model binds to a specific database via its Meta class. If you query across databases, you'll get errors. Keep models organized:
# Define separate base classes per database
class AppDatabaseA(Model):
class Meta:
database = db_a
class AppDatabaseB(Model):
class Meta:
database = db_b
class User(AppDatabaseA):
name = CharField()
class AuditLog(AppDatabaseB):
user_id = IntegerField()
action = CharField()
Pitfall: Overlooking Peewee's Built-in Extensions
Peewee's playhouse module contains powerful extensions that replace common legacy patterns:
- playhouse.migrate — Schema migrations (replaces Alembic/Django migrations)
- playhouse.pool — Connection pooling (replaces SQLAlchemy pool)
- playhouse.shortcuts — Model-to-dict conversion, retry logic, case-insensitive utilities
- playhouse.sqlite_ext — Full-text search, JSON extensions for SQLite
- playhouse.postgres_ext — Array fields, JSONB, hstore, and more for PostgreSQL
# Example: model_to_dict replaces Django's model serialization
from playhouse.shortcuts import model_to_dict
author = Author.get_by_id(1)
data = model_to_dict(author, backrefs=True, exclude=[Author.email])
# data includes nested books due to backrefs=True
# Example: retry on database errors
from playhouse.shortcuts import RetryOperationalError
@RetryOperationalError(retries=3, delay=1)
def resilient_save(author):
author.save()
Conclusion
Migrating from a legacy framework to Peewee is not just about replacing one ORM with another—it's about reclaiming control over your data layer. Peewee's explicit design philosophy means every line of code tells you exactly what it does. There are no hidden queries triggered by attribute access, no implicit joins based on string field names, and no framework-level magic that fails silently in production.
The migration path is incremental and forgiving. You can keep raw SQL where needed, adopt Peewee's expressive query builder for everyday operations, and gradually phase out legacy dependencies. The result is a codebase that's lighter, faster to start, easier to debug, and deployable anywhere—from a Raspberry Pi to a serverless function. Whether you're escaping Django's configuration overhead, SQLAlchemy's complexity, or the brittleness of raw string SQL, Peewee offers a pragmatic, Pythonic destination that respects both your database and your development velocity.