Use peewee instead of SQLAlchemy

created at 07-28-2021 views: 11

I have used SQLAlchemy for several years, but to be honest, it is very complicated, including the supporting DB Migration tool alembic. It is also very complicated to configure an automatically generated migration. I have used peewee for a few days recently and found it is really simple and easy to use.

How to use it?

Common operations

Let's take a look at common operations:

Define model (

import datetime

from peewee import (

from .base import BaseModel

class Investment(BaseModel):
     project_name = CharField(help_text="name of funded project")
     user_name = CharField(help_text="investor")
     money = DecimalField(help_text="number of funds")
     transfer_at = DateTimeField(help_text="transfer date",
     remark = CharField(help_text="remarks")

The attributes of the specific Field are detailed in: here 

Add, delete, check and modify (

import datetime

from playhouse.db_url import connect
from peewee import (

from config import config

db = connect(config.DB_URL)

class BaseModel(Model):
    class Meta:
        database = db

    created_at = DateTimeField(
    updated_at = DateTimeField(
    deleted_at = DateTimeField(null=True, default=None)

    def get_all(cls):

Or the following example (from the official website, not directly run):

>>> user = User.get( == 1)
>>> user.delete_instance()  # Returns the number of rows deleted.

>>> User.get( == 1)
UserDoesNotExist: instance matching query does not exist:
SQL: SELECT t1."id", t1."username" FROM "user" AS t1 WHERE t1."id" = ?
>>> query = Tweet.delete().where(Tweet.creation_date < one_year_ago)
>>> query.execute()  # Returns the number of rows deleted.

Database migration

Use peewee_migrate to operate, but it is not good enough by itself. In order to combine it with the project, I wrote a script (

import sys

from peewee_migrate import Router

from models import db

def gen():
    with db:
        router = Router(db)

def run():
    with db:
        router = Router(db)

def new():
    with db:
        router = Router(db)

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: ./ gen|run|new")

    action = sys.argv[1]
    if action == "run":
    elif action == "gen":
    elif action == "new":

In this way, the corresponding operation can be completed by executing the command:

$ ./ gen # Automatically generate migration operations based on model definitions
$ ./ run # Execute migration operation
$ ./ new # Create a new empty migration operation

Database connection pool

As above, I used the connect in playhouse, how should I deal with the connection pool? Write the connection string as something like mysql+pool://username:password@ But if you want to configure some parameters of the connection pool, it seems that you can't write this way, you have to use PooledMySQLDatabase yourself, you can refer to the document: documnet

Use with Flask

As above, after the connection pool is configured, if you run directly, you will find that after a certain amount of request, an exception that the connection pool is full will be reported. What should I do? The answer is to add db.close after every use:

# This hook ensures that a connection is opened to handle any queries
# generated by the request.
def _db_connect():

# This hook ensures that the connection is closed when we've finished
# processing the request.
def _db_close(exc):
    if not db.is_closed():

Well, that's almost it. Generally speaking, peewee is much simpler than SQLAlchemy, and it will be my queen ORM in the future.

created at:07-28-2021
edited at: 07-28-2021: