Peewee ORM

Peewee provides several SQLite database implementations depending on your requirements for performance, features, or security:

• SqliteDatabase: The core implementation. Supports pragmas, WAL-mode, user-defined functions, ATTACH/DETACH, and JSON. Full-text search (FTS) is available via playhouse.sqlite_ext.
• CySqliteDatabase (playhouse.cysqlite_ext): Extends SqliteDatabase using the cysqlite driver. Adds table-value functions, commit/rollback/update/progress/trace hooks, BLOB I/O, online backups, and supports encryption via SQLCipher.
• APSWDatabase (playhouse.apsw_ext): Extends SqliteDatabase using the apsw driver, providing access to the full range of SQLite functionality.
• SqlCipherDatabase (playhouse.sqlcipher_ext): Extends SqliteDatabase using the sqlcipher3 driver for transparent 256-bit AES full-database encryption.
• SqliteQueueDatabase (playhouse.sqliteq): Extends SqliteDatabase by using a long-lived background writer thread. This prevents timeouts and locking issues in multi-threaded environments with frequent writes.

When using db.atomic(), Peewee tracks nesting depth. The first call starts a transaction; subsequent calls within that block create savepoints. This allows for granular error recovery.

Example of rolling back only a specific part of a transaction:

with db.atomic():                        # Transaction begins
    User.create(username='charlie')

    with db.atomic() as sp:              # Savepoint begins
        User.create(username='huey')
        sp.rollback()                    # Rolls back 'huey' only
        User.create(username='alice')    # New savepoint begins here

    User.create(username='mickey')
# Result: charlie, alice, and mickey are committed. huey is not.

with db.atomic():                        # Transaction begins.
    User.create(username='charlie')

    with db.atomic() as sp:              # Savepoint begins.
        User.create(username='huey')
        sp.rollback()                    # Rolls back huey only.
        User.create(username='alice')    # New savepoint begins here.

    User.create(username='mickey')
# Committed: charlie, alice, mickey. huey was rolled back.

By default, to_pydantic treats foreign keys as flat scalar values (using the underlying column name, e.g., user_id). To embed the related object instead, use the relationships parameter.

Nested Foreign Keys

Pass a dictionary mapping the ForeignKeyField to the desired Pydantic schema.

# Include the id field in the response
UserSchema = to_pydantic(User, exclude_autofield=False)

# Embed the User object inside the Tweet response
TweetResponse = to_pydantic(
    Tweet, 
    exclude_autofield=False, 
    relationships={Tweet.user: UserSchema}
)

# To avoid extra SELECT queries during validation, use a JOIN
tweet = (Tweet.select(Tweet, User).join(User).get())
data = TweetResponse.model_validate(tweet)

Nested Back-references

Back-references (e.g., User.tweets) can also be nested, but because they represent a collection, the schema must be wrapped in typing.List.

from typing import List

# Exclude the 'user' FK from Tweet to prevent circular nesting
TweetResponse = to_pydantic(Tweet, exclude={'user'}, exclude_autofield=False)

# Map the backref to a list of TweetResponse schemas
UserDetail = to_pydantic(
    User, 
    exclude_autofield=False, 
    relationships={User.tweets: List[TweetResponse]}
)

# To avoid extra queries, use .with_related()
users = (User.select().where(User.id == 123).with_related(Load(User.tweets)))
data = UserDetail.model_validate(users[0])

Async Considerations

In async applications using the asyncio extension, lazy-loading a relation outside of db.run() will raise a MissingGreenletBridge. To safely validate models with unloaded relations, run the validation inside the bridge: data = await db.run(UserDetail.model_validate, user)

# Nested foreign key example
UserSchema = to_pydantic(User, exclude_autofield=False)
TweetResponse = to_pydantic(
    Tweet,
    exclude_autofield=False,
    relationships={Tweet.user: UserSchema})

tweet = Tweet.create(user=huey, content='hello')
data = TweetResponse.model_validate(tweet)
print(data.model_dump())
# {'id': 1, 'content': 'hello', 'user': {'id': 1, 'name': 'Huey', ...}, ...}

Peewee provides FTS5Model to leverage SQLite's FTS5 extension for high-performance text searching. An FTS index is typically used alongside a canonical source table. The index stores SearchField columns and uses a rowid (which can be explicitly declared via RowIDField) to link back to the source data.

Key Constraints:

• Only MATCH and rowid lookups are efficient; other queries cause full table scans.
• Constraints, foreign keys, and secondary indexes are not supported on FTS tables.
• All columns in an FTS5Model must be SearchField instances (though they can be marked unindexed=True to store metadata without making it searchable).

Storage Modes (Meta.options['content']):

• Default: The index keeps its own copy of the text. Simplest to use.
• External Content (content=Model): The index stores only search structures. Text is read from the source table on demand. Requires manual synchronization (e.g., via triggers) between the source and index.
• Contentless (content=''): Searchable text is indexed and then discarded. Matches return rowids, but no text can be read back from the index.

from peewee import *
from playhouse.sqlite_ext import FTS5Model, SearchField, RowIDField

db = SqliteDatabase('app.db')

class Document(Model):
    author = TextField()
    title = TextField()
    content = TextField()
    timestamp = DateTimeField()

    class Meta:
        database = db

class DocumentIndex(FTS5Model):
    rowid = RowIDField()
    title = SearchField()
    content = SearchField()
    author = SearchField(unindexed=True)  # Stored but not searchable.

    class Meta:
        database = db
        options = {'tokenize': 'porter unicode61', 'prefix': [3, 4]}

Migration files are Python scripts with a numeric prefix. They must define an up(migrator, db) function and optionally a down(migrator, db) function.

• Schema Migrations: Generated migrations use migrator.migrate(...) to perform operations like add_column, drop_table, or add_index.
• Data Migrations: For changes that require manipulating data (not just schema), use pwmigrate create <name> to scaffold a blank file, then write plain Python logic inside the up() and down() functions.
• Decoupling: When writing manual migrations, it is recommended to define local 'stub' models inside the up() function to decouple the migration from the current state of your application's model classes.

# Example of a manual/data migration
from peewee import *

def up(migrator, db):
    # Define a local stub to avoid dependency on app models
    class User(db.Model):
        class Meta:
            database = db
            table_name = 'user'

    # Perform data manipulation
    for user in User.select():
        user.karma = 10
        user.save()

def down(migrator, db):
    # Logic to revert the data change
    pass

The SqliteQueueDatabase serializes all write queries through a single long-lived connection on a dedicated background thread. This prevents write conflicts and timeouts in multi-threaded applications.

Key Constraints:

• No Transactions: Because writes from different threads are interleaved in the queue, atomic() and transaction() methods will raise a ValueError. Use this only when you do not need transaction guarantees.
• Read/Write Split: Read queries work normally (per-request connection), but only writes are funneled through the queue.

Lifecycle Management:

• If autostart=False, you must call db.start() manually.
• Use db.stop() on application shutdown to ensure pending writes are flushed.

Bypassing the Queue: To perform bulk imports or direct writes, use db.pause() to disconnect the writer thread, then db.unpause() to resume. While paused, queue writes will raise WriterPaused.

from playhouse.sqliteq import SqliteQueueDatabase

db = SqliteQueueDatabase(
    'my_app.db',
    use_gevent=False,
    autostart=True,
    queue_max_size=64,
    results_timeout=5.0,
    pragmas={'journal_mode': 'wal'}
)

# To bypass the queue for bulk operations:
db.pause()
# ... perform direct writes ...
db.unpause()

If your framework is not explicitly listed, follow this general pattern to manage Peewee connections:

1. Before Request: Find the hook that runs before every request handler and call db.connect().
2. After Request: Find the hook that runs after every request (handling both success and error cases) and call db.close() if the connection is open.

For WSGI frameworks: Use a middleware that wraps the application callable. This is a synchronous pattern.

For ASGI frameworks: Use an ASGI middleware to manage async connections (via playhouse.pwasyncio). Note that db.connect() and db.close() only work inside the async bridge; for async databases, use the async equivalents within the middleware.

# Synchronous WSGI Middleware Pattern
class PeeweeMiddleware:
    def __init__(self, app, database):
        self.app = app
        self.db = database

    def __call__(self, environ, start_response):
        self.db.connect()
        try:
            return self.app(environ, start_response)
        finally:
            if not self.db.is_closed():
                self.db.close()

# Wrap your WSGI app:
application = PeeweeMiddleware(application, db)

Expressions in Peewee are composed using Field instances and SQL aggregations/functions via the fn helper. You can use these expressions for comparisons, arithmetic, and even atomic updates.

Comparison Operators

Use standard Python comparison operators to compare fields against values:

• == (equal)
• < (less than)
• > (greater than)
• etc.

Combining Expressions

Use bitwise operators to combine multiple conditions. Note that Python's operator precedence applies, so use parentheses for clarity:

• & (AND)
• | (OR)
• ~ (NOT)

Arithmetic and Atomic Updates

Expressions support arithmetic operations. This is particularly useful for atomic updates, where you can increment or modify a value directly in the database without fetching it first.

# Comparison
User.username == 'charlie'

# Combining with bitwise operators
(User.is_admin == True) & (User.last_login >= today)
(User.is_active & ~User.is_admin)

# Arithmetic in expressions
(User.failed_logins > (User.login_count * .5))

# Atomic update: incrementing a value
User.update(login_count=User.login_count + 1).where(User.id == user_id)