Fix: Peewee Not Working — Connection Pooling, Field Errors, and Migration Setup

The Error

You define a Peewee model and queries throw OperationalError:

from peewee import Model, IntegerField, CharField

class User(Model):
    name = CharField()

User.select()
# AttributeError: type object 'User' has no attribute '_meta'
# Or: OperationalError: no such table: user

Or the database “locks” under load:

peewee.OperationalError: database is locked

Or migrations don’t generate correctly:

$ peewee_migrate create new_migration --auto
# Migration is empty — no model changes detected

Or queries do N+1 across relationships:

for user in User.select():
    print(user.profile.bio)
    # Separate SELECT for each user — slow

Or you mix sync and async (Peewee is sync-only, surprising in 2024+):

async def get_user(id):
    user = User.get_by_id(id)   # Blocks the event loop!
    return user

Peewee is the lightweight Python ORM — small (single-file or two-file), sync-only, supports SQLite/Postgres/MySQL with consistent API. It’s been around since 2010 and remains popular for small projects, Flask apps, and quick prototypes. The model definition style differs from SQLAlchemy (no declarative_base) and Django (no Manager class) — explicit Model class + inner Meta. This guide covers the common failure modes.

Why This Happens

Peewee uses class-attribute-based field definitions: name = CharField() directly on the model class. The Meta inner class binds the model to a database. Without Meta.database, the model isn’t connected to any database — operations fail with confusing errors.

Peewee is fully synchronous. Calls block the calling thread — fine in scripts or sync web frameworks (Flask), problematic in async frameworks (FastAPI, aiohttp). For async, use a different ORM (Tortoise, SQLModel) or run Peewee in a thread pool.

Fix 1: Database Binding

from peewee import SqliteDatabase, PostgresqlDatabase, MySQLDatabase, Model, CharField, IntegerField

db = SqliteDatabase("app.db")
# Or
db = PostgresqlDatabase("mydb", user="postgres", password="...", host="localhost", port=5432)

class BaseModel(Model):
    class Meta:
        database = db   # All subclasses use this database

class User(BaseModel):
    name = CharField(max_length=100)
    age = IntegerField()

Connect and create tables:

db.connect()
db.create_tables([User])

# Use
user = User.create(name="Alice", age=30)
print(User.select().where(User.age > 18).count())

db.close()

Common Mistake: Forgetting the Meta.database reference. Models without a database can be defined but not queried. Always inherit from a BaseModel that has the database set — keeps it DRY and ensures every model is connected.

Use db.connection_context() for automatic connection management:

with db.connection_context():
    # Connection opens
    user = User.get(User.id == 1)
    user.name = "Bob"
    user.save()
# Connection closes

For Flask integration, use the request lifecycle:

from flask import Flask
from peewee import SqliteDatabase

app = Flask(__name__)
db = SqliteDatabase("app.db")

@app.before_request
def _db_connect():
    db.connect()

@app.teardown_request
def _db_close(exc):
    if not db.is_closed():
        db.close()

Or use the official flask-peewee extension for less boilerplate.

Fix 2: Connection Pooling

For multi-threaded or production apps, Peewee provides pool implementations:

from playhouse.pool import PooledPostgresqlDatabase, PooledMySQLDatabase, PooledSqliteDatabase

db = PooledPostgresqlDatabase(
    "mydb",
    max_connections=20,
    stale_timeout=300,        # Close conns idle for 5 min
    user="postgres",
    password="...",
    host="localhost",
)

Pool methods:

# Manual connection
db.connect()
# ... queries
db.close()   # Returns to pool

# Or context manager
with db.connection_context():
    # ...

Pool sizing: 5-20 for typical apps; 2 × CPU cores for the database is a starting point. PostgreSQL max_connections (default 100) caps total connections — overshooting causes refused connections.

Common Mistake: Using regular PostgresqlDatabase in a multi-threaded server. Each thread tries to share the single connection — Peewee raises OperationalError: connection already open. Always use PooledPostgresqlDatabase in any production deployment with multiple workers/threads.

Fix 3: SQLite “database is locked”

peewee.OperationalError: database is locked

SQLite locks the entire database for writes. If two processes (or threads) try to write simultaneously, one waits — and the default timeout is short.

Increase the timeout:

db = SqliteDatabase("app.db", timeout=30)   # Wait up to 30s for the lock

Enable WAL mode (Write-Ahead Logging) for much better concurrent performance:

db = SqliteDatabase(
    "app.db",
    pragmas={
        "journal_mode": "wal",
        "cache_size": -1024 * 64,   # 64MB cache
        "foreign_keys": 1,
        "synchronous": 0,            # Faster but less durable
    },
)
db.connect()

WAL allows multiple readers concurrent with one writer — the typical web app pattern works much better.

Pro Tip: For any SQLite app in production, enable WAL mode and a generous cache. The performance improvement is dramatic — sites that ran fine on Postgres often run fine on SQLite + WAL once the right pragmas are set. SQLite is wildly underrated for read-heavy workloads with a single writer.

For high write concurrency, SQLite isn’t the right tool — switch to PostgreSQL or MySQL.

Fix 4: Field Definitions

from peewee import (
    Model, CharField, IntegerField, BooleanField,
    DateTimeField, ForeignKeyField, TextField,
    DecimalField, FloatField, BlobField, UUIDField,
)
from playhouse.postgres_ext import JSONField, ArrayField
from datetime import datetime

class User(BaseModel):
    id = IntegerField(primary_key=True)   # Auto-increment by default
    name = CharField(max_length=100, index=True)
    email = CharField(unique=True)
    age = IntegerField(null=True)
    active = BooleanField(default=True)
    created_at = DateTimeField(default=datetime.now)
    profile_data = JSONField(default=dict)   # PostgreSQL-only

class Post(BaseModel):
    title = CharField(max_length=200)
    body = TextField()
    author = ForeignKeyField(User, backref="posts", on_delete="CASCADE")
    published_at = DateTimeField(null=True)

Field options:

ForeignKey specifics:

author = ForeignKeyField(
    User,
    backref="posts",         # Reverse: user.posts queryset
    on_delete="CASCADE",      # On delete user, delete posts
    on_update="CASCADE",
    column_name="author_id",  # Custom FK column name
)

Common Mistake: Defining a field as id = AutoField() then setting another field as primary_key=True. Peewee auto-adds an id IntegerField as primary key unless you specify a custom primary key. Defining both creates conflicting primary keys.

For PostgreSQL-specific patterns that bypass Peewee for performance, see PostgreSQL connection refused.

Fix 5: Querying

# Get
user = User.get(User.id == 1)
user = User.get_or_none(User.email == "[email protected]")
user, created = User.get_or_create(email="[email protected]", defaults={"name": "Alice"})

# Select with filters
users = User.select().where(User.age > 18)
users = User.select().where((User.age > 18) & (User.active == True))
# Note: & for AND, | for OR — Python's `and`/`or` don't work

# In list
users = User.select().where(User.id.in_([1, 2, 3]))

# Like
users = User.select().where(User.name.contains("alice"))
users = User.select().where(User.name.startswith("A"))

# Ordering
users = User.select().order_by(User.created_at.desc())
users = User.select().order_by(User.age, User.name)

# Pagination
users = User.select().limit(10).offset(20)

# Count
count = User.select().count()
count = User.select().where(User.active).count()

# Exists
exists = User.select().where(User.email == "x").exists()

Common Mistake: Using Python’s and/or instead of &/| in filters:

# WRONG — Python's `and` returns the second value, not a Peewee expression
users = User.select().where(User.age > 18 and User.active == True)

# CORRECT — & for AND, | for OR
users = User.select().where((User.age > 18) & (User.active == True))

Parentheses required around each clause because Python’s operator precedence differs from SQL.

Aggregations:

from peewee import fn

# Count
count = User.select(fn.Count(User.id)).scalar()

# Sum
total = Order.select(fn.Sum(Order.amount)).scalar()

# Group by
result = User.select(
    User.country,
    fn.Count(User.id).alias("count")
).group_by(User.country)

for row in result:
    print(row.country, row.count)

Fix 6: Avoiding N+1 Queries

# WRONG — N+1 (one query per user for their profile)
for user in User.select():
    print(user.profile.bio)   # New SELECT per iteration

# CORRECT — single query with JOIN
for user in User.select(User, Profile).join(Profile):
    print(user.profile.bio)

prefetch for reverse relationships:

from peewee import prefetch

users = User.select()
posts = Post.select()

# Fetch users and their posts in 2 queries
users_with_posts = prefetch(users, posts)

for user in users_with_posts:
    for post in user.posts:   # No additional queries
        print(post.title)

.objects() for raw queries without ORM overhead:

# Use objects() for tight loops over large result sets
for user in User.select().objects():
    process(user)   # Lighter than full Model instances

Pro Tip: Use prefetch() for reverse relationships (one-to-many), join() for forward foreign keys. Like SQLAlchemy/Tortoise patterns, the wrong choice silently produces extra queries. Profile with db.set_logger(...) or db.execution_context_depth to see actual query counts.

Fix 7: Migrations with peewee-migrate

pip install peewee-migrate

Configure:

# Create migrations directory
mkdir migrations

# Initialize router
pw_migrate create --auto -d migrations --database "sqlite:///app.db"

Workflow:

# 1. Edit models
# 2. Generate migration
pw_migrate create --auto -d migrations --database "sqlite:///app.db" "add_email_to_user"

# 3. Apply
pw_migrate migrate -d migrations --database "sqlite:///app.db"

# 4. Rollback if needed
pw_migrate rollback -d migrations --database "sqlite:///app.db"

Common Mistake: Forgetting to commit migration files. Each developer regenerates their own migrations from their local model state — and they conflict. Always commit migrations/*.py files to source control immediately after generating.

Alternative: playhouse.migrate (built-in but more manual):

from playhouse.migrate import SqliteMigrator, migrate
from peewee import CharField, SqliteDatabase

db = SqliteDatabase("app.db")
migrator = SqliteMigrator(db)

migrate(
    migrator.add_column("user", "email", CharField(default="")),
    migrator.add_index("user", ("email",), unique=True),
)

For a more powerful migration tool, consider switching to Alembic with SQLAlchemy. For Alembic patterns, see Alembic not working.

Fix 8: Transactions

from peewee import IntegrityError

# Decorator
@db.atomic()
def create_user_and_profile(user_data, profile_data):
    user = User.create(**user_data)
    Profile.create(user=user, **profile_data)
    return user

# Context manager
with db.atomic() as txn:
    User.create(name="Alice")
    User.create(name="Bob")
    # Both committed at end of block, or both rolled back on exception

# Manual savepoints
with db.atomic() as txn:
    User.create(...)
    try:
        with db.atomic() as nested:   # SAVEPOINT
            User.create(...)
            raise ValueError
    except ValueError:
        pass   # Inner rolled back, outer continues
    User.create(...)

Explicit rollback:

with db.atomic() as txn:
    User.create(name="Alice")
    if some_condition:
        txn.rollback()   # Manual rollback

Multi-statement INSERT for bulk:

data = [
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": 25},
    {"name": "Charlie", "age": 35},
]

with db.atomic():
    User.insert_many(data).execute()

insert_many is much faster than calling User.create() in a loop — single INSERT vs many.

Still Not Working?

Peewee vs SQLAlchemy vs Tortoise vs SQLModel

• Peewee — Lightweight, sync-only, simple. Best for small projects, scripts, Flask apps.
• SQLAlchemy — Heavyweight, most flexible, sync + async. Best for complex enterprise apps. See SQLAlchemy not working.
• Tortoise — Django-style async ORM. See Tortoise ORM not working.
• SQLModel — FastAPI-aligned with Pydantic. See SQLModel not working.

Peewee is the right choice for projects where SQLAlchemy feels overkill — CLIs, scripts, small Flask apps, single-file demos. Its simplicity is the feature.

Async Wrapper (sync ORM in async code)

Run Peewee queries in a thread pool to avoid blocking:

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    loop = asyncio.get_running_loop()
    user = await loop.run_in_executor(None, User.get_or_none, User.id == user_id)
    if user is None:
        raise HTTPException(404)
    return {"id": user.id, "name": user.name}

For async code, prefer a native async ORM. Peewee in a thread pool works but adds overhead — performance-sensitive endpoints should use Tortoise or SQLModel + asyncpg.

For FastAPI dependency patterns with sync ORMs, see FastAPI dependency injection error.

Testing with In-Memory SQLite

import pytest
from peewee import SqliteDatabase

@pytest.fixture
def db():
    test_db = SqliteDatabase(":memory:")
    User._meta.database = test_db
    Post._meta.database = test_db
    test_db.connect()
    test_db.create_tables([User, Post])
    yield test_db
    test_db.drop_tables([User, Post])
    test_db.close()

peewee_test library provides utilities for cleaner test setups:

pip install peewee-test

For pytest fixture patterns with database state, see pytest fixture not found.

Signals (Hooks)

from playhouse.signals import post_save, pre_save, Model

class User(Model):
    name = CharField()

@post_save(sender=User)
def on_user_save(sender, instance, created):
    if created:
        send_welcome_email(instance.email)

Requires from playhouse.signals import Model (different from peewee.Model).

Database URL Parsing

For 12-factor apps reading the database URL from env:

from playhouse.db_url import connect

db = connect("postgres://user:pass@localhost:5432/mydb")
db = connect("sqlite:///app.db")
db = connect("mysql://user:pass@localhost/mydb")
db = connect(os.environ["DATABASE_URL"])

playhouse.db_url.connect returns the right Database subclass based on the scheme — much cleaner than instantiating each type manually.

Working with Existing Tables (Read-Only Mode)

When you can’t run migrations (read-only databases, third-party DBs):

class LegacyOrder(BaseModel):
    id = IntegerField()
    customer_name = CharField()

    class Meta:
        table_name = "legacy_orders"
        primary_key = False   # No PK constraint required

Set primary_key = False when reading tables without explicit primary keys (Peewee normally requires one).

Schema Introspection

# Generate models from existing database
python -m pwiz -e sqlite -o app.db > models.py

pwiz is Peewee’s introspection tool — useful when adopting Peewee on an existing database. Adjust the generated models manually after generation.