pip install peewee

git clone https://github.com/coleifer/peewee.git
cd peewee
python setup.py install

# Build the C extension and place shared libraries alongside other modules.
python setup.py build_ext -i

python runtests.py

python runtests.py --help

fatal error: sqlite3.h: No such file or directory

$ NO_SQLITE=1 python setup.py install

from peewee import *
db = SqliteDatabase('people.db')
class Person(Model):


    name = CharField()


    birthday = DateField()


    class Meta:


        database = db # This model uses the "people.db" database.

Also note that we named our model Person instead of People. This is a convention you should follow -- even though the table will contain multiple people, we always name the class using the singular form.

class Pet(Model):


    owner = ForeignKeyField(Person, backref='pets')


    name = CharField()


    animal_type = CharField()


    class Meta:


        database = db # this model uses the "people.db" database

db.connect()

db.create_tables([Person, Pet])

from datetime import date
uncle_bob = Person(name='Bob', birthday=date(1960, 1, 15))
uncle_bob.save() # bob is now stored in the database
# Returns: 1

grandma = Person.create(name='Grandma', birthday=date(1935, 3, 1))
herb = Person.create(name='Herb', birthday=date(1950, 5, 5))

grandma.name = 'Grandma L.'
grandma.save()  # Update grandma's name in the database.
# Returns: 1

bob_kitty = Pet.create(owner=uncle_bob, name='Kitty', animal_type='cat')
herb_fido = Pet.create(owner=herb, name='Fido', animal_type='dog')
herb_mittens = Pet.create(owner=herb, name='Mittens', animal_type='cat')
herb_mittens_jr = Pet.create(owner=herb, name='Mittens Jr', animal_type='cat')

herb_mittens.delete_instance() # he had a great life
# Returns: 1

herb_fido.owner = uncle_bob
herb_fido.save()

grandma = Person.select().where(Person.name == 'Grandma L.').get()

grandma = Person.get(Person.name == 'Grandma L.')

for person in Person.select():


    print(person.name)
# prints:
# Bob
# Grandma L.
# Herb

query = Pet.select().where(Pet.animal_type == 'cat')
for pet in query:


    print(pet.name, pet.owner.name)
# prints:
# Kitty Bob
# Mittens Jr Herb

For an in-depth guide to working with relationships and joins, refer to the relationships documentation.

query = (Pet


         .select(Pet, Person)


         .join(Person)


         .where(Pet.animal_type == 'cat'))
for pet in query:


    print(pet.name, pet.owner.name)
# prints:
# Kitty Bob
# Mittens Jr Herb

for pet in Pet.select().join(Person).where(Person.name == 'Bob'):


    print(pet.name)
# prints:
# Kitty
# Fido

for pet in Pet.select().where(Pet.owner == uncle_bob):


    print(pet.name)

for pet in Pet.select().where(Pet.owner == uncle_bob).order_by(Pet.name):


    print(pet.name)
# prints:
# Fido
# Kitty

for person in Person.select().order_by(Person.birthday.desc()):


    print(person.name, person.birthday)
# prints:
# Bob 1960-01-15
# Herb 1950-05-05
# Grandma L. 1935-03-01

d1940 = date(1940, 1, 1)
d1960 = date(1960, 1, 1)
query = (Person


         .select()


         .where((Person.birthday < d1940) | (Person.birthday > d1960)))
for person in query:


    print(person.name, person.birthday)
# prints:
# Bob 1960-01-15
# Grandma L. 1935-03-01

query = (Person


         .select()


         .where(Person.birthday.between(d1940, d1960)))
for person in query:


    print(person.name, person.birthday)
# prints:
# Herb 1950-05-05

for person in Person.select():


    print(person.name, person.pets.count(), 'pets')
# prints:
# Bob 2 pets
# Grandma L. 0 pets
# Herb 1 pets

query = (Person


         .select(Person, fn.COUNT(Pet.id).alias('pet_count'))


         .join(Pet, JOIN.LEFT_OUTER)  # include people without pets.


         .group_by(Person)


         .order_by(Person.name))
for person in query:


    # "pet_count" becomes an attribute on the returned model instances.


    print(person.name, person.pet_count, 'pets')
# prints:
# Bob 2 pets
# Grandma L. 0 pets
# Herb 1 pets

query = (Person


         .select(Person, Pet)


         .join(Pet, JOIN.LEFT_OUTER)


         .order_by(Person.name, Pet.name))
for person in query:


    # We need to check if they have a pet instance attached, since not all


    # people have pets.


    if hasattr(person, 'pet'):


        print(person.name, person.pet.name)


    else:


        print(person.name, 'no pets')
# prints:
# Bob Fido
# Bob Kitty
# Grandma L. no pets
# Herb Mittens Jr

query = Person.select().order_by(Person.name).prefetch(Pet)
for person in query:


    print(person.name)


    for pet in person.pets:


        print('  *', pet.name)
# prints:
# Bob
#   * Kitty
#   * Fido
# Grandma L.
# Herb
#   * Mittens Jr

expression = fn.Lower(fn.Substr(Person.name, 1, 1)) == 'g'
for person in Person.select().where(expression):


    print(person.name)
# prints:
# Grandma L.

db.close()

python -m pwiz -e postgresql charles_blog > blog_models.py

pip install flask

python run_example.py

# create a peewee database instance -- our models will use this database to
# persist information
database = SqliteDatabase(DATABASE)
# model definitions -- the standard "pattern" is to define a base model class
# that specifies which database to use.  then, any subclasses will automatically
# use the correct storage.
class BaseModel(Model):


    class Meta:


        database = database
# the user model specifies its fields (or columns) declaratively, like django
class User(BaseModel):


    username = CharField(unique=True)


    password = CharField()


    email = CharField()


    join_date = DateTimeField()
# this model contains two foreign keys to user -- it essentially allows us to
# model a "many-to-many" relationship between users.  by querying and joining
# on different columns we can expose who a user is "related to" and who is
# "related to" a given user
class Relationship(BaseModel):


    from_user = ForeignKeyField(User, backref='relationships')


    to_user = ForeignKeyField(User, backref='related_to')


    class Meta:


        # `indexes` is a tuple of 2-tuples, where the 2-tuples are


        # a tuple of column names to index and a boolean indicating


        # whether the index is unique or not.


        indexes = (


            # Specify a unique multi-column index on from/to-user.


            (('from_user', 'to_user'), True),


        )
# a dead simple one-to-many relationship: one user has 0..n messages, exposed by
# the foreign key.  because we didn't specify, a users messages will be accessible
# as a special attribute, User.messages
class Message(BaseModel):


    user = ForeignKeyField(User, backref='messages')


    content = TextField()


    pub_date = DateTimeField()

def create_tables():


    with database:


        database.create_tables([User, Relationship, Message])

>>> from app import *
>>> create_tables()

Alternatively, you can use the schema migrations extension to alter your database schema using Python.

DATABASE = 'tweepee.db'
# Create a database instance that will manage the connection and
# execute queries
database = SqliteDatabase(DATABASE)
# Create a base-class all our models will inherit, which defines
# the database we'll be using.
class BaseModel(Model):


    class Meta:


        database = database

@app.before_request
def before_request():


    database.connect()
@app.after_request
def after_request(response):


    database.close()


    return response

def following(self):


    # query other users through the "relationship" table


    return (User


            .select()


            .join(Relationship, on=Relationship.to_user)


            .where(Relationship.from_user == self)


            .order_by(User.username))
def followers(self):


    return (User


            .select()


            .join(Relationship, on=Relationship.from_user)


            .where(Relationship.to_user == self)


            .order_by(User.username))

try:


    with database.atomic():


        # Attempt to create the user. If the username is taken, due to the


        # unique constraint, the database will raise an IntegrityError.


        user = User.create(


            username=request.form['username'],


            password=md5(request.form['password']).hexdigest(),


            email=request.form['email'],


            join_date=datetime.datetime.now())


    # mark the user as being 'authenticated' by setting the session vars


    auth_user(user)


    return redirect(url_for('homepage'))
except IntegrityError:


    flash('That username is already taken')

user = get_object_or_404(User, username=username)
try:


    with database.atomic():


        Relationship.create(


            from_user=get_current_user(),


            to_user=user)
except IntegrityError:


    pass

# python code
user = get_current_user()
messages = (Message


            .select()


            .where(Message.user.in_(user.following()))


            .order_by(Message.pub_date.desc()))

SELECT t1."id", t1."user_id", t1."content", t1."pub_date"
FROM "message" AS t1
WHERE t1."user_id" IN (


    SELECT t2."id"


    FROM "user" AS t2


    INNER JOIN "relationship" AS t3


        ON t2."id" = t3."to_user_id"


    WHERE t3."from_user_id" = ?
)

CREATE TABLE IF NOT EXISTS "event" (


    "id" INTEGER NOT NULL PRIMARY KEY,


    "key" TEXT NOT NULL,


    "timestamp" DATETIME NOT NULL,


    "metadata" TEXT NOT NULL);

>>> from peewee import SqliteDatabase
>>> from playhouse.reflection import generate_models, print_model, print_table_sql
>>>

>>> db = SqliteDatabase('events.db')
>>> models = generate_models(db)
>>> list(models.items())
[('events', <Model: event>)]
>>> globals().update(models)  # Inject models into global namespace.
>>> event
<Model: event>

>>> print_model(event)
event


  id AUTO


  key TEXT


  timestamp DATETIME


  metadata TEXT

>>> print_table_sql(event)
CREATE TABLE IF NOT EXISTS "event" (


  "id" INTEGER NOT NULL PRIMARY KEY,


  "key" TEXT NOT NULL,


  "timestamp" DATETIME NOT NULL,


  "metadata" TEXT NOT NULL)

>>> for e in event.select().order_by(event.timestamp).limit(5):
...     print(e.key, e.timestamp)
...
e00 2019-01-01 00:01:00
e01 2019-01-01 00:02:00
e02 2019-01-01 00:03:00
e03 2019-01-01 00:04:00
e04 2019-01-01 00:05:00
>>> event.select(fn.MIN(event.timestamp), fn.MAX(event.timestamp)).scalar(as_tuple=True)
(datetime.datetime(2019, 1, 1, 0, 1), datetime.datetime(2019, 1, 1, 1, 0))
>>> event.select().count()  # Or, len(event)
60

$ pwiz -e sqlite events.db > events.py

from peewee import *
# SQLite database using WAL journal mode and 64MB cache.
sqlite_db = SqliteDatabase('/path/to/app.db', pragmas={


    'journal_mode': 'wal',


    'cache_size': -1024 * 64})
# Connect to a MySQL database on network.
mysql_db = MySQLDatabase('my_app', user='app', password='db_password',


                         host='10.1.0.8', port=3306)
# Connect to a Postgres database.
pg_db = PostgresqlDatabase('my_app', user='postgres', password='secret',


                           host='10.1.0.9', port=5432)

from playhouse.sqlite_ext import SqliteExtDatabase
# Use SQLite (will register a REGEXP function and set busy timeout to 3s).
db = SqliteExtDatabase('/path/to/app.db', regexp_function=True, timeout=3,


                       pragmas={'journal_mode': 'wal'})
from playhouse.postgres_ext import PostgresqlExtDatabase
# Use Postgres (and register hstore extension).
db = PostgresqlExtDatabase('my_app', user='postgres', register_hstore=True)
from playhouse.cockroachdb import CockroachDatabase
# Use CockroachDB.
db = CockroachDatabase('my_app', user='root', port=26257, host='10.1.0.8')

db = PostgresqlDatabase(


    'database_name',  # Required by Peewee.


    user='postgres',  # Will be passed directly to psycopg2.


    password='secret',  # Ditto.


    host='db.mysite.com')  # Ditto.

db = MySQLDatabase('database_name', user='www-data', charset='utf8mb4')

psql_db = PostgresqlDatabase('my_database', user='postgres')
class BaseModel(Model):


    """A base model that will use our Postgresql database"""


    class Meta:


        database = psql_db
class User(BaseModel):


    username = CharField()

from playhouse.postgres_ext import PostgresqlExtDatabase
psql_db = PostgresqlExtDatabase('my_database', user='postgres')

from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE
db = PostgresqlDatabase('my_app', user='postgres', host='db-host',


                        isolation_level=ISOLATION_LEVEL_SERIALIZABLE)

db = PostgresqlDatabase(...)
conn = db.connection()  # returns current connection.
from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE
conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

To run this every time a connection is created, subclass and implement the _initialize_database() hook, which is designed for this purpose:

class SerializedPostgresqlDatabase(PostgresqlDatabase):


    def _initialize_connection(self, conn):


        conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

from playhouse.cockroachdb import CockroachDatabase
db = CockroachDatabase('my_app', user='root', port=26257, host='localhost')

def create_user(email):


    # Callable that accepts a single argument (the database instance) and


    # which is responsible for executing the transactional SQL.


    def callback(db_ref):


        return User.create(email=email)


    return db.run_transaction(callback, max_attempts=10)
huey = create_user('huey@example.com')

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

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.

db = SqliteDatabase('my_app.db', pragmas={


    'journal_mode': 'wal',


    'cache_size': 10000,  # 10000 pages, or ~40MB


    'foreign_keys': 1,  # Enforce foreign-key constraints
})

# 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)

db = SqliteDatabase('my_app.db', pragmas={


    'journal_mode': 'wal',


    'cache_size': -1 * 64000,  # 64MB


    'foreign_keys': 1,


    'ignore_check_constraints': 0,


    'synchronous': 0})

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()))

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))

@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'))

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

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()

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

mysql_db = MySQLDatabase('my_database')
class BaseModel(Model):


    """A base model that will use our MySQL database"""


    class Meta:


        database = mysql_db
class User(BaseModel):


    username = CharField()


    # etc, etc

import os
from peewee import *
from playhouse.db_url import connect
# Connect to the database URL defined in the environment, falling
# back to a local Sqlite database if no database URL is specified.
db = connect(os.environ.get('DATABASE') or 'sqlite:///default.db')
class BaseModel(Model):


    class Meta:


        database = db

database = PostgresqlDatabase(None)  # Un-initialized database.
class SomeModel(Model):


    class Meta:


        database = database

>>> database.connect()
Exception: Error, database not properly initialized before opening connection

database_name = input('What is the name of the db? ')
database.init(database_name, host='localhost', user='postgres')

database_proxy = DatabaseProxy()  # Create a proxy for our db.
class BaseModel(Model):


    class Meta:


        database = database_proxy  # Use proxy for our DB.
class User(BaseModel):


    username = CharField()
# Based on configuration, use a different database.
if app.config['DEBUG']:


    database = SqliteDatabase('local.db')
elif app.config['TESTING']:


    database = SqliteDatabase(':memory:')
else:


    database = PostgresqlDatabase('mega_production_db')
# Configure our proxy to use the db we specified in config.
database_proxy.initialize(database)

However, if it is only connection values that vary at run-time, such as the path to the database file, or the database host, you should instead use Database.init(). See Run-time database configuration for more details.

# The usual way:
db = SqliteDatabase('my_app.db', pragmas={'journal_mode': 'wal'})
# Specify the details at run-time:
db = SqliteDatabase(None)
...
db.init(db_filename, pragmas={'journal_mode': 'wal'})
# Or use a placeholder:
db = DatabaseProxy()
...
db.initialize(SqliteDatabase('my_app.db', pragmas={'journal_mode': 'wal'}))

class User(Model):


    username = TextField()
class Tweet(Model):


    user = ForeignKeyField(User, backref='tweets')


    content = TextField()


    timestamp = TimestampField()

postgres_db = PostgresqlDatabase('my_app', user='postgres')
sqlite_db = SqliteDatabase('my_app.db')
# At this point, the User and Tweet models are NOT bound to any database.
# Let's bind them to the Postgres database:
postgres_db.bind([User, Tweet])
# Now we will temporarily bind them to the sqlite database:
with sqlite_db.bind_ctx([User, Tweet]):


    # User and Tweet are now bound to the sqlite database.


    assert User._meta.database is sqlite_db
# User and Tweet are once again bound to the Postgres database.
assert User._meta.database is postgres_db

# Bind the user model to the sqlite db. By default, Peewee will also
# bind any models that are related to User via foreign-key as well.
User.bind(sqlite_db)
assert User._meta.database is sqlite_db
assert Tweet._meta.database is sqlite_db  # Related models bound too.
# Here we will temporarily bind *just* the User model to the postgres db.
with User.bind_ctx(postgres_db, bind_backrefs=False):


    assert User._meta.database is postgres_db


    assert Tweet._meta.database is sqlite_db  # Has not changed.
# And now User is back to being bound to the sqlite_db.
assert User._meta.database is sqlite_db

>>> db = SqliteDatabase(':memory:')  # In-memory SQLite database.
>>> db.connect()
True

>>> db.connect()
Traceback (most recent call last):


  File "<stdin>", line 1, in <module>


  File "/home/charles/pypath/peewee.py", line 2390, in connect


    raise OperationalError('Connection already opened.')
peewee.OperationalError: Connection already opened.

>>> db.close()  # Close connection.
True
>>> db.connect()
True
>>> db.connect(reuse_if_open=True)
False

>>> db.close()
True

>>> db.connect()  # Open connection.
True
>>> db.close()  # Close connection.
True
>>> db.close()  # Connection already closed, returns False.
False

>>> db.is_closed()
True

db = PostgresqlDatabase('my_app', user='postgres', autoconnect=False)

>>> db.is_closed()
True
>>> with db:
...     print(db.is_closed())  # db is open inside context manager.
...
False
>>> db.is_closed()  # db is closed.
True

>>> with db.connection_context():
...     # db connection is open.
...     pass
...
>>> db.is_closed()  # db connection is closed.
True

@db.connection_context()
def prepare_database():


    # DB connection will be managed by the decorator, which opens


    # a connection, calls function, and closes upon returning.


    db.create_tables(MODELS)  # Create schema.


    load_fixture_data(db)

>>> db.connection()
<sqlite3.Connection object at 0x7f94e9362f10>

from playhouse.pool import PooledPostgresqlExtDatabase
db = PooledPostgresqlExtDatabase(


    'my_database',


    max_connections=8,


    stale_timeout=300,


    user='postgres')
class BaseModel(Model):


    class Meta:


        database = db

# tests.py
import unittest
from my_app.models import EventLog, Relationship, Tweet, User
MODELS = [User, Tweet, EventLog, Relationship]
# use an in-memory SQLite for tests.
test_db = SqliteDatabase(':memory:')
class BaseTestCase(unittest.TestCase):


    def setUp(self):


        # Bind model classes to test db. Since we have a complete list of


        # all models, we do not need to recursively bind dependencies.


        test_db.bind(MODELS, bind_refs=False, bind_backrefs=False)


        test_db.connect()


        test_db.create_tables(MODELS)


    def tearDown(self):


        # Not strictly necessary since SQLite in-memory databases only live


        # for the duration of the connection, and in the next step we close


        # the connection...but a good practice all the same.


        test_db.drop_tables(MODELS)


        # Close connection to db.


        test_db.close()


        # If we wanted, we could re-bind the models to their original


        # database here. But for tests this is probably not necessary.

from gevent.socket import wait_read, wait_write
from psycopg2 import extensions
# Call this function after monkey-patching socket (etc).
def patch_psycopg2():


    extensions.set_wait_callback(_psycopg2_gevent_callback)
def _psycopg2_gevent_callback(conn, timeout=None):


    while True:


        state = conn.poll()


        if state == extensions.POLL_OK:


            break


        elif state == extensions.POLL_READ:


            wait_read(conn.fileno(), timeout=timeout)


        elif state == extensions.POLL_WRITE:


            wait_write(conn.fileno(), timeout=timeout)


        else:


            raise ValueError('poll() returned unexpected result')

from flask import Flask
from peewee import *
database = SqliteDatabase('my_app.db')
app = Flask(__name__)
# This hook ensures that a connection is opened to handle any queries
# generated by the request.
@app.before_request
def _db_connect():


    database.connect()
# This hook ensures that the connection is closed when we've finished
# processing the request.
@app.teardown_request
def _db_close(exc):


    if not database.is_closed():


        database.close()

# middleware.py
from my_blog.db import database  # Import the peewee database instance.
def PeeweeConnectionMiddleware(get_response):


    def middleware(request):


        database.connect()


        try:


            response = get_response(request)


        finally:


            if not database.is_closed():


                database.close()


        return response


    return middleware
# Older Django < 1.10 middleware.
class PeeweeConnectionMiddleware(object):


    def process_request(self, request):


        database.connect()


    def process_response(self, request, response):


        if not database.is_closed():


            database.close()


        return response

# settings.py
MIDDLEWARE_CLASSES = (


    # Our custom middleware appears first in the list.


    'my_blog.middleware.PeeweeConnectionMiddleware',


    # These are the default Django 1.7 middlewares. Yours may differ,


    # but the important this is that our Peewee middleware comes first.


    'django.middleware.common.CommonMiddleware',


    'django.contrib.sessions.middleware.SessionMiddleware',


    'django.middleware.csrf.CsrfViewMiddleware',


    'django.contrib.auth.middleware.AuthenticationMiddleware',


    'django.contrib.messages.middleware.MessageMiddleware',
)
# ... other Django settings ...

# app.py
from bottle import hook  #, route, etc, etc.
from peewee import *
db = SqliteDatabase('my-bottle-app.db')
@hook('before_request')
def _connect_db():


    db.connect()
@hook('after_request')
def _close_db():


    if not db.is_closed():


        db.close()
# Rest of your bottle app goes here.

db = SqliteDatabase('my_webpy_app.db')
def connection_processor(handler):


    db.connect()


    try:


        return handler()


    finally:


        if not db.is_closed():


            db.close()
app.add_processor(connection_processor)

from tornado.web import RequestHandler
db = SqliteDatabase('my_db.db')
class PeeweeRequestHandler(RequestHandler):


    def prepare(self):


        db.connect()


        return super(PeeweeRequestHandler, self).prepare()


    def on_finish(self):


        if not db.is_closed():


            db.close()


        return super(PeeweeRequestHandler, self).on_finish()

def peewee_middleware(request, following):


    db.connect()


    try:


        response = following(request)


    finally:


        if not db.is_closed():


            db.close()


    return response
app = WSGIApplication(middleware=[


    lambda x: peewee_middleware,


    # ... other middlewares ...
])

import falcon
from peewee import *
database = SqliteDatabase('my_app.db')
class PeeweeConnectionMiddleware(object):


    def process_request(self, req, resp):


        database.connect()


    def process_response(self, req, resp, resource, req_succeeded):


        if not database.is_closed():


            database.close()
application = falcon.API(middleware=[


    PeeweeConnectionMiddleware(),


    # ... other middlewares ...
])

from pyramid.request import Request
db = SqliteDatabase('pyramidapp.db')
class MyRequest(Request):


    def __init__(self, *args, **kwargs):


        super().__init__(*args, **kwargs)


        db.connect()


        self.add_finished_callback(self.finish)


    def finish(self, request):


        if not db.is_closed():


            db.close()

def main(global_settings, **settings):


    config = Configurator(settings=settings, ...)


    config.set_request_factory(MyRequest)

def _db_connect():


    db.connect()
def _db_close():


    if not db.is_closed():


        db.close()
cherrypy.engine.subscribe('before_request', _db_connect)
cherrypy.engine.subscribe('after_request', _db_close)

# app.py
@app.middleware('request')
async def handle_request(request):


    db.connect()
@app.middleware('response')
async def handle_response(request, response):


    if not db.is_closed():


        db.close()

from fastapi import FastAPI
from peewee import *
db = SqliteDatabase('my_app.db')
app = FastAPI()
# This hook ensures that a connection is opened to handle any queries
# generated by the request.
@app.on_event("startup")
def startup():


    db.connect()
# This hook ensures that the connection is closed when we've finished
# processing the request.
@app.on_event("shutdown")
def shutdown():


    if not db.is_closed():


        db.close()

db = SqliteDatabase('my_app.db')
db.connect()
# Example of executing a simple query and ignoring the results.
db.execute_sql("ATTACH DATABASE ':memory:' AS cache;")
# Example of iterating over the results of a query using the cursor.
cursor = db.execute_sql('SELECT * FROM users WHERE status = ?', (ACTIVE,))
for row in cursor.fetchall():


    # Do something with row, which is a tuple containing column data.


    pass

db = SqliteDatabase(':memory:')
with db.atomic() as txn:


    # This is the outer-most level, so this block corresponds to


    # a transaction.


    User.create(username='charlie')


    with db.atomic() as nested_txn:


        # This block corresponds to a savepoint.


        User.create(username='huey')


        # This will roll back the above create() query.


        nested_txn.rollback()


    User.create(username='mickey')
# When the block ends, the transaction is committed (assuming no error
# occurs). At that point there will be two users, "charlie" and "mickey".

try:


    with db.atomic():


        user = User.create(username=username)


    return 'Success'
except peewee.IntegrityError:


    return 'Failure: %s is already in use.' % username

@db.atomic()
def create_user(username):


    # This statement will run in a transaction. If the caller is already


    # running in an `atomic` block, then a savepoint will be used instead.


    return User.create(username=username)
create_user('charlie')

with db.atomic() as txn:


    perform_operation()


    with db.atomic() as nested_txn:


        perform_another_operation()

db = SqliteDatabase(':memory:')
with db.transaction() as txn:


    # Delete the user and their associated tweets.


    user.delete_instance(recursive=True)

with db.transaction() as txn:


    User.create(username='mickey')


    txn.commit()  # Changes are saved and a new transaction begins.


    User.create(username='huey')


    # Roll back. "huey" will not be saved, but since "mickey" was already


    # committed, that row will remain in the database.


    txn.rollback()
with db.transaction() as txn:


    User.create(username='whiskers')


    # Roll back changes, which removes "whiskers".


    txn.rollback()


    # Create a new row for "mr. whiskers" which will be implicitly committed


    # at the end of the `with` block.


    User.create(username='mr. whiskers')

with db.transaction() as txn:


    with db.savepoint() as sp:


        User.create(username='mickey')


    with db.savepoint() as sp2:


        User.create(username='zaizee')


        sp2.rollback()  # "zaizee" will not be saved, but "mickey" will be.

with db.manual_commit():


    db.begin()  # Have to begin transaction explicitly.


    try:


        user.delete_instance(recursive=True)


    except:


        db.rollback()  # Rollback! An error occurred.


        raise


    else:


        try:


            db.commit()  # Commit changes.


        except:


            db.rollback()


            raise

# Print all queries to stderr.
import logging
logger = logging.getLogger('peewee')
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)

from peewee import Database
import foodb  # Our fictional DB-API 2.0 driver.
class FooDatabase(Database):


    def _connect(self, database, **kwargs):


        return foodb.connect(database, **kwargs)

class FooDatabase(Database):


    def _connect(self, database, **kwargs):


        return foodb.connect(database, **kwargs)


    def get_tables(self):


        res = self.execute('SHOW TABLES;')


        return [r[0] for r in res.fetchall()]

from peewee import *
from foodb_ext import FooDatabase
db = FooDatabase('my_database', user='foo', password='secret')
class BaseModel(Model):


    class Meta:


        database = db
class Blog(BaseModel):


    title = CharField()


    contents = TextField()


    pub_date = DateTimeField()

import datetime
from peewee import *
db = SqliteDatabase('my_app.db')
class BaseModel(Model):


    class Meta:


        database = db
class User(BaseModel):


    username = CharField(unique=True)
class Tweet(BaseModel):


    user = ForeignKeyField(User, backref='tweets')


    message = TextField()


    created_date = DateTimeField(default=datetime.datetime.now)


    is_published = BooleanField(default=True)

class User(Model):


    username = CharField()


    join_date = DateTimeField()


    about_me = TextField()

class Message(Model):


    user = ForeignKeyField(User, backref='messages')


    body = TextField()


    send_date = DateTimeField(default=datetime.datetime.now)

>>> print(some_message.user.username)
Some User
>>> for message in some_user.messages:
...     print(message.body)
some message
another message
yet another message

To add database (server-side) constraints, use the constraints parameter.

class Message(Model):


    context = TextField()


    read_count = IntegerField(default=0)

class Message(Model):


    context = TextField()


    timestamp = DateTimeField(default=datetime.datetime.now)

class Message(Model):


    context = TextField()


    timestamp = DateTimeField(constraints=[SQL('DEFAULT CURRENT_TIMESTAMP')])

tweets = (Tweet


          .select(Tweet, User)


          .join(User)


          .order_by(Tweet.created_date.desc()))
for tweet in tweets:


    print(tweet.user.username, tweet.message)

tweets = Tweet.select().order_by(Tweet.created_date.desc())
for tweet in tweets:


    # WARNING: an additional query will be issued for EACH tweet


    # to fetch the associated User data.


    print(tweet.user.username, tweet.message)

tweets = Tweet.select()
for tweet in tweets:


    # Instead of "tweet.user", we will just get the raw ID value stored


    # in the column.


    print(tweet.user_id, tweet.message)

class Tweet(Model):


    # ... same fields, except we declare the user FK to have


    # lazy-load disabled:


    user = ForeignKeyField(User, backref='tweets', lazy_load=False)
for tweet in Tweet.select():


    print(tweet.user, tweet.message)
# With lazy-load disabled, accessing tweet.user will not perform an extra
# query and the user ID value is returned instead.
# e.g.:
# 1  tweet from user1
# 1  another from user1
# 2  tweet from user2
# However, if we eagerly load the related user object, then the user
# foreign key will behave like usual:
for tweet in Tweet.select(Tweet, User).join(User):


    print(tweet.user.username, tweet.message)
# user1  tweet from user1
# user1  another from user1
# user2  tweet from user1

class Message(Model):


    from_user = ForeignKeyField(User, backref='outbox')


    to_user = ForeignKeyField(User, backref='inbox')


    text = TextField()
for message in some_user.outbox:


    # We are iterating over all Messages whose from_user is some_user.


    print(message)
for message in some_user.inbox:


    # We are iterating over all Messages whose to_user is some_user


    print(message)

# Get the current time.
now = datetime.datetime.now()
# Get days that have events for the current month.
Event.select(Event.event_date.day.alias('day')).where(


    (Event.event_date.year == now.year) &


    (Event.event_date.month == now.month))

class Post(Model):


    content = TextField()


    flags = BitField()


    is_favorite = flags.flag(1)


    is_sticky = flags.flag(2)


    is_minimized = flags.flag(4)


    is_deleted = flags.flag(8)

>>> p = Post()
>>> p.is_sticky = True
>>> p.is_minimized = True
>>> print(p.flags)  # Prints 4 | 2 --> "6"
6
>>> p.is_favorite
False
>>> p.is_sticky
True

# Generates a WHERE clause that looks like:
# WHERE (post.flags & 1 != 0)
favorites = Post.select().where(Post.is_favorite)
# Query for sticky + favorite posts:
sticky_faves = Post.select().where(Post.is_sticky & Post.is_favorite)

# Set the 4th bit on all Post objects.
Post.update(flags=Post.flags | 8).execute()
# Clear the 1st and 3rd bits on all Post objects.
Post.update(flags=Post.flags & ~(1 | 4)).execute()

# Set the "is_deleted" bit on all posts.
Post.update(flags=Post.is_deleted.set()).execute()
# Clear the "is_deleted" bit on all posts.
Post.update(flags=Post.is_deleted.clear()).execute()

class Bitmap(Model):


    data = BigBitField()
bitmap = Bitmap()
# Sets the ith bit, e.g. the 1st bit, the 11th bit, the 63rd, etc.
bits_to_set = (1, 11, 63, 31, 55, 48, 100, 99)
for bit_idx in bits_to_set:


    bitmap.data.set_bit(bit_idx)
# We can test whether a bit is set using "is_set":
assert bitmap.data.is_set(11)
assert not bitmap.data.is_set(12)
# We can clear a bit:
bitmap.data.clear_bit(11)
assert not bitmap.data.is_set(11)
# We can also "toggle" a bit. Recall that the 63rd bit was set earlier.
assert bitmap.data.toggle_bit(63) is False
assert bitmap.data.toggle_bit(63) is True
assert bitmap.data.is_set(63)

db = SqliteDatabase(':memory:')
class Junk(Model):


    anything = BareField()


    class Meta:


        database = db
# Store multiple data-types in the Junk.anything column:
Junk.create(anything='a string')
Junk.create(anything=12345)
Junk.create(anything=3.14159)

class UUIDField(Field):


    field_type = 'uuid'

import uuid
class UUIDField(Field):


    field_type = 'uuid'


    def db_value(self, value):


        return value.hex  # convert UUID to hex string.


    def python_value(self, value):


        return uuid.UUID(value) # convert hex string to UUID

class LogEntry(Model):


    event = TextField()


    create = TimestampField()  # Uh-oh.


    update = TimestampField()  # Uh-oh.

class LogEntry(Model):


    event = TextField()


    create_ = TimestampField(column_name='create')


    update_ = TimestampField(column_name='update')

# Connect to our database.
db.connect()
# Create the tables.
db.create_tables([User, Tweet])

from peewee import *
contacts_db = SqliteDatabase('contacts.db')
class Person(Model):


    name = CharField()


    class Meta:


        database = contacts_db

>>> Person.Meta
Traceback (most recent call last):


  File "<stdin>", line 1, in <module>
AttributeError: type object 'Person' has no attribute 'Meta'
>>> Person._meta
<peewee.ModelOptions object at 0x7f51a2f03790>

>>> Person._meta.fields
{'id': <peewee.AutoField object at 0x7f51a2e92750>,


 'name': <peewee.CharField object at 0x7f51a2f0a510>}
>>> Person._meta.primary_key
<peewee.AutoField object at 0x7f51a2e92750>
>>> Person._meta.database
<peewee.SqliteDatabase object at 0x7f519bff6dd0>

>>> db = SqliteDatabase(':memory:')
>>> class ModelOne(Model):
...     class Meta:
...         database = db
...         table_name = 'model_one_tbl'
...
>>> class ModelTwo(ModelOne):
...     pass
...
>>> ModelOne._meta.database is ModelTwo._meta.database
True
>>> ModelOne._meta.table_name == ModelTwo._meta.table_name
False

class BlogToTag(Model):


    """A simple "through" table for many-to-many relationship."""


    blog = ForeignKeyField(Blog)


    tag = ForeignKeyField(Tag)


    class Meta:


        primary_key = CompositeKey('blog', 'tag')
class NoPrimaryKey(Model):


    data = IntegerField()


    class Meta:


        primary_key = False

In the next major release (Peewee 4.0), legacy_table_names will have a default value of False.

class UserProfile(Model):


    class Meta:


        table_name = 'user_profile_tbl'

def make_table_name(model_class):


    model_name = model_class.__name__


    return model_name.lower() + '_tbl'
class BaseModel(Model):


    class Meta:


        table_function = make_table_name
class User(BaseModel):


    # table_name will be "user_tbl".
class UserProfile(BaseModel):


    # table_name will be "userprofile_tbl".

class User(Model):


    username = CharField(unique=True)


    email = CharField(index=True)

class Product(Model):


    name = CharField(unique=True)


    price = DecimalField(constraints=[Check('price < 10000')])


    created = DateTimeField(


        constraints=[SQL("DEFAULT (datetime('now'))")])

class Transaction(Model):


    from_acct = CharField()


    to_acct = CharField()


    amount = DecimalField()


    date = DateTimeField()


    class Meta:


        indexes = (


            # create a unique on from/to/date


            (('from_acct', 'to_acct', 'date'), True),


            # create a non-unique on from/to


            (('from_acct', 'to_acct'), False),


        )

class Article(Model):


    name = TextField()


    timestamp = TimestampField()


    status = IntegerField()


    flags = IntegerField()
# Add an index on "name" and "timestamp" columns.
Article.add_index(Article.name, Article.timestamp)
# Add a partial index on name and timestamp where status = 1.
Article.add_index(Article.name, Article.timestamp,


                  where=(Article.status == 1))
# Create a unique index on timestamp desc, status & 4.
idx = Article.index(


    Article.timestamp.desc(),


    Article.flags.bin_and(4),


    unique=True)
Article.add_index(idx)

# SQLite does not support parameterized CREATE INDEX queries, so
# we declare it manually.
Article.add_index(SQL('CREATE INDEX ...'))

See add_index() for details.

class Person(Model):


    first = CharField()


    last = CharField()


    class Meta:


        primary_key = CompositeKey('first', 'last')
class Pet(Model):


    owner_first = CharField()


    owner_last = CharField()


    pet_name = CharField()


    class Meta:


        constraints = [SQL('FOREIGN KEY(owner_first, owner_last) '


                           'REFERENCES person(first, last)')]

class Product(Model):


    name = CharField(unique=True)


    price = DecimalField()


    class Meta:


        constraints = [Check('price < 10000')]

class Event(Model):


    event_id = AutoField()  # Event.event_id will be auto-incrementing PK.


    name = CharField()


    timestamp = DateTimeField(default=datetime.datetime.now)


    metadata = BlobField()

class Person(Model):


    email = CharField(primary_key=True)


    name = TextField()


    dob = DateField()

class MyModel(Model):


    id = IntegerField(primary_key=True)

Peewee understands the above model declaration as a model with an integer primary key, but the value of that ID is determined by the application. To create an auto-incrementing integer primary key, you would instead write:

class MyModel(Model):


    id = AutoField()  # primary_key=True is implied.

class Image(Model):


    filename = TextField()


    mimetype = CharField()
class Tag(Model):


    label = CharField()
class ImageTag(Model):  # Many-to-many relationship.


    image = ForeignKeyField(Image)


    tag = ForeignKeyField(Tag)


    class Meta:


        primary_key = CompositeKey('image', 'tag')

from peewee import *
class UUIDModel(Model):


    id = UUIDField(primary_key=True)

# This works because .create() will specify `force_insert=True`.
obj1 = UUIDModel.create(id=uuid.uuid4())
# This will not work, however. Peewee will attempt to do an update:
obj2 = UUIDModel(id=uuid.uuid4())
obj2.save() # WRONG
obj2.save(force_insert=True) # CORRECT
# Once the object has been created, you can call save() normally.
obj2.save()

class BlogToTag(Model):


    """A simple "through" table for many-to-many relationship."""


    blog = ForeignKeyField(Blog)


    tag = ForeignKeyField(Tag)


    class Meta:


        primary_key = CompositeKey('blog', 'tag')

data = load_user_csv() # load up a bunch of data
User._meta.auto_increment = False # turn off auto incrementing IDs
with db.atomic():


    for row in data:


        u = User(id=row[0], username=row[1])


        u.save(force_insert=True) # <-- force peewee to insert row
User._meta.auto_increment = True

data = load_user_csv()
fields = [User.id, User.username]
with db.atomic():


    User.insert_many(data, fields=fields).execute()

class User(BaseModel):


    id = IntegerField(primary_key=True)


    username = CharField()
>>> u = User.create(id=999, username='somebody')
>>> u.id
999
>>> User.get(User.username == 'somebody').id
999

class MyData(BaseModel):


    timestamp = DateTimeField()


    value = IntegerField()


    class Meta:


        primary_key = False

CREATE TABLE "mydata" (


  "timestamp" DATETIME NOT NULL,


  "value" INTEGER NOT NULL
)

class Category(Model):


    name = CharField()


    parent = ForeignKeyField('self', null=True, backref='children')

Parent = Category.alias()
GrandParent = Category.alias()
query = (Category


         .select(Category, Parent)


         .join(Parent, on=(Category.parent == Parent.id))


         .join(GrandParent, on=(Parent.parent == GrandParent.id))


         .where(GrandParent.name == 'some category')


         .order_by(Category.name))

class User(Model):


    username = CharField()


    favorite_tweet = ForeignKeyField(Tweet, null=True)  # NameError!!
class Tweet(Model):


    message = TextField()


    user = ForeignKeyField(User, backref='tweets')

class User(Model):


    username = CharField()


    favorite_tweet_id = IntegerField(null=True)

class User(Model):


    username = CharField()


    # Tweet has not been defined yet so use the deferred reference.


    favorite_tweet = DeferredForeignKey('Tweet', null=True)
class Tweet(Model):


    message = TextField()


    user = ForeignKeyField(User, backref='tweets')
# Now that Tweet is defined, "favorite_tweet" has been converted into
# a ForeignKeyField.
print(User.favorite_tweet)
# <ForeignKeyField: "user"."favorite_tweet">

# Will create the User and Tweet tables, but does *not* create a
# foreign-key constraint on User.favorite_tweet.
db.create_tables([User, Tweet])
# Create the foreign-key constraint:
User._schema.create_foreign_key(User.favorite_tweet)

>>> User.create(username='Charlie')
<__main__.User object at 0x2529350>

>>> user = User(username='Charlie')
>>> user.save()  # save() returns the number of rows modified.
1
>>> user.id
1
>>> huey = User()
>>> huey.username = 'Huey'
>>> huey.save()
1
>>> huey.id
2

>>> tweet = Tweet.create(user=huey, message='Hello!')

>>> tweet = Tweet.create(user=2, message='Hello again!')

>>> User.insert(username='Mickey').execute()
3

data_source = [


    {'field1': 'val1-1', 'field2': 'val1-2'},


    {'field1': 'val2-1', 'field2': 'val2-2'},


    # ...
]
for data_dict in data_source:


    MyModel.create(**data_dict)

# This is much faster.
with db.atomic():


    for data_dict in data_source:


        MyModel.create(**data_dict)

data_source = [


    {'field1': 'val1-1', 'field2': 'val1-2'},


    {'field1': 'val2-1', 'field2': 'val2-2'},


    # ...
]
# Fastest way to INSERT multiple rows.
MyModel.insert_many(data_source).execute()

# We can INSERT tuples as well...
data = [('val1-1', 'val1-2'),


        ('val2-1', 'val2-2'),


        ('val3-1', 'val3-2')]
# But we need to indicate which fields the values correspond to.
MyModel.insert_many(data, fields=[MyModel.field1, MyModel.field2]).execute()

# You can, of course, wrap this in a transaction as well:
with db.atomic():


    MyModel.insert_many(data, fields=fields).execute()

# Insert rows 100 at a time.
with db.atomic():


    for idx in range(0, len(data_source), 100):


        MyModel.insert_many(data_source[idx:idx+100]).execute()

from peewee import chunked
# Insert rows 100 at a time.
with db.atomic():


    for batch in chunked(data_source, 100):


        MyModel.insert_many(batch).execute()

# Read list of usernames from a file, for example.
with open('user_list.txt') as fh:


    # Create a list of unsaved User instances.


    users = [User(username=line.strip()) for line in fh.readlines()]
# Wrap the operation in a transaction and batch INSERT the users
# 100 at a time.
with db.atomic():


    User.bulk_create(users, batch_size=100)

# First, create 3 users with usernames u1, u2, u3.
u1, u2, u3 = [User.create(username='u%s' % i) for i in (1, 2, 3)]
# Now we'll modify the user instances.
u1.username = 'u1-x'
u2.username = 'u2-y'
u3.username = 'u3-z'
# Update all three users with a single UPDATE query.
User.bulk_update([u1, u2, u3], fields=[User.username])

# List of row data to insert.
row_data = [{'username': 'u1'}, {'username': 'u2'}, ...]
# Assume there are 789 items in row_data. The following code will result in
# 8 total transactions (7x100 rows + 1x89 rows).
for row in db.batch_commit(row_data, 100):


    User.create(**row)

res = (TweetArchive


       .insert_from(


           Tweet.select(Tweet.user, Tweet.message),


           fields=[TweetArchive.user, TweetArchive.message])


       .execute())

INSERT INTO "tweet_archive" ("user_id", "message")
SELECT "user_id", "message" FROM "tweet";

>>> user.save()  # save() returns the number of rows modified.
1
>>> user.id
1
>>> user.save()
>>> user.id
1
>>> huey.save()
1
>>> huey.id
2

>>> today = datetime.today()
>>> query = Tweet.update(is_published=True).where(Tweet.creation_date < today)
>>> query.execute()  # Returns the number of rows that were updated.
4

>>> for stat in Stat.select().where(Stat.url == request.url):
...     stat.counter += 1
...     stat.save()

>>> query = Stat.update(counter=Stat.counter + 1).where(Stat.url == request.url)
>>> query.execute()

>>> query = Employee.update(bonus=(Employee.bonus + (Employee.salary * .1)))
>>> query.execute()  # Give everyone a bonus!

>>> subquery = Tweet.select(fn.COUNT(Tweet.id)).where(Tweet.user == User.id)
>>> update = User.update(num_tweets=subquery)
>>> update.execute()

class User(Model):


    username = TextField(unique=True)


    last_login = DateTimeField(null=True)
# Insert or update the user. The "last_login" value will be updated
# regardless of whether the user existed previously.
user_id = (User


           .replace(username='the-user', last_login=datetime.now())


           .execute())
# This query is equivalent:
user_id = (User


           .insert(username='the-user', last_login=datetime.now())


           .on_conflict_replace()


           .execute())

class User(Model):


    username = TextField(unique=True)


    last_login = DateTimeField(null=True)


    login_count = IntegerField()
# Insert a new user.
User.create(username='huey', login_count=0)
# Simulate the user logging in. The login count and timestamp will be
# either created or updated correctly.
now = datetime.now()
rowid = (User


         .insert(username='huey', last_login=now, login_count=1)


         .on_conflict(


             preserve=[User.last_login],  # Use the value we would have inserted.


             update={User.login_count: User.login_count + 1})


         .execute())

class User(Model):


    username = TextField(unique=True)


    last_login = DateTimeField(null=True)


    login_count = IntegerField()
# Insert a new user.
User.create(username='huey', login_count=0)
# Simulate the user logging in. The login count and timestamp will be
# either created or updated correctly.
now = datetime.now()
rowid = (User


         .insert(username='huey', last_login=now, login_count=1)


         .on_conflict(


             conflict_target=[User.username],  # Which constraint?


             preserve=[User.last_login],  # Use the value we would have inserted.


             update={User.login_count: User.login_count + 1})


         .execute())

class KV(Model):


    key = CharField(unique=True)


    value = IntegerField()
# Create one row.
KV.create(key='k1', value=1)
# Demonstrate usage of EXCLUDED.
# Here we will attempt to insert a new value for a given key. If that
# key already exists, then we will update its value with the *sum* of its
# original value and the value we attempted to insert -- provided that
# the new value is larger than the original value.
query = (KV.insert(key='k1', value=10)


         .on_conflict(conflict_target=[KV.key],


                      update={KV.value: KV.value + EXCLUDED.value},


                      where=(EXCLUDED.value > KV.value)))
# Executing the above query will result in the following data being
# present in the "kv" table:
# (key='k1', value=11)
query.execute()
# If we attempted to execute the query *again*, then nothing would be
# updated, as the new value (10) is now less than the value in the
# original row (11).

>>> user = User.get(User.id == 1)
>>> user.delete_instance()  # Returns the number of rows deleted.
1
>>> User.get(User.id == 1)
UserDoesNotExist: instance matching query does not exist:
SQL: SELECT t1."id", t1."username" FROM "user" AS t1 WHERE t1."id" = ?
PARAMS: [1]

>>> query = Tweet.delete().where(Tweet.creation_date < one_year_ago)
>>> query.execute()  # Returns the number of rows deleted.
7

>>> User.get(User.id == 1)
<__main__.User object at 0x25294d0>
>>> User.get_by_id(1)  # Same as above.
<__main__.User object at 0x252df10>
>>> User[1]  # Also same as above.
<__main__.User object at 0x252dd10>
>>> User.get(User.id == 1).username
u'Charlie'
>>> User.get(User.username == 'Charlie')
<__main__.User object at 0x2529410>
>>> User.get(User.username == 'nobody')
UserDoesNotExist: instance matching query does not exist:
SQL: SELECT t1."id", t1."username" FROM "user" AS t1 WHERE t1."username" = ?
PARAMS: ['nobody']

>>> (Tweet
...  .select()
...  .join(User)
...  .where(User.username == 'charlie')
...  .order_by(Tweet.created_date.desc())
...  .get())
<__main__.Tweet object at 0x2623410>

try:


    with db.atomic():


        return User.create(username=username)
except peewee.IntegrityError:


    # `username` is a unique column, so this username already exists,


    # making it safe to call .get().


    return User.get(User.username == username)

user, created = User.get_or_create(username=username)

person, created = Person.get_or_create(


    first_name=first_name,


    last_name=last_name,


    defaults={'dob': dob, 'favorite_color': 'green'})

>>> query = User.select()
>>> [user.username for user in query]
['Charlie', 'Huey', 'Peewee']
>>> query[1]
<__main__.User at 0x7f83e80f5550>
>>> query[1].username
'Huey'
>>> query[:2]
[<__main__.User at 0x7f83e80f53a8>, <__main__.User at 0x7f83e80f5550>]

>>> for user in User.select():
...     print user.username
...
Charlie
Huey
Peewee

>>> tweet = Tweet.get()
>>> tweet.user  # Accessing a foreign key returns the related model.
<tw.User at 0x7f3ceb017f50>
>>> user = User.get()
>>> user.tweets  # Accessing a back-reference returns a query.
<peewee.ModelSelect at 0x7f73db3bafd0>

>>> for tweet in user.tweets:
...     print(tweet.message)
...
hello world
this is fun
look at this picture of my food

>>> query = User.select().dicts()
>>> for row in query:
...     print(row)
{'id': 1, 'username': 'Charlie'}
{'id': 2, 'username': 'Huey'}
{'id': 3, 'username': 'Peewee'}

# Let's assume we've got 10 million stat objects to dump to a csv file.
stats = Stat.select()
# Our imaginary serializer class
serializer = CSVSerializer()
# Loop over all the stats and serialize.
for stat in stats.iterator():


    serializer.serialize_object(stat)

# Let's assume we've got 10 million stat objects to dump to a csv file.
stats = Stat.select()
# Our imaginary serializer class
serializer = CSVSerializer()
# Loop over all the stats (rendered as tuples, without caching) and serialize.
for stat_tuple in stats.tuples().iterator():


    serializer.serialize_tuple(stat_tuple)

query = (Tweet


         .select(Tweet, User)  # Select tweet and user data.


         .join(User))
# Note that the user columns are stored in a separate User instance
# accessible at tweet.user:
for tweet in query:


    print(tweet.user.username, tweet.content)
# Using ".objects()" will not create the tweet.user object and assigns all
# user attributes to the tweet instance:
for tweet in query.objects():


    print(tweet.username, tweet.content)

query = Tweet.select(Tweet.content, User.username).join(User)
cursor = database.execute(query)
for (content, username) in cursor:


    print(username, '->', content)

>>> user = User.get(User.username == 'Charlie')
>>> for tweet in Tweet.select().where(Tweet.user == user, Tweet.is_published == True):
...     print(tweet.user.username, '->', tweet.message)
...
Charlie -> hello world
Charlie -> this is fun
>>> for tweet in Tweet.select().where(Tweet.created_date < datetime.datetime(2011, 1, 1)):
...     print(tweet.message, tweet.created_date)
...
Really old tweet 2010-01-01 00:00:00

>>> for tweet in Tweet.select().join(User).where(User.username == 'Charlie'):
...     print(tweet.message)
hello world
this is fun
look at this picture of my food

>>> Tweet.select().join(User).where(
...     (User.username == 'Charlie') |
...     (User.username == 'Peewee Herman'))

You can also nest queries, for example tweets by users whose username starts with "a":

# get users whose username starts with "a"
a_users = User.select().where(fn.Lower(fn.Substr(User.username, 1, 1)) == 'a')
# the ".in_()" method signifies an "IN" query
a_user_tweets = Tweet.select().where(Tweet.user.in_(a_users))

User.select().where(User.active == True)

User.select().where(


    (User.is_staff == True) | (User.is_superuser == True))

Tweet.select().join(User).where(User.username == 'charlie')

Tweet.select().join(User).where(


    (User.is_staff == True) | (User.is_superuser == True))

staff_super = User.select(User.id).where(


    (User.is_staff == True) | (User.is_superuser == True))
Tweet.select().where(Tweet.user.in_(staff_super))

>>> for t in Tweet.select().order_by(Tweet.created_date):
...     print(t.pub_date)
...
2010-01-01 00:00:00
2011-06-07 14:08:48
2011-06-07 14:12:57
>>> for t in Tweet.select().order_by(Tweet.created_date.desc()):
...     print(t.pub_date)
...
2011-06-07 14:12:57
2011-06-07 14:08:48
2010-01-01 00:00:00

# The following queries are equivalent:
Tweet.select().order_by(Tweet.created_date.desc())
Tweet.select().order_by(-Tweet.created_date)  # Note the "-" prefix.
# Similarly you can use "+" to indicate ascending order, though ascending
# is the default when no ordering is otherwise specified.
User.select().order_by(+User.username)

query = (Tweet


         .select()


         .join(User)


         .order_by(User.username, Tweet.created_date.desc()))

SELECT t1."id", t1."user_id", t1."message", t1."is_published", t1."created_date"
FROM "tweet" AS t1
INNER JOIN "user" AS t2


  ON t1."user_id" = t2."id"
ORDER BY t2."username", t1."created_date" DESC

# Let's start with our base query. We want to get all usernames and the number of
# tweets they've made. We wish to sort this list from users with most tweets to
# users with fewest tweets.
query = (User


         .select(User.username, fn.COUNT(Tweet.id).alias('num_tweets'))


         .join(Tweet, JOIN.LEFT_OUTER)


         .group_by(User.username))

query = (User


         .select(User.username, fn.COUNT(Tweet.id).alias('num_tweets'))


         .join(Tweet, JOIN.LEFT_OUTER)


         .group_by(User.username)


         .order_by(fn.COUNT(Tweet.id).desc()))

query = (User


         .select(User.username, fn.COUNT(Tweet.id).alias('num_tweets'))


         .join(Tweet, JOIN.LEFT_OUTER)


         .group_by(User.username)


         .order_by(SQL('num_tweets').desc()))

ntweets = fn.COUNT(Tweet.id)
query = (User


         .select(User.username, ntweets.alias('num_tweets'))


         .join(Tweet, JOIN.LEFT_OUTER)


         .group_by(User.username)


         .order_by(ntweets.desc())

# Pick 5 lucky winners:
LotteryNumber.select().order_by(fn.Random()).limit(5)

# Pick 5 lucky winners:
LotteryNumber.select().order_by(fn.Rand()).limit(5)

>>> for tweet in Tweet.select().order_by(Tweet.id).paginate(2, 10):
...     print(tweet.message)
...
tweet 10
tweet 11
tweet 12
tweet 13
tweet 14
tweet 15
tweet 16
tweet 17
tweet 18
tweet 19

>>> Tweet.select().count()
100
>>> Tweet.select().where(Tweet.id > 50).count()
50

SELECT COUNT(1) FROM ( ... your query ... );

query = (User


         .select(User, fn.Count(Tweet.id).alias('count'))


         .join(Tweet, JOIN.LEFT_OUTER)


         .group_by(User))

class Photo(Model):


    image = CharField()
class Tag(Model):


    name = CharField()
class PhotoTag(Model):


    photo = ForeignKeyField(Photo)


    tag = ForeignKeyField(Tag)

query = (Tag


         .select()


         .join(PhotoTag)


         .join(Photo)


         .group_by(Tag)


         .having(fn.Count(Photo.id) > 5))

SELECT t1."id", t1."name"
FROM "tag" AS t1
INNER JOIN "phototag" AS t2 ON t1."id" = t2."tag_id"
INNER JOIN "photo" AS t3 ON t2."photo_id" = t3."id"
GROUP BY t1."id", t1."name"
HAVING Count(t3."id") > 5

query = (Tag


         .select(Tag, fn.Count(Photo.id).alias('count'))


         .join(PhotoTag)


         .join(Photo)


         .group_by(Tag)


         .having(fn.Count(Photo.id) > 5))

>>> PageView.select(fn.Count(fn.Distinct(PageView.url))).scalar()
100

>>> Employee.select(
...     fn.Min(Employee.salary), fn.Max(Employee.salary)
... ).scalar(as_tuple=True)
(30000, 50000)

class Sample(Model):


    counter = IntegerField()


    value = FloatField()
data = [(1, 10),


        (1, 20),


        (2, 1),


        (2, 3),


        (3, 100)]
Sample.insert_many(data, fields=[Sample.counter, Sample.value]).execute()

query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.SUM(Sample.value).over(order_by=[Sample.id]).alias('total'))
for sample in query:


    print(sample.counter, sample.value, sample.total)
# 1    10.    10.
# 1    20.    30.
# 2     1.    31.
# 2     3.    34.
# 3   100    134.

difference = Sample.value - fn.LAG(Sample.value, 1).over(order_by=[Sample.id])
query = Sample.select(


    Sample.counter,


    Sample.value,


    difference.alias('diff'))
for sample in query:


    print(sample.counter, sample.value, sample.diff)
# 1    10.   NULL
# 1    20.    10.  -- (20 - 10)
# 2     1.   -19.  -- (1 - 20)
# 2     3.     2.  -- (3 - 1)
# 3   100     97.  -- (100 - 3)

query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.AVG(Sample.value).over(partition_by=[Sample.counter]).alias('cavg'))
for sample in query:


    print(sample.counter, sample.value, sample.cavg)
# 1    10.    15.
# 1    20.    15.
# 2     1.     2.
# 2     3.     2.
# 3   100    100.

query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.RANK().over(


        order_by=[Sample.value],


        partition_by=[Sample.counter]).alias('rank'))
for sample in query:


    print(sample.counter, sample.value, sample.rank)
# 1    10.    1
# 1    20.    2
# 2     1.    1
# 2     3.    2
# 3   100     1

query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.SUM(Sample.value).over(


        order_by=[Sample.id],


        start=Window.preceding(2),


        end=Window.CURRENT_ROW).alias('rsum'))
for sample in query:


    print(sample.counter, sample.value, sample.rsum)
# 1    10.    10.
# 1    20.    30.  -- (20 + 10)
# 2     1.    31.  -- (1 + 20 + 10)
# 2     3.    24.  -- (3 + 1 + 20)
# 3   100    104.  -- (100 + 3 + 1)

query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.SUM(Sample.value).over(


        order_by=[Sample.id],


        start=Window.CURRENT_ROW,


        end=Window.following()).alias('rsum'))
# 1    10.   134.  -- (10 + 20 + 1 + 3 + 100)
# 1    20.   124.  -- (20 + 1 + 3 + 100)
# 2     1.   104.  -- (1 + 3 + 100)
# 2     3.   103.  -- (3 + 100)
# 3   100    100.  -- (100)

query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.SUM(Sample.value).filter(Sample.counter != 2).over(


        order_by=[Sample.id]).alias('csum'))
for sample in query:


    print(sample.counter, sample.value, sample.csum)
# 1    10.    10.
# 1    20.    30.
# 2     1.    30.
# 2     3.    30.
# 3   100    130.

win = Window(order_by=[Sample.id])
query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.LEAD(Sample.value).over(win),


    fn.LAG(Sample.value).over(win),


    fn.SUM(Sample.value).over(win)
).window(win)  # Include our window definition in query.
for row in query.tuples():


    print(row)
# counter  value  lead()  lag()  sum()
# 1          10.     20.   NULL    10.
# 1          20.      1.    10.    30.
# 2           1.      3.    20.    31.
# 2           3.    100.     1.    34.
# 3         100.    NULL     3.   134.

w1 = Window(order_by=[Sample.id]).alias('w1')
w2 = Window(partition_by=[Sample.counter]).alias('w2')
query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.SUM(Sample.value).over(w1).alias('rsum'),  # Running total.


    fn.AVG(Sample.value).over(w2).alias('cavg')   # Avg per category.
).window(w1, w2)  # Include our window definitions.
for sample in query:


    print(sample.counter, sample.value, sample.rsum, sample.cavg)
# counter  value   rsum     cavg
# 1          10.     10.     15.
# 1          20.     30.     15.
# 2           1.     31.      2.
# 2           3.     34.      2.
# 3         100     134.    100.

w1 = Window(partition_by=[Sample.counter]).alias('w1')
# By extending w1, this window definition will also be partitioned
# by "counter".
w2 = Window(extends=w1, order_by=[Sample.value.desc()]).alias('w2')
query = (Sample


         .select(Sample.counter, Sample.value,


                 fn.SUM(Sample.value).over(w1).alias('group_sum'),


                 fn.RANK().over(w2).alias('revrank'))


         .window(w1, w2)


         .order_by(Sample.id))
for sample in query:


    print(sample.counter, sample.value, sample.group_sum, sample.revrank)
# counter  value   group_sum   revrank
# 1        10.     30.         2
# 1        20.     30.         1
# 2        1.      4.          2
# 2        3.      4.          1
# 3        100.    100.        1

>>> Sample.create(counter=1, value=20.)
<Sample 6>
>>> Sample.create(counter=2, value=1.)
<Sample 7>

query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.SUM(Sample.value).over(


        order_by=[Sample.counter, Sample.value],


        frame_type=Window.RANGE).alias('rsum'))
for sample in query.order_by(Sample.counter, Sample.value):


    print(sample.counter, sample.value, sample.rsum)
# counter  value   rsum
# 1          10.     10.
# 1          20.     50.
# 1          20.     50.
# 2           1.     52.
# 2           1.     52.
# 2           3.     55.
# 3         100     155.

query = Sample.select(


    Sample.counter,


    Sample.value,


    fn.SUM(Sample.value).over(


        order_by=[Sample.counter, Sample.value],


        frame_type=Window.ROWS).alias('rsum'))
for sample in query.order_by(Sample.counter, Sample.value):


    print(sample.counter, sample.value, sample.rsum)
# counter  value   rsum
# 1          10.     10.
# 1          20.     30.
# 1          20.     50.
# 2           1.     51.
# 2           1.     52.
# 2           3.     55.
# 3         100     155.

query = (Sample


         .select(Sample.counter, Sample.value,


                 fn.SUM(Sample.value).over(


                    order_by=[Sample.counter, Sample.value],


                    frame_type=Window.GROUPS,


                    start=Window.preceding(1)).alias('gsum'))


         .order_by(Sample.counter, Sample.value))
for sample in query:


    print(sample.counter, sample.value, sample.gsum)
#  counter   value    gsum
#  1         10       10
#  1         20       50
#  1         20       50   (10) + (20+0)
#  2         1        42
#  2         1        42   (20+20) + (1+1)
#  2         3        5    (1+1) + 3
#  3         100      103  (3) + 100

For general information on window functions, read the postgres window functions tutorial

Additionally, the postgres docs and the sqlite docs contain a lot of good information.

stats = (Stat


         .select(Stat.url, fn.Count(Stat.url))


         .group_by(Stat.url)


         .tuples())
# iterate over a list of 2-tuples containing the url and count
for stat_url, stat_count in stats:


    print(stat_url, stat_count)

stats = (Stat


         .select(Stat.url, fn.Count(Stat.url).alias('ct'))


         .group_by(Stat.url)


         .dicts())
# iterate over a list of 2-tuples containing the url and count
for stat in stats:


    print(stat['url'], stat['ct'])

query = (User


         .update(is_active=False)


         .where(User.registration_expired == True)


         .returning(User))
# Send an email to every user that was deactivated.
for deactivate_user in query.execute():


    send_deactivation_email(deactivated_user.email)

query = (User


         .insert(email='foo@bar.com', created=fn.now())


         .returning(User))  # Shorthand for all columns on User.
# When using RETURNING, execute() returns a cursor.
cursor = query.execute()
# Get the user object we just inserted and log the data:
user = cursor[0]
logger.info('Created user %s (id=%s) at %s', user.email, user.id, user.created)

data = [{'name': 'charlie'}, {'name': 'huey'}, {'name': 'mickey'}]
query = (User


         .insert_many(data)


         .returning(User.id, User.username)


         .dicts())
for new_user in query.execute():


    print('Added user "%s", id=%s' % (new_user['username'], new_user['id']))

class Sample(Model):


    key = TextField()


    value = FloatField()
data = (


    ('a', (1.25, 1.5, 1.75)),


    ('b', (2.1, 2.3, 2.5, 2.7, 2.9)),


    ('c', (3.5, 3.5)))
# Populate data.
for key, values in data:


    Sample.insert_many([(key, value) for value in values],


                       fields=[Sample.key, Sample.value]).execute()

# First we'll declare the query that will be used as a CTE. This query
# simply determines the average value for each key.
cte = (Sample


       .select(Sample.key, fn.AVG(Sample.value).alias('avg_value'))


       .group_by(Sample.key)


       .cte('key_avgs', columns=('key', 'avg_value')))
# Now we'll query the sample table, using our CTE to find rows whose value
# exceeds the average for the given key. We'll calculate how far above the
# average the given sample's value is, as well.
query = (Sample


         .select(Sample.key, Sample.value)


         .join(cte, on=(Sample.key == cte.c.key))


         .where(Sample.value > cte.c.avg_value)


         .order_by(Sample.value)


         .with_cte(cte))

>>> for sample in query:
...     print(sample.key, sample.value)
# 'a', 1.75
# 'b', 2.7
# 'b', 2.9

class Order(Model):


    region = TextField()


    amount = FloatField()


    product = TextField()


    quantity = IntegerField()

WITH regional_sales AS (


    SELECT region, SUM(amount) AS total_sales


    FROM orders


    GROUP BY region


  ), top_regions AS (


    SELECT region


    FROM regional_sales


    WHERE total_sales > (SELECT SUM(total_sales) / 10 FROM regional_sales)


  )
SELECT region,


       product,


       SUM(quantity) AS product_units,


       SUM(amount) AS product_sales
FROM orders
WHERE region IN (SELECT region FROM top_regions)
GROUP BY region, product;

reg_sales = (Order


             .select(Order.region,


                     fn.SUM(Order.amount).alias('total_sales'))


             .group_by(Order.region)


             .cte('regional_sales'))
top_regions = (reg_sales


               .select(reg_sales.c.region)


               .where(reg_sales.c.total_sales > (


                   reg_sales.select(fn.SUM(reg_sales.c.total_sales) / 10)))


               .cte('top_regions'))
query = (Order


         .select(Order.region,


                 Order.product,


                 fn.SUM(Order.quantity).alias('product_units'),


                 fn.SUM(Order.amount).alias('product_sales'))


         .where(Order.region.in_(top_regions.select(top_regions.c.region)))


         .group_by(Order.region, Order.product)


         .with_cte(regional_sales, top_regions))

class Category(Model):


    name = TextField()


    parent = ForeignKeyField('self', backref='children', null=True)

# Define the base case of our recursive CTE. This will be categories that
# have a null parent foreign-key.
Base = Category.alias()
level = Value(1).alias('level')
path = Base.name.alias('path')
base_case = (Base


             .select(Base.id, Base.name, Base.parent, level, path)


             .where(Base.parent.is_null())


             .cte('base', recursive=True))
# Define the recursive terms.
RTerm = Category.alias()
rlevel = (base_case.c.level + 1).alias('level')
rpath = base_case.c.path.concat('->').concat(RTerm.name).alias('path')
recursive = (RTerm


             .select(RTerm.id, RTerm.name, RTerm.parent, rlevel, rpath)


             .join(base_case, on=(RTerm.parent == base_case.c.id)))
# The recursive CTE is created by taking the base case and UNION ALL with
# the recursive term.
cte = base_case.union_all(recursive)
# We will now query from the CTE to get the categories, their levels,  and
# their paths.
query = (cte


         .select_from(cte.c.name, cte.c.level, cte.c.path)


         .order_by(cte.c.path))
# We can now iterate over a list of all categories and print their names,
# absolute levels, and path from root -> category.
for category in query:


    print(category.name, category.level, category.path)
# Example output:
# root, 1, root
# p1, 2, root->p1
# c1-1, 3, root->p1->c1-1
# c1-2, 3, root->p1->c1-2
# p2, 2, root->p2
# c2-1, 3, root->p2->c2-1

# Find the user whose username is "charlie".
User.select().where(User.username == 'charlie')
# Find the users whose username is in [charlie, huey, mickey]
User.select().where(User.username.in_(['charlie', 'huey', 'mickey']))
Employee.select().where(Employee.salary.between(50000, 60000))
Employee.select().where(Employee.name.startswith('C'))
Blog.select().where(Blog.title.contains(search_string))

# Find any users who are active administrations.
User.select().where(


  (User.is_admin == True) &


  (User.is_active == True))
# Find any users who are either administrators or super-users.
User.select().where(


  (User.is_admin == True) |


  (User.is_superuser == True))
# Find any Tweets by users who are not admins (NOT IN).
admins = User.select().where(User.is_admin == True)
non_admin_tweets = Tweet.select().where(Tweet.user.not_in(admins))
# Find any users who are not my friends (strangers).
friends = User.select().where(User.username.in_(['charlie', 'huey', 'mickey']))
strangers = User.select().where(User.id.not_in(friends))

So just remember:

Because SQLite's LIKE operation is case-insensitive by default, peewee will use the SQLite GLOB operation for case-sensitive searches. The glob operation uses asterisks for wildcards as opposed to the usual percent-sign. If you are using SQLite and want case-sensitive partial string matching, remember to use asterisks for the wildcard.

# Get all User objects whose last login is NULL.
User.select().where(User.last_login >> None)
# Get users whose username is in the given list.
usernames = ['charlie', 'huey', 'mickey']
User.select().where(User.username << usernames)

# Get all User objects whose last login is NULL.
User.select().where(User.last_login.is_null(True))
# Get users whose username is in the given list.
usernames = ['charlie', 'huey', 'mickey']
User.select().where(User.username.in_(usernames))

# Get all User objects whose last login is *NOT* NULL.
User.select().where(User.last_login.is_null(False))
# Using unary negation instead.
User.select().where(~(User.last_login >> None))
# Get users whose username is *NOT* in the given list.
usernames = ['charlie', 'huey', 'mickey']
User.select().where(User.username.not_in(usernames))
# Using unary negation instead.
usernames = ['charlie', 'huey', 'mickey']
User.select().where(~(User.username << usernames))

from peewee import *
from peewee import Expression # the building block for expressions
def mod(lhs, rhs):


    return Expression(lhs, '%', rhs)

# Users with even ids.
User.select().where(mod(User.id, 2) == 0)

class User(Model):


    username = CharField()


    is_admin = BooleanField()


    is_active = BooleanField()


    last_login = DateTimeField()


    login_count = IntegerField()


    failed_logins = IntegerField()

# username is equal to 'charlie'
User.username == 'charlie'
# user has logged in less than 5 times
User.login_count < 5

# User is both and admin and has logged in today
(User.is_admin == True) & (User.last_login >= today)
# User's username is either charlie or charles
(User.username == 'charlie') | (User.username == 'charles')

# user's username starts with a 'g' or a 'G':
fn.Lower(fn.Substr(User.username, 1, 1)) == 'g'

# users who entered the incorrect more than half the time and have logged
# in at least 10 times
(User.failed_logins > (User.login_count * .5)) & (User.login_count > 10)

# when a user logs in we want to increment their login count:
User.update(login_count=User.login_count + 1).where(User.id == user_id)

# If for some reason your schema stores dates in separate columns ("year",
# "month" and "day"), you can use row-values to find all rows that happened
# in a given month:
Tuple(Event.year, Event.month) == (2019, 1)

class EventLog(Model):


    event_type = TextField()


    source = TextField()


    data = TextField()


    timestamp = TimestampField()
class IncidentLog(Model):


    incident_type = TextField()


    source = TextField()


    traceback = TextField()


    timestamp = TimestampField()
# Get a list of all the incident types and sources that have occurred today.
incidents = (IncidentLog


             .select(IncidentLog.incident_type, IncidentLog.source)


             .where(IncidentLog.timestamp >= datetime.date.today()))
# Find all events that correlate with the type and source of the
# incidents that occurred today.
events = (EventLog


          .select()


          .where(Tuple(EventLog.event_type, EventLog.source).in_(incidents))


          .order_by(EventLog.timestamp))

# Get all users and the number of tweets they've authored. Sort the
# results from most tweets -> fewest tweets.
query = (User


         .select(User, fn.COUNT(Tweet.id).alias('tweet_count'))


         .join(Tweet, JOIN.LEFT_OUTER)


         .group_by(User)


         .order_by(fn.COUNT(Tweet.id).desc()))
for user in query:


    print('%s -- %s tweets' % (user.username, user.tweet_count))

# Select the user's id, username and the first letter of their username, lower-cased
first_letter = fn.LOWER(fn.SUBSTR(User.username, 1, 1))
query = User.select(User, first_letter.alias('first_letter'))
# Alternatively we could select only users whose username begins with 'a'
a_users = User.select().where(first_letter == 'a')
>>> for user in a_users:
...    print(user.username)

# We'll query the user table and annotate it with a count of tweets for
# the given user
query = (User


         .select(User, fn.Count(Tweet.id).alias('ct'))


         .join(Tweet)


         .group_by(User))
# Now we will order by the count, which was aliased to "ct"
query = query.order_by(SQL('ct'))
# You could, of course, also write this as:
query = query.order_by(fn.COUNT(Tweet.id))

# Bad! DO NOT DO THIS!
query = MyModel.raw('SELECT * FROM my_table WHERE data = %s' % (user_data,))
# Good. `user_data` will be treated as a parameter to the query.
query = MyModel.raw('SELECT * FROM my_table WHERE data = %s', user_data)
# Bad! DO NOT DO THIS!
query = MyModel.select().where(SQL('Some SQL expression %s' % user_data))
# Good. `user_data` will be treated as a parameter.
query = MyModel.select().where(SQL('Some SQL expression %s', user_data))

import datetime
from peewee import *
db = SqliteDatabase(':memory:')
class BaseModel(Model):


    class Meta:


        database = db
class User(BaseModel):


    username = TextField()
class Tweet(BaseModel):


    content = TextField()


    timestamp = DateTimeField(default=datetime.datetime.now)


    user = ForeignKeyField(User, backref='tweets')
class Favorite(BaseModel):


    user = ForeignKeyField(User, backref='favorites')


    tweet = ForeignKeyField(Tweet, backref='favorites')

def populate_test_data():


    db.create_tables([User, Tweet, Favorite])


    data = (


        ('huey', ('meow', 'hiss', 'purr')),


        ('mickey', ('woof', 'whine')),


        ('zaizee', ()))


    for username, tweets in data:


        user = User.create(username=username)


        for tweet in tweets:


            Tweet.create(user=user, content=tweet)


    # Populate a few favorites for our users, such that:


    favorite_data = (


        ('huey', ['whine']),


        ('mickey', ['purr']),


        ('zaizee', ['meow', 'purr']))


    for username, favorites in favorite_data:


        user = User.get(User.username == username)


        for content in favorites:


            tweet = Tweet.get(Tweet.content == content)


            Favorite.create(user=user, tweet=tweet)

>>> query = Tweet.select().join(User).where(User.username == 'huey')
>>> for tweet in query:
...     print(tweet.content)
...
meow
hiss
purr

The following code is equivalent, but more explicit:

query = (Tweet


         .select()


         .join(User, on=(Tweet.user == User.id))


         .where(User.username == 'huey'))

>>> huey = User.get(User.username == 'huey')
>>> for tweet in huey.tweets:
...     print(tweet.content)
...
meow
hiss
purr

>>> huey.tweets
<peewee.ModelSelect at 0x7f0483931fd0>
>>> huey.tweets.sql()
('SELECT "t1"."id", "t1"."content", "t1"."timestamp", "t1"."user_id"


  FROM "tweet" AS "t1" WHERE ("t1"."user_id" = ?)', [1])

SELECT user.username, COUNT(favorite.id)
FROM user
LEFT OUTER JOIN tweet ON tweet.user_id = user.id
LEFT OUTER JOIN favorite ON favorite.tweet_id = tweet.id
GROUP BY user.username

query = (User


         .select(User.username, fn.COUNT(Favorite.id).alias('count'))


         .join(Tweet, JOIN.LEFT_OUTER)  # Joins user -> tweet.


         .join(Favorite, JOIN.LEFT_OUTER)  # Joins tweet -> favorite.


         .group_by(User.username))

>>> for user in query:
...     print(user.username, user.count)
...
huey 3
mickey 1
zaizee 0

SELECT tweet.content, COUNT(favorite.id)
FROM tweet
INNER JOIN user ON tweet.user_id = user.id
LEFT OUTER JOIN favorite ON favorite.tweet_id = tweet.id
WHERE user.username = 'huey'
GROUP BY tweet.content;

query = (Tweet


         .select(Tweet.content, fn.COUNT(Favorite.id).alias('count'))


         .join(User)  # Join from tweet -> user.


         .switch(Tweet)  # Move "join context" back to tweet.


         .join(Favorite, JOIN.LEFT_OUTER)  # Join from tweet -> favorite.


         .where(User.username == 'huey')


         .group_by(Tweet.content))

query = (Tweet


         .select(Tweet.content, fn.COUNT(Favorite.id).alias('count'))


         .join_from(Tweet, User)  # Join tweet -> user.


         .join_from(Tweet, Favorite, JOIN.LEFT_OUTER)  # Join tweet -> favorite.


         .where(User.username == 'huey')


         .group_by(Tweet.content))

>>> for tweet in query:
...     print('%s favorited %d times' % (tweet.content, tweet.count))
...
meow favorited 1 times
hiss favorited 0 times
purr favorited 2 times

>>> for tweet in Tweet.select():
...     print(tweet.user.username, '->', tweet.content)
...
huey -> meow
huey -> hiss
huey -> purr
mickey -> woof
mickey -> whine

SELECT tweet.content, user.username
FROM tweet
INNER JOIN user ON tweet.user_id = user.id;

>>> for row in Tweet.select(Tweet.content, User.username).join(User).dicts():
...     print(row)
...
{'content': 'meow', 'username': 'huey'}
{'content': 'hiss', 'username': 'huey'}
{'content': 'purr', 'username': 'huey'}
{'content': 'woof', 'username': 'mickey'}
{'content': 'whine', 'username': 'mickey'}

>>> for tweet in Tweet.select(Tweet.content, User.username).join(User):
...     print(tweet.user.username, '->', tweet.content)
...
huey -> meow
huey -> hiss
huey -> purr
mickey -> woof
mickey -> whine

>>> query = Tweet.select(Tweet.content, User.username).join(User, attr='author')
>>> for tweet in query:
...     print(tweet.author.username, '->', tweet.content)
...
huey -> meow
huey -> hiss
huey -> purr
mickey -> woof
mickey -> whine

>>> for tweet in query.objects():
...     print(tweet.username, '->', tweet.content)
...
huey -> meow
(etc)

SELECT owner.username, tweet.content, author.username AS author
FROM favorite
INNER JOIN user AS owner ON (favorite.user_id = owner.id)
INNER JOIN tweet ON (favorite.tweet_id = tweet.id)
INNER JOIN user AS author ON (tweet.user_id = author.id);

Owner = User.alias()
query = (Favorite


         .select(Favorite, Tweet.content, User.username, Owner.username)


         .join(Owner)  # Join favorite -> user (owner of favorite).


         .switch(Favorite)


         .join(Tweet)  # Join favorite -> tweet


         .join(User))   # Join tweet -> user

>>> for fav in query:
...     print(fav.user.username, 'liked', fav.tweet.content, 'by', fav.tweet.user.username)
...
huey liked whine by mickey
mickey liked purr by huey
zaizee liked meow by huey
zaizee liked purr by huey

SELECT tweet.*, user.*
FROM tweet
INNER JOIN (


    SELECT latest.user_id, MAX(latest.timestamp) AS max_ts


    FROM tweet AS latest


    GROUP BY latest.user_id) AS latest_query
ON ((tweet.user_id = latest_query.user_id) AND (tweet.timestamp = latest_query.max_ts))
INNER JOIN user ON (tweet.user_id = user.id)

# Define our subquery first. We'll use an alias of the Tweet model, since
# we will be querying from the Tweet model directly in the outer query.
Latest = Tweet.alias()
latest_query = (Latest


                .select(Latest.user, fn.MAX(Latest.timestamp).alias('max_ts'))


                .group_by(Latest.user)


                .alias('latest_query'))
# Our join predicate will ensure that we match tweets based on their
# timestamp *and* user_id.
predicate = ((Tweet.user == latest_query.c.user_id) &


             (Tweet.timestamp == latest_query.c.max_ts))
# We put it all together, querying from tweet and joining on the subquery
# using the above predicate.
query = (Tweet


         .select(Tweet, User)  # Select all columns from tweet and user.


         .join(latest_query, on=predicate)  # Join tweet -> subquery.


         .join_from(Tweet, User))  # Join from tweet -> user.

>>> for tweet in query:
...     print(tweet.user.username, '->', tweet.content)
...
huey -> purr
mickey -> whine

WITH latest AS (


    SELECT user_id, MAX(timestamp) AS max_ts


    FROM tweet


    GROUP BY user_id)
SELECT tweet.*, user.*
FROM tweet
INNER JOIN latest


    ON ((latest.user_id = tweet.user_id) AND (latest.max_ts = tweet.timestamp))
INNER JOIN user


    ON (tweet.user_id = user.id)

# Define our CTE first. We'll use an alias of the Tweet model, since
# we will be querying from the Tweet model directly in the main query.
Latest = Tweet.alias()
cte = (Latest


       .select(Latest.user, fn.MAX(Latest.timestamp).alias('max_ts'))


       .group_by(Latest.user)


       .cte('latest'))
# Our join predicate will ensure that we match tweets based on their
# timestamp *and* user_id.
predicate = ((Tweet.user == cte.c.user_id) &


             (Tweet.timestamp == cte.c.max_ts))
# We put it all together, querying from tweet and joining on the CTE
# using the above predicate.
query = (Tweet


         .select(Tweet, User)  # Select all columns from tweet and user.


         .join(cte, on=predicate)  # Join tweet -> CTE.


         .join_from(Tweet, User)  # Join from tweet -> user.


         .with_cte(cte))

>>> for tweet in query:
...     print(tweet.user.username, '->', tweet.content)
...
huey -> purr
mickey -> whine

class Relationship(BaseModel):


    from_user = ForeignKeyField(User, backref='relationships')


    to_user = ForeignKeyField(User, backref='related_to')


    class Meta:


        indexes = (


            # Specify a unique multi-column index on from/to-user.


            (('from_user', 'to_user'), True),


        )

(User


 .select()


 .join(Relationship, on=Relationship.to_user)


 .where(Relationship.from_user == charlie))

(User


 .select()


 .join(Relationship, on=Relationship.from_user)


 .where(Relationship.to_user == charlie))

user_log = (User


            .select(User, ActivityLog)


            .join(ActivityLog, on=(User.id == ActivityLog.object_id), attr='log')


            .where(


                (ActivityLog.activity_type == 'user_activity') &


                (User.username == 'charlie')))
for user in user_log:


    print(user.username, user.log.description)
#### Print something like ####
charlie logged in
charlie posted a tweet
charlie retweeted
charlie posted a tweet
charlie logged out

join(ActivityLog, on=(User.id == ActivityLog.object_id), attr='log')

Then when iterating over the query, we were able to directly access the joined ActivityLog without incurring an additional query:

for user in user_log:


    print(user.username, user.log.description)

class Category(Model):


    name = CharField()


    parent = ForeignKeyField('self', backref='children')

Parent = Category.alias()
query = (Category


         .select()


         .join(Parent, on=(Category.parent == Parent.id))


         .where(Parent.name == 'Electronics'))

Parent = Category.alias()
join_query = Parent.select().where(Parent.name == 'Electronics')
# Subqueries used as JOINs need to have an alias.
join_query = join_query.alias('jq')
query = (Category


         .select()


         .join(join_query, on=(Category.parent == join_query.c.id)))

SELECT t1."id", t1."name", t1."parent_id"
FROM "category" AS t1
INNER JOIN (


  SELECT t2."id"


  FROM "category" AS t2


  WHERE (t2."name" = ?)) AS jq ON (t1."parent_id" = "jq"."id")

Category.parent == join_query.c.id
# Becomes: (t1."parent_id" = "jq"."id")

class Student(Model):


    name = CharField()
class Course(Model):


    name = CharField()
class StudentCourse(Model):


    student = ForeignKeyField(Student)


    course = ForeignKeyField(Course)

query = (Student


         .select()


         .join(StudentCourse)


         .join(Course)


         .where(Course.name == 'math'))
for student in query:


    print(student.name)

courses = (Course


           .select()


           .join(StudentCourse)


           .join(Student)


           .where(Student.name == 'da vinci'))
for course in courses:


    print(course.name)

query = (StudentCourse


         .select(StudentCourse, Student, Course)


         .join(Course)


         .switch(StudentCourse)


         .join(Student)


         .order_by(Student.name))

for student_course in query:


    print(student_course.student.name, '->', student_course.course.name)

from peewee import *
db = SqliteDatabase('school.db')
class BaseModel(Model):


    class Meta:


        database = db
class Student(BaseModel):


    name = CharField()
class Course(BaseModel):


    name = CharField()


    students = ManyToManyField(Student, backref='courses')
StudentCourse = Course.students.get_through_model()
db.create_tables([


    Student,


    Course,


    StudentCourse])
# Get all classes that "huey" is enrolled in:
huey = Student.get(Student.name == 'Huey')
for course in huey.courses.order_by(Course.name):


    print(course.name)
# Get all students in "English 101":
engl_101 = Course.get(Course.name == 'English 101')
for student in engl_101.students:


    print(student.name)
# When adding objects to a many-to-many relationship, we can pass
# in either a single model instance, a list of models, or even a
# query of models:
huey.courses.add(Course.select().where(Course.name.contains('English')))
engl_101.students.add(Student.get(Student.name == 'Mickey'))
engl_101.students.add([


    Student.get(Student.name == 'Charlie'),


    Student.get(Student.name == 'Zaizee')])
# The same rules apply for removing items from a many-to-many:
huey.courses.remove(Course.select().where(Course.name.startswith('CS')))
engl_101.students.remove(huey)
# Calling .clear() will remove all associated objects:
cs_150.students.clear()

A ManyToManyField, despite its name, is not a field in the usual sense. Instead of being a column on a table, the many-to-many field covers the fact that behind-the-scenes there's actually a separate table with two foreign-key pointers (the through table).

Therefore, when a subclass is created that inherits a many-to-many field, what actually needs to be inherited is the through table. Because of the potential for subtle bugs, Peewee does not attempt to automatically subclass the through model and modify its foreign-key pointers. As a result, many-to-many fields typically will not work with inheritance.

query = (Tweet


         .select(Tweet, User)  # Note that we are selecting both models.


         .join(User)  # Use an INNER join because every tweet has an author.


         .order_by(Tweet.id.desc())  # Get the most recent tweets.


         .limit(10))
for tweet in query:


    print(tweet.user.username, '-', tweet.message)

week_ago = datetime.date.today() - datetime.timedelta(days=7)
users = User.select()
tweets = (Tweet


          .select()


          .where(Tweet.timestamp >= week_ago))
# This will perform two queries.
users_with_tweets = prefetch(users, tweets)
for user in users_with_tweets:


    print(user.username)


    for tweet in user.tweets:


        print('  ', tweet.message)

Parameters

• Executing queries
• Managing connections
• Transactions
• Introspection

param = '?'

String used as parameter placeholder in SQL queries.

quote = '"'

Type of quotation-mark to use to denote entities such as tables or columns.

init(database[, **kwargs])

Initialize a deferred database. See deferring_initialization for more info.

__enter__()

The Database instance can be used as a context-manager, in which case a connection will be held open for the duration of the wrapped block.

Additionally, any SQL executed within the wrapped block will be executed in a transaction.

connection_context()

Create a context-manager that will hold open a connection for the duration of the wrapped block.

Example:

def on_app_startup():


    # When app starts up, create the database tables, being sure


    # the connection is closed upon completion.


    with database.connection_context():


        database.create_tables(APP_MODELS)

connect([reuse_if_open=False])

Open a connection to the database.

close()

Close the connection to the database.

is_closed()

connection()

Return the open connection. If a connection is not open, one will be opened. The connection will be whatever the underlying database-driver uses to encapsulate a database connection.

cursor([commit=None])

Return a cursor object on the current connection. If a connection is not open, one will be opened. The cursor will be whatever the underlying database-driver uses to encapsulate a database cursor.

execute_sql(sql[, params=None[, commit=SENTINEL]])

Execute a SQL query and return a cursor over the results.

execute(query[, commit=SENTINEL[, **context_options]])

Execute a SQL query by compiling a Query instance and executing the resulting SQL.

last_insert_id(cursor[, query_type=None])

rows_affected(cursor)

in_transaction()

atomic()

Create a context-manager which runs any queries in the wrapped block in a transaction (or save-point if blocks are nested).

Calls to atomic() can be nested.

atomic() can also be used as a decorator.

Example code:

with db.atomic() as txn:


    perform_operation()


    with db.atomic() as nested_txn:


        perform_another_operation()

Transactions and save-points can be explicitly committed or rolled-back within the wrapped block. If this occurs, a new transaction or savepoint is begun after the commit/rollback.

Example:

with db.atomic() as txn:


    User.create(username='mickey')


    txn.commit()  # Changes are saved and a new transaction begins.


    User.create(username='huey')


    txn.rollback()  # "huey" will not be saved.


    User.create(username='zaizee')
# Print the usernames of all users.
print [u.username for u in User.select()]
# Prints ["mickey", "zaizee"]

manual_commit()

Create a context-manager which disables all transaction management for the duration of the wrapped block.

Example:

with db.manual_commit():


    db.begin()  # Begin transaction explicitly.


    try:


        user.delete_instance(recursive=True)


    except:


        db.rollback()  # Rollback -- an error occurred.


        raise


    else:


        try:


            db.commit()  # Attempt to commit changes.


        except:


            db.rollback()  # Error committing, rollback.


            raise

The above code is equivalent to the following:

with db.atomic():


    user.delete_instance(recursive=True)

session_start()

Begin a new transaction (without using a context-manager or decorator). This method is useful if you intend to execute a sequence of operations inside a transaction, but using a decorator or context-manager would not be appropriate.

NOTE:

WARNING:

session_commit()

Commit any changes made during a transaction begun with session_start().

session_rollback()

Roll back any changes made during a transaction begun with session_start().

transaction()

Create a context-manager that runs all queries in the wrapped block in a transaction.

WARNING:

savepoint()

Create a context-manager that runs all queries in the wrapped block in a savepoint. Savepoints can be nested arbitrarily.

WARNING:

begin()

Begin a transaction when using manual-commit mode.

NOTE:

commit()

Manually commit the currently-active transaction.

NOTE:

rollback()

Manually roll-back the currently-active transaction.

NOTE:

batch_commit(it, n)

The purpose of this method is to simplify batching large operations, such as inserts, updates, etc. You pass in an iterable and the number of items-per-batch, and the items will be returned by an equivalent iterator that wraps each batch in a transaction.

Example:

# Some list or iterable containing data to insert.
row_data = [{'username': 'u1'}, {'username': 'u2'}, ...]
# Insert all data, committing every 100 rows. If, for example,
# there are 789 items in the list, then there will be a total of
# 8 transactions (7x100 and 1x89).
for row in db.batch_commit(row_data, 100):


    User.create(**row)

An alternative that may be more efficient is to batch the data into a multi-value INSERT statement (for example, using Model.insert_many()):

with db.atomic():


    for idx in range(0, len(row_data), 100):


        # Insert 100 rows at a time.


        rows = row_data[idx:idx + 100]


        User.insert_many(rows).execute()

table_exists(table[, schema=None])

get_tables([schema=None])

get_indexes(table[, schema=None])

Return a list of IndexMetadata tuples.

Example:

print(db.get_indexes('entry'))
[IndexMetadata(


     name='entry_public_list',


     sql='CREATE INDEX "entry_public_list" ...',


     columns=['timestamp'],


     unique=False,


     table='entry'),


 IndexMetadata(


     name='entry_slug',


     sql='CREATE UNIQUE INDEX "entry_slug" ON "entry" ("slug")',


     columns=['slug'],


     unique=True,


     table='entry')]

get_columns(table[, schema=None])

Return a list of ColumnMetadata tuples.

Example:

print(db.get_columns('entry'))
[ColumnMetadata(


     name='id',


     data_type='INTEGER',


     null=False,


     primary_key=True,


     table='entry'),


 ColumnMetadata(


     name='title',


     data_type='TEXT',


     null=False,


     primary_key=False,


     table='entry'),


 ...]

get_primary_keys(table[, schema=None])

Return a list of column names that comprise the primary key.

Example:

print(db.get_primary_keys('entry'))
['id']

get_foreign_keys(table[, schema=None])

Return a list of ForeignKeyMetadata tuples for keys present on the table.

Example:

print(db.get_foreign_keys('entrytag'))
[ForeignKeyMetadata(


     column='entry_id',


     dest_table='entry',


     dest_column='id',


     table='entrytag'),


 ...]

get_views([schema=None])

Return a list of ViewMetadata tuples for VIEWs present in the database.

Example:

print(db.get_views())
[ViewMetadata(


     name='entries_public',


     sql='CREATE VIEW entries_public AS SELECT ... '),


 ...]

sequence_exists(seq)

create_tables(models[, **options])

Create tables, indexes and associated metadata for the given list of models.

Dependencies are resolved so that tables are created in the appropriate order.

drop_tables(models[, **options])