SQLAlchemy 是一个很强大的 Python ORM 工具, 这里介绍一下最常用的功能, 以供参考.
1. engine创建
1
2
3
4
5
6
7
|
from sqlalchemy import create_engine
import config
engine = create_engine(
f'mysql+mysqldb://{config.MYSQL_USER}:{config.MYSQL_PASSWORD}@{config.MYSQL_HOST}:{config.MYSQL_PORT}/{config.MYSQL_DATABASE}?charset=utf8',
pool_size=10,
pool_recycle=7200)
|
2. 定义模型
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime, Text, Integer
from sqlalchemy.sql import func
ModelBase = declarative_base()
class User(ModelBase):
__tablename__ = "sqlchemy_demo_user"
id = Column(Integer, primary_key=True)
username = Column(String(length=255))
phone_number = Column(String(length=255), index=True)
age = Column(Integer)
description = Column(Text)
created_at = Column(DateTime(timezone=False), server_default=func.now())
def __repr__(self):
return f'<User {self.openid!r}>'
def to_json(self):
return {
"id": self.id,
"username": self.username,
"phone_number": self.phone_number,
"age": self.age,
"description": self.description,
"created_at": str(self.created_at),
}
|
3. 初始化数据库
1
2
|
ModelBase.metadata.drop_all(engine)
ModelBase.metadata.create_all(engine)
|
4. 增删改查
1
2
3
4
|
from sqlalchemy import func
from sqlalchemy import distinct
from sqlalchemy.orm import Session
s = Session(engine)
|
如果是在 Web 应用中, 最好使用 scoped_session.
1
2
|
from sqlalchemy.orm import sessionmaker, scoped_session
s = scoped_session(sessionmaker(bind=engine))
|
当一个线程调用时, ScopedSession工厂会查找 registry, 看看之前是否已经为此线程创建过 session,
如果有直接返回这个 session, 没有的话创建一个新的 session, 并注册到 registry.
传递参数
1
|
s.query(User).filter("id>:id").params(id=1).all()
|
查询一个元素
1
|
s.query(User).filter(User.username == 'ZX').first()
|
查询列表
1
2
3
4
5
|
s.query(User).filter(User.age == 20, User.phone_number == '0').all()
s.query(User).filter(User.age == 20, User.phone_number == '0').limit(2).all()
s.query(User).filter(User.age == 20, User.phone_number == '0').offset(2).all()
s.query(User).filter(User.age == 20, User.phone_number == '0').offset(2).all()
s.query(User).filter(User.age == 20, User.phone_number == '0').slice(1, 4).all()
|
排序
1
|
s.query(User).order_by(User.created_at.desc()).all()
|
模糊查询
1
|
s.query(User).filter(User.username.like("Z%")).all()
|
多条件
1
2
|
s.query(User).filter(and_(User.username.like("Z%"), User.description.like("hello%"))).all()
s.query(User).filter(or_(User.username.like("Z%"), User.description.like("hello%"))).all()
|
计数
1
2
3
|
s.query(func.count(User.id))
s.query(func.count(User.id)).group_by(User.username)
s.query(func.count(distinct(User.username)))
|
存在判断
1
|
s.query(User).filter(exists().where(User.id == 10))
|
聚合
1
2
|
s.query(User.username, func.count('*').label("user_cnt")).group_by(User.username).all()
s.query(func.count(User.phone_number).label("phone_number_cnt")).first()
|
JOIN
1
2
|
s.query(User.username, func.count(Article.id)).join(Article, User.id == Article.uid).group_by(User.id).order_by(func.count(Article.id).desc()).all()
# 如果不写join的条件，默认使用外键作为条件连接。
|
限制返回字段
1
|
s.query(User.username, User.phone_number, User.created_at).filter_by(name="ZX").order_by(User.created_at).first()
|
创建新元素
1
2
3
|
u = User(username=username, phone_number=phone_number)
s.add(u)
s.commit()
|
删除
1
2
|
s.query(User).filter(User.username == 'ZX').delete()
s.commit()
|
5. Alembic
Alembic是 SQLAlchemy 官方出的一个迁移工具.
5.1 安装
5.2 初始化
会生成一个 alembic 目录 和一个 alembic.ini
5.3 配置
修改 alembic.ini
sqlalchemy.url 修改为自己的数据库连接
1
|
sqlalchemy.url = mysql+mysqldb://root:PASSWORD@127.0.0.1:3306/demo?charset=utf8
|
修改 alembic/env.py
target_metadata = None 修改为自己的 ModelBase 的 metadata
1
2
|
from models import ModelBase
target_metadata = ModelBase.metadata
|
5.4 migration
1
2
|
alembic revision --autogenerate -m "init" # 参数是描述信息
alembic upgrade head # 同步最新的修改到数据库
|
修改模型定义之后, 可生成新的migration文件, 文件保存在 alembic/versions .
1
2
|
alembic revision --autogenerate -m "modify"
alembic upgrade head
|
如果要降级
5.5 更多命令参考