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?
Let's take a look at common operations:
Define model (
import datetime from peewee import ( CharField, DecimalField, DateTimeField, ) 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", default=datetime.datetime.now) remark = CharField(help_text="remarks")
The attributes of the specific Field are detailed in: here
Add, delete, check and modify (base.py)
import datetime from playhouse.db_url import connect from peewee import ( Model, DateTimeField, ) from config import config db = connect(config.DB_URL) class BaseModel(Model): class Meta: database = db created_at = DateTimeField(default=datetime.datetime.now) updated_at = DateTimeField(default=datetime.datetime.now) deleted_at = DateTimeField(null=True, default=None) @classmethod def get_all(cls): return cls.select().order_by(cls.id.desc())
Or the following example (from the official website, not directly run):
>>> 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:  >>> query = Tweet.delete().where(Tweet.creation_date < one_year_ago) >>> query.execute() # Returns the number of rows deleted. 7
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 (migrate.py):
#!/home/jiajun/.pyenv/shims/python import sys from peewee_migrate import Router from models import db def gen(): with db: router = Router(db) router.create(auto=True) def run(): with db: router = Router(db) router.run() def new(): with db: router = Router(db) router.create() if __name__ == "__main__": if len(sys.argv) != 2: print("Usage: ./migrate.py gen|run|new") sys.exit(-1) action = sys.argv if action == "run": run() elif action == "gen": gen() elif action == "new": new()
In this way, the corresponding operation can be completed by executing the command:
$ ./migrate.py gen # Automatically generate migration operations based on model definitions $ ./migrate.py run # Execute migration operation $ ./migrate.py new # Create a new empty migration operation
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:email@example.com:3306/db. 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
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. @app.before_request def _db_connect(): db.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 db.is_closed(): db.close()
Well, that's almost it. Generally speaking, peewee is much simpler than SQLAlchemy, and it will be my queen ORM in the future.