peewee
Peewee is a small ORM for SQLite, Postgres, MySQL and MariaDB that fits in one Python module with no required dependencies. You declare models as classes with field objects, and query them with a builder that stays close to SQL: User.select().join(Tweet).where(Tweet.created > cutoff).group_by(User) reads roughly like the statement it produces, and calling .sql() on any query shows you exactly what will run. There is no session, no unit of work and no identity map; you call .save() on an instance or .execute() on a query and something happens immediately. The playhouse package that ships alongside it holds the parts most people eventually need: connection pooling, database URLs, reflection, Flask and FastAPI helpers, Postgres and SQLite extensions including full-text search, and since version 4.0 a preliminary asyncio layer built on aiosqlite, asyncpg and aiomysql. It has been maintained by one person since 2010.
For SQLite-backed tools and small services, peewee gives you more usable SQL per line of learning than anything else in Python. The absence of a released migration runner and the single-maintainer risk are the two things to weigh before it becomes the ORM behind a long-lived production system.
Use it if
- You want an ORM a new developer can hold in their head: a few concepts, one module of source you can actually read, and query objects that map one to one onto the SQL they generate
- You are building something small or self-contained (a CLI, a scraper, a single-file service, a desktop app) where SQLite plus zero dependencies is exactly the right shape
- You use SQLite seriously. playhouse.sqlite_ext covers FTS4 and FTS5 with bm25 ranking, virtual tables, user-defined functions, the JSON1 extension and WAL tuning, and 4.3.0 added FTS5Model.web_query() so a search box's quoted phrases, AND/OR/NOT and -exclusions translate into a valid FTS5 query without you escaping anything
- You want to see and control the SQL. Window functions, CTEs, lateral joins, ON CONFLICT and RETURNING are all expressible in the builder rather than requiring a drop to raw strings
- You need real schema migrations. playhouse.migrate gives you a SchemaMigrator with add_column, rename_column and friends, but there is no version table, no ordering, no CLI and no autogeneration in 4.3.0. A migration runner called pwmigrate exists on master and is not in a release yet, so today you either write and track the scripts yourself or bolt on a separate tool
- Bus factor matters to you. One person has maintained this since 2010 and the GitHub issue tracker currently sits at zero open items, which is genuinely impressive discipline and also tells you exactly how many people are holding the project up
- Your stack is async-first. asyncio support arrived in 4.0 and the changelog still calls it preliminary; it works by bridging through greenlet for SQLite and MySQL and needs a separate model base and a-prefixed methods. SQLAlchemy 2.0's async support and Tortoise have far more production mileage
- You depend on static typing. Models use dynamic attributes and the documented style is from peewee import *, so mypy and pyright see almost nothing useful. SQLAlchemy 2.0 with Mapped annotations, or SQLModel, will actually catch a typo in a field name before runtime
- You are coming from SQLAlchemy and expect a session. There is no identity map, so loading the same row twice gives you two unrelated objects, and no dirty tracking across a transaction: every write is an explicit save or query. Relations lazy-load one query at a time unless you join or prefetch, which makes N+1 the default rather than the exception
- You need the surrounding ecosystem more than the ORM. Alembic, Django admin, dbt-style tooling and the long tail of Stack Overflow answers all point at the two big options; peewee questions are answered by reading peewee's docs
Setup reality
pip install peewee installs a single module with nothing behind it, and SQLite works immediately through the standard library's sqlite3. Other backends are extras: peewee[postgres] for psycopg2, peewee[psycopg3], peewee[mysql] for pymysql, and peewee[aiosqlite], peewee[asyncpg] or peewee[aiomysql] for the async layer, each of which also pulls greenlet where the bridge needs it. If both psycopg2 and psycopg3 are installed, 4.x prefers psycopg2 unless you pass prefer_psycopg3=True to PostgresqlDatabase. Two upgrade traps if you are moving from 3.x: 4.0 removed the Cython _sqlite_ext module, so CSqliteExtDatabase users move to SqliteExtDatabase or the new cysqlite_ext, and 3.19 stopped shipping the SQLite C extensions in wheels at all, so the ranking functions now come from pure Python unless you install from the sdist with --no-binary. The other thing that bites people is connection lifetime: peewee keeps one connection per thread and raises OperationalError if you connect twice, so web apps need an explicit open before each request and close after, or the Flask and FastAPI helpers in playhouse that do it for you.
Patterns
Declare models against a databasedefine-models
import datetime
from peewee import *
db = SqliteDatabase('app.db', pragmas={
'journal_mode': 'wal',
'foreign_keys': 1,
'busy_timeout': 5000,
})
class BaseModel(Model):
class Meta:
database = db
class User(BaseModel):
username = CharField(unique=True, max_length=64)
created = DateTimeField(default=datetime.datetime.now)
class Tweet(BaseModel):
user = ForeignKeyField(User, backref='tweets', on_delete='CASCADE')
body = TextField()
created = DateTimeField(default=datetime.datetime.now, index=True)
db.connect()
db.create_tables([User, Tweet])SQLite ignores foreign keys unless the pragma is on, and WAL plus a busy_timeout is what stops 'database is locked' the moment a second process appears. default takes the callable, not the call: DateTimeField(default=datetime.datetime.now()) freezes one timestamp at import time. create_tables is CREATE TABLE IF NOT EXISTS, so it never alters an existing table.
Bind the database after models are importeddeferred-database
from peewee import DatabaseProxy, Model
from playhouse.db_url import connect
database = DatabaseProxy()
class BaseModel(Model):
class Meta:
database = database
# in your entrypoint, once config is loaded
database.initialize(connect(os.environ['DATABASE_URL']))
# postgresql://user:pw@host/db, sqlite:///app.db, mysql://...Without the proxy, your model modules need the connection string at import time, which makes tests awkward. db_url.connect() understands postgresql, postgresql+psycopg3, mysql, sqlite and their extended variants, and in 4.3.0 raises ValueError for a URL with no database name rather than quietly reading it as a hostname.
Create, read, update, delete a rowcrud-basics
user = User.create(username='huey') # INSERT, returns the instance
found = User.get_or_none(User.username == 'huey') # None instead of raising
user, created = User.get_or_create(username='mickey', defaults={'active': True})
user.username = 'huey2'
user.save() # UPDATE by primary key
user.delete_instance(recursive=False) # DELETE this row
User.delete().where(User.active == False).execute() # bulk DELETEUser.get() raises User.DoesNotExist; get_or_none is usually what you meant. save() on an instance with no primary key set does an INSERT instead of an UPDATE, so an object built with User(...) and saved twice creates two rows unless you pass save(force_insert=False). delete_instance is the instance-level delete; Model.delete() is a classmethod that builds a query over the whole table.
Load related rows in one queryavoid-n-plus-one-join
# BAD: one extra SELECT per tweet
for tweet in Tweet.select():
print(tweet.user.username)
# GOOD: select both models, join, and peewee attaches the User to each Tweet
query = (Tweet
.select(Tweet, User)
.join(User)
.where(Tweet.created >= cutoff)
.order_by(Tweet.created.desc()))
for tweet in query:
print(tweet.user.username) # no further queriesPassing User to select() is the part people miss: .join(User) alone filters and orders but does not select the columns, so the attribute access still fires a query. This is the single most common peewee performance bug. Add .objects() if you would rather have the joined columns flattened onto the Tweet instance than nested.
Load the many side without a cartesian joinprefetch-collections
from peewee import prefetch
users = User.select().where(User.active == True)
tweets = Tweet.select().order_by(Tweet.created.desc())
for user in prefetch(users, tweets):
for tweet in user.tweets: # already populated, no query
print(user.username, tweet.body)A join duplicates the parent row once per child, which is fine for one-to-one and wasteful for one-to-many. prefetch runs one query per level and stitches the results in Python. It cannot apply a per-parent limit, so 'the three most recent tweets for each user' still needs a window function or a lateral join.
Insert many rows and handle conflictsbulk-insert-and-upsert
from peewee import chunked, EXCLUDED
rows = [{'username': n, 'active': True} for n in names]
with db.atomic():
for batch in chunked(rows, 500):
(User
.insert_many(batch)
.on_conflict(
conflict_target=[User.username],
preserve=[User.active], # take the new value
update={User.updated: datetime.datetime.now()})
.execute())chunked exists because SQLite caps bound variables per statement (999 on older builds, 32766 on newer ones), so a naive insert_many of ten thousand dicts raises OperationalError. Wrap the loop in atomic() or each batch commits separately and a failure halfway leaves partial data. on_conflict syntax differs by backend; MySQL ignores conflict_target.
Group writes and nest them safelytransactions
with db.atomic() as txn:
order = Order.create(user=user, total=0)
try:
with db.atomic(): # SAVEPOINT, because we are already in a txn
for line in lines:
OrderLine.create(order=order, **line)
except IntegrityError:
pass # inner savepoint rolled back, order survives
order.total = compute_total(order)
order.save()atomic() opens a transaction at the top level and a savepoint when nested, so you can use it in a helper without knowing whether a caller already started one. Do not mix it with db.begin() or manual commit in the same code path. Long-running atomic blocks hold a write lock on SQLite for their whole duration, so keep network calls out of them.
Open and close a connection per requestconnection-lifecycle
# FastAPI
from fastapi import Depends, FastAPI
app = FastAPI()
def get_db():
db.connect(reuse_if_open=True)
try:
yield db
finally:
if not db.is_closed():
db.close()
@app.get('/users')
def list_users(_=Depends(get_db)):
return [u.username for u in User.select()]peewee keeps one connection per thread and db.connect() raises OperationalError if one is already open, which is why reuse_if_open exists. Leaving connections open across requests under a threaded server exhausts the server's connection limit; for Postgres and MySQL use playhouse.pool.PooledPostgresqlDatabase instead of opening a fresh socket every request. As of 4.3.0 the pools roll back transactions left open at check-in and probe idle Postgres connections with SELECT 1 before handing them out.
See the statement before it runsinspect-generated-sql
query = (User
.select(User.username, fn.COUNT(Tweet.id).alias('n'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User.username)
.having(fn.COUNT(Tweet.id) > 5))
print(query.sql()) # (sql_string, [params])
import logging
logging.basicConfig()
logging.getLogger('peewee').setLevel(logging.DEBUG) # log every querysql() returns the statement and the parameter list separately, not an interpolated string, so do not paste the output into a client and expect it to run. Turning the peewee logger to DEBUG in a test run is the fastest way to find an N+1: count the lines.
Search with SQLite FTS5 and a user-facing query syntaxfull-text-search
from playhouse.sqlite_ext import FTS5Model, SearchField
class DocIndex(FTS5Model):
title = SearchField()
body = SearchField()
class Meta:
database = db
options = {'tokenize': 'porter unicode61'}
db.create_tables([DocIndex])
DocIndex.insert({'title': t, 'body': b}).execute()
results = (DocIndex
.search_bm25(DocIndex.web_query(user_input), weights={'title': 2.0})
.limit(20))web_query landed in 4.3.0 and translates what people type into a search box (quoted phrases, AND/OR/NOT, -exclusions, column: filters) into a valid FTS5 query, so covid-19 and c++ no longer raise a syntax error. Passing raw user input straight to search() is the bug it fixes. FTS5 tables are virtual: no foreign keys, and you keep them in sync with your content table yourself or use an external-content table.
Alter a table with the migratorschema-migration
from playhouse.migrate import SqliteMigrator, migrate
from peewee import BooleanField, CharField
migrator = SqliteMigrator(db)
with db.atomic():
migrate(
migrator.add_column('user', 'is_admin', BooleanField(default=False)),
migrator.rename_column('user', 'name', 'username'),
migrator.add_index('tweet', ('user_id', 'created'), False),
)This is a set of operations, not a migration system: nothing records that it ran, nothing orders it against other changes, and nothing generates it from your models. You supply the history table and the runner, or you keep a numbered scripts directory and a table of applied names. A first-party runner (pwmigrate) is on master and not in 4.3.0.
Run peewee on an event loopasyncio-queries
import asyncio
from peewee import CharField, ForeignKeyField, TextField
from playhouse.pwasyncio import AsyncPostgresqlDatabase
db = AsyncPostgresqlDatabase('my_app')
class User(db.Model):
username = CharField(unique=True)
class Tweet(db.Model):
user = ForeignKeyField(User, backref='tweets')
body = TextField()
async def main():
async with db:
await db.acreate_tables([User, Tweet])
huey = await User.acreate(username='huey')
await Tweet.acreate(user=huey, body='meow')
query = Tweet.select(Tweet, User).join(User)
for tweet in await query.aexecute():
print(tweet.user.username, tweet.body)
async for tweet in db.iterate(query): # server-side cursor
print(tweet.body)
await db.close_pool()
asyncio.run(main())Models inherit from db.Model, not peewee.Model, and every I/O method gains an a prefix: acreate, asave, aexecute, acreate_tables. Calling the sync version on an async database is the mistake that gets made, and it blocks the loop rather than failing loudly. The changelog still describes this layer as preliminary, and aiosqlite and aiomysql go through a greenlet bridge, so read the asyncio chapter before committing to it.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| sqlalchemy | PyPI | You need mature async, typed models, Alembic migrations and an ecosystem that assumes you are there |
| django | PyPI | You would rather get ORM, migrations, admin and auth as one opinionated package than assemble four things |
| tortoise-orm | PyPI | Your application is async from the ground up and you want a Django-shaped API that was designed for it |