Using SQLite

To connect to a SQLite database, we will use SqliteDatabase. Thefirst 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 providesmany SQLite-specific features such as full-text search,json extension support, and much, much more. If you wouldlike to use these awesome features, use the SqliteExtDatabase fromthe 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 throughPRAGMA 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 canspecify them as a dictionary or a list of 2-tuples containing the pragma nameand 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 thepragma() method or the special properties exposed onthe 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 berun whenever a connection is opened, specify permanent=True.

Note

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

Recommended Settings

The following settings are what I use with SQLite for a typical webapplication 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. TheSqliteDatabase class supports three types of user-definedextensions:

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

Note

For even more extension support, see SqliteExtDatabase, whichis 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 TableFunctionand 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 isperformed. The first read creates a shared lockand the first write creates a reserved lock.Because the acquisition of the lock is deferred until actually needed, it ispossible that another thread or process could create a separate transactionand write to the database after the BEGIN on the current thread has executed.
• Immediate - a reserved lockis acquired immediately. In this mode, no other database may write to thedatabase or open an immediate or exclusive transaction. Other processescan continue to read from the database, however.
• Exclusive - opens an exclusive lockwhich prevents all (except for read uncommitted) connections from accessingthe 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 Transactionsdocumentation.

APSW, an Advanced SQLite Driver

Peewee also comes with an alternate SQLite database that uses apsw, an advanced sqlite driver, anadvanced Python SQLite driver. More information on APSW can be obtained on theAPSW project website. APSW providesspecial 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 theapsw_ext module:

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