用peewee代替SQLAlchemy

用了几年SQLAlchemy,但是说实话,复杂的很,包括配套的DB Migration工具alembic。想要配置一个自动生成migration也是很复杂。 最近用了几天peewee,发现真的是简单好用。

咋用呢?

常见操作

我们来看看常见的操作:

  • 定义model(investment.py)
import datetime

from peewee import (
    CharField,
    DecimalField,
    DateTimeField,
)

from .base import BaseModel


class Investment(BaseModel):
    project_name = CharField(help_text="出资项目名")
    user_name = CharField(help_text="出资人")
    money = DecimalField(help_text="资金数量")
    transfer_at = DateTimeField(help_text="转账日期", default=datetime.datetime.now)
    remark = CharField(help_text="备注")

具体Field的属性详见:http://docs.peewee-orm.com/en/latest/peewee/models.html#field-types-table

  • 增删查改(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())

或者如下例子(来自官网,不可直接运行):

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

数据库migration

使用 peewee_migrate 来操作,但是它本身还不够好用,为了把它和项目结合起来, 我写了一个脚本(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[1]
    if action == "run":
        run()
    elif action == "gen":
        gen()
    elif action == "new":
        new()

这样就可以通过执行命令来完成对应的操作:

$ ./migrate.py gen  # 根据model定义自动生成migration操作
$ ./migrate.py run  # 执行migration操作
$ ./migrate.py new  # 创建新的空的migration操作

数据库连接池

如上,我用了playhouse里的connect,那该咋处理连接池呢?把连接串写成类似 mysql+pool://username:password@127.0.0.1:3306/db 就可以了。但是如果你想要配置连接池的一些参数,似乎就不能这样写了,就得自己用 PooledMySQLDatabase 了,可以参照文档:

http://docs.peewee-orm.com/en/latest/peewee/playhouse.html#connection-pool

和Flask一起用

如上,配置了连接池之后,如果你直接跑,你会发现请求到了一定量之后,会报连接池满了的异常。那咋办呢?答案就是每次使用完 之后,都加上 db.close

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

好了,差不多就是这些,总体来说,peewee比SQLAlchemy要简单太多,以后就是我的御用ORM了。


微信公众号
关注公众号,获得及时更新

更多文章
  • HTTP 路由的两种常见设计形式
  • Golang的short variable declaration
  • Greenlet和Stackless Python
  • 写一个简单的ORM
  • 从源码看Python的descriptor
  • Python字符串格式化
  • Gunicorn 简明教程
  • Raft 论文阅读笔记
  • 什么是 Golang Comparable Types
  • GFS 论文阅读
  • MapReduce 论文阅读
  • 一起来做贼:Goroutine原理和Work stealing
  • Go语言的defer, panic和recover
  • 再读 Python Language Reference
  • 再读vim help:vim小技巧