.. code-block:: console

pip install peewee[mysql] # Install peewee with pymysql. pip install peewee[postgres] # Install peewee with psycopg2. pip install peewee[psycopg3] # Install peewee with psycopg3.

# AsyncIO implementations. pip install peewee[aiosqlite] # Install peewee with aiosqlite. pip install peewee[aiomysql] # Install peewee with aiomysql. pip install peewee[asyncpg] # Install peewee with asyncpg.

Examples

Defining models is similar to Django or SQLAlchemy:

.. code-block:: python

from peewee import * import datetime

db = SqliteDatabase('my_database.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)

Connect to the database and create tables:

.. code-block:: python

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

Create a few rows:

.. code-block:: python

charlie = User.create(username='charlie') huey = User(username='huey') huey.save()

Queries are expressive and composable:

.. code-block:: python

# A simple query selecting a user. User.get(User.username == 'charlie')

# Get tweets created by one of several users. usernames = ['charlie', 'huey', 'mickey'] users = User.select().where(User.username.in_(usernames)) tweets = Tweet.select().where(Tweet.user.in_(users))

# We could accomplish the same using a JOIN: tweets = (Tweet .select() .join(User) .where(User.username.in_(usernames)))

# How many tweets were published today? tweets_today = (Tweet .select() .where( (Tweet.created_date >= datetime.date.today()) & (Tweet.is_published == True)) .count())

# Paginate the user table and show me page 3 (users 41-60). User.select().order_by(User.username).paginate(3, 20)

# Order users by the number of tweets they've created: tweet_ct = fn.Count(Tweet.id) users = (User .select(User, tweet_ct.alias('ct')) .join(Tweet, JOIN.LEFT_OUTER) .group_by(User) .orderby(tweetct.desc()))

# Do an atomic update (for illustrative purposes only, imagine a simple # table for tracking a "count" associated with each URL). We don't want to # naively get the save in two separate steps since this is prone to race # conditions. Counter.update(count=Counter.count + 1).where(Counter.url == request.url).execute()

Asyncio

.. code-block:: python

import asyncio from peewee import * from playhouse.pwasyncio import AsyncPostgresqlDatabase

db = AsyncPostgresqlDatabase('my_app')

class User(db.Model): username = CharField(unique=True)

class Tweet(db.Model): user = ForeignKeyField(User, backref='tweets') message = TextField()

async def main(): async with db: await db.acreate_tables([User, Tweet])

# Queries are awaited on the event loop using asyncpg. huey = await User.acreate(username='huey') tweet = await Tweet.acreate(user=huey, message='meow')

async with db.atomic(): tweet.message = 'purr' await tweet.asave()

# Create a query - nothing is executed yet. query = Tweet.select(Tweet, User).join(User)

# Execute and buffer the results. tweets = await query.aexecute() # Or: await db.list(query) for tweet in tweets: print(tweet.user.username, '->', tweet.message)

# Streaming results via server-side cursor. async for tweet in db.iterate(query): print(tweet.user.username, '->', tweet.message)

await db.close_pool()

asyncio.run(main())

Learning more

Still want more info?

.. image:: https://media.charlesleifer.com/blog/photos/wat.jpg

I've written a number of blog posts about building applications and web-services with peewee (and usually Flask). If you'd like to see some real-life applications that use peewee, the following resources may be useful: