Using SQLite

To connect to a SQLite database, we will use SqliteDatabase. The first parameter is the filename containing the database, or the string ':memory:' to create an in-memory database. After the database filename, you can specify a list or pragmas or any other arbitrary sqlite3 parameters.

sqlite_db = SqliteDatabase('my_app.db', pragmas={'journal_mode': 'wal'})
class BaseModel(Model):
    """A base model that will use our Sqlite database."""
    class Meta:
        database = sqlite_db
class User(BaseModel):
    username = TextField()
    # etc, etc

Peewee includes a SQLite extension module which provides many SQLite-specific features such as full-text search, json extension support, and much, much more. If you would like to use these awesome features, use the SqliteExtDatabase from the playhouse.sqlite_ext module:

from playhouse.sqlite_ext import SqliteExtDatabase
sqlite_db = SqliteExtDatabase('my_app.db', pragmas={
    'journal_mode': 'wal',  # WAL-mode.
    'cache_size': -64 * 1000,  # 64MB cache.
    'synchronous': 0})  # Let the OS manage syncing.

PRAGMA statements

SQLite allows run-time configuration of a number of parameters through PRAGMA statements (SQLite documentation). These statements are typically run when a new database connection is created. To run one or more PRAGMA statements against new connections, you can specify them as a dictionary or a list of 2-tuples containing the pragma name and value:

db = SqliteDatabase('my_app.db', pragmas={
    'journal_mode': 'wal',
    'cache_size': 10000,  # 10000 pages, or ~40MB
    'foreign_keys': 1,  # Enforce foreign-key constraints
})

PRAGMAs may also be configured dynamically using either the pragma() method or the special properties exposed on the SqliteDatabase object:

# Set cache size to 64MB for *current connection*.
db.pragma('cache_size', -1024 * 64)
# Same as above.
db.cache_size = -1024 * 64
# Read the value of several pragmas:
print('cache_size:', db.cache_size)
print('foreign_keys:', db.foreign_keys)
print('journal_mode:', db.journal_mode)
print('page_size:', db.page_size)
# Set foreign_keys pragma on current connection *AND* on all
# connections opened subsequently.
db.pragma('foreign_keys', 1, permanent=True)

Attention

Pragmas set using the pragma() method, by default, do not persist after the connection is closed. To configure a pragma to be run whenever a connection is opened, specify permanent=True.

Note

A full list of PRAGMA settings, their meaning and accepted values can be found in the SQLite documentation: http://sqlite.org/pragma.html

Recommended Settings

The following settings are what I use with SQLite for a typical web application database.

Example database using the above options:

db = SqliteDatabase('my_app.db', pragmas={
    'journal_mode': 'wal',
    'cache_size': -1 * 64000,  # 64MB
    'foreign_keys': 1,
    'ignore_check_constraints': 0,
    'synchronous': 0})

User-defined functions

SQLite can be extended with user-defined Python code. The SqliteDatabase class supports three types of user-defined extensions:

• Functions - which take any number of parameters and return a single value.
• Aggregates - which aggregate parameters from multiple rows and return a single value.
• Collations - which describe how to sort some value.

Note

For even more extension support, see SqliteExtDatabase, which is in the playhouse.sqlite_ext module.

Example user-defined function:

db = SqliteDatabase('analytics.db')
from urllib.parse import urlparse
@db.func('hostname')
def hostname(url):
    if url is not None:
        return urlparse(url).netloc
# Call this function in our code:
# The following finds the most common hostnames of referrers by count:
query = (PageView
         .select(fn.hostname(PageView.referrer), fn.COUNT(PageView.id))
         .group_by(fn.hostname(PageView.referrer))
         .order_by(fn.COUNT(PageView.id).desc()))

Example user-defined aggregate:

from hashlib import md5
@db.aggregate('md5')
class MD5Checksum(object):
    def __init__(self):
        self.checksum = md5()
    def step(self, value):
        self.checksum.update(value.encode('utf-8'))
    def finalize(self):
        return self.checksum.hexdigest()
# Usage:
# The following computes an aggregate MD5 checksum for files broken
# up into chunks and stored in the database.
query = (FileChunk
         .select(FileChunk.filename, fn.MD5(FileChunk.data))
         .group_by(FileChunk.filename)
         .order_by(FileChunk.filename, FileChunk.sequence))

Example collation:

@db.collation('ireverse')
def collate_reverse(s1, s2):
    # Case-insensitive reverse.
    s1, s2 = s1.lower(), s2.lower()
    return (s1 < s2) - (s1 > s2)  # Equivalent to -cmp(s1, s2)
# To use this collation to sort books in reverse order...
Book.select().order_by(collate_reverse.collation(Book.title))
# Or...
Book.select().order_by(Book.title.asc(collation='reverse'))

Example user-defined table-value function (see TableFunction and table_function) for additional details:

from playhouse.sqlite_ext import TableFunction
db = SqliteDatabase('my_app.db')
@db.table_function('series')
class Series(TableFunction):
    columns = ['value']
    params = ['start', 'stop', 'step']
    def initialize(self, start=0, stop=None, step=1):
        """
        Table-functions declare an initialize() method, which is
        called with whatever arguments the user has called the
        function with.
        """
        self.start = self.current = start
        self.stop = stop or float('Inf')
        self.step = step
    def iterate(self, idx):
        """
        Iterate is called repeatedly by the SQLite database engine
        until the required number of rows has been read **or** the
        function raises a `StopIteration` signalling no more rows
        are available.
        """
        if self.current > self.stop:
            raise StopIteration
        ret, self.current = self.current, self.current + self.step
        return (ret,)
# Usage:
cursor = db.execute_sql('SELECT * FROM series(?, ?, ?)', (0, 5, 2))
for value, in cursor:
    print(value)
# Prints:
# 0
# 2
# 4

For more information, see:

• SqliteDatabase.func()
• SqliteDatabase.aggregate()
• SqliteDatabase.collation()
• SqliteDatabase.table_function()
• For even more SQLite extensions, see SQLite Extensions

Set locking mode for transaction

SQLite transactions can be opened in three different modes:

• Deferred (default) - only acquires lock when a read or write is performed. The first read creates a shared lock and the first write creates a reserved lock. Because the acquisition of the lock is deferred until actually needed, it is possible that another thread or process could create a separate transaction and write to the database after the BEGIN on the current thread has executed.
• Immediate - a reserved lock is acquired immediately. In this mode, no other database may write to the database or open an immediate or exclusive transaction. Other processes can continue to read from the database, however.
• Exclusive - opens an exclusive lock which prevents all (except for read uncommitted) connections from accessing the database until the transaction is complete.

Example specifying the locking mode:

db = SqliteDatabase('app.db')
with db.atomic('EXCLUSIVE'):
    do_something()
@db.atomic('IMMEDIATE')
def some_other_function():
    # This function is wrapped in an "IMMEDIATE" transaction.
    do_something_else()

For more information, see the SQLite locking documentation. To learn more about transactions in Peewee, see the Managing Transactions documentation.

APSW, an Advanced SQLite Driver

Peewee also comes with an alternate SQLite database that uses apsw, an advanced sqlite driver, an advanced Python SQLite driver. More information on APSW can be obtained on the APSW project website. APSW provides special features like:

• Virtual tables, virtual file-systems, Blob I/O, backups and file control.
• Connections can be shared across threads without any additional locking.
• Transactions are managed explicitly by your code.
• Unicode is handled correctly.
• APSW is faster that the standard library sqlite3 module.
• Exposes pretty much the entire SQLite C API to your Python app.

If you would like to use APSW, use the APSWDatabase from the apsw_ext module:

from playhouse.apsw_ext import APSWDatabase
apsw_db = APSWDatabase('my_app.db')