Sqlalchemy Python ORM(Object Relational Mapping) 使用介绍
介绍
Sqlalchemy API 分新旧两种：
1.3 及之前的老 API1.4 及之后的 2.0 新 API
| 概念 | 对应数据库 | 说明 |
|---|
| Engine | 连接 | 驱动引擎 |
| Session | 连接池，事务 | 由此开始查询 |
| Model | 表 | 类定义 |
| Column | 列 | |
| Query | 若干行 | 可以链式添加多个条件 |
| 数据类型 | 数据库数据类型 | python数据类型 | 说明 |
|---|
| Integer | int | int | 整形，32位 |
| String | varchar | string | 字符串 |
| Text | text | string | 长字符串 |
| Float | float | float | 浮点型 |
| Boolean | tinyint | bool | True / False |
| Date | date | datetime.date | 存储时间年月日 |
| DateTime | datetime | datetime.datetime | 存储年月日时分秒毫秒等 |
| Time | time | datetime.datetime | 存储时分秒 |
使用步骤
- 导入SQLAlchemy
- 创建类，定义表名和字段名
- 初始化数据库链接：
create_engine('数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名') - 创建session：
sessionmaker(bind=engine) session.add(对象)session.commit()session.close()
安装
pip install sqlalchemy
create_engine
from sqlalchemy import create_engine
engine = create_engine(
DATABASE_URI,
# connect_args={"ssl": {"key": SQLALCHEMY_DATABASE_PEM}},
echo=True,
pool_size=8,
pool_recycle=60*30,)
说明：
echo 为 True 时打印 sql 语句，一般与 debug 配合使用pool_size 连接池的大小，默认为5个，设置为0时表示连接无限制pool_recycle 设置数据库多久没连接自动断开
模型
class User(Base):
"""User account."""
__tablename__ = "user"
id = Column(Integer, primary_key=True, autoincrement="auto")
username = Column(String(255), unique=True, nullable=False)
created_at = Column(DateTime, server_default=func.now())
updated_at = Column(DateTime, onupdate=func.now())
def __repr__(self):
return "<User %r>" % self.username
生成数据表
Base.metadata.create_all(engine)
操作数据
from sqlalchemy.orm import sessionmaker
# Create database session
Session = sessionmaker(bind=engine)
session = Session()
session 的常见操作：
flush 预提交，未写入数据库中
commit 提交事务
rollback 回滚事务
close 关闭事务
增加
add_user = Users("test")
session.add(add_user)
session.commit()
| filter | filter_by |
|---|
| 比较运算符，相等比较用== | 只能使用"="，"!=“和”><" |
| 类名.属性名 | 属性名 |
| 类名.属性名 | 参数是**kwargs，支持组合查询 |
| 支持and，or和in等 | |
existing_user = (
session.query(User).filter(User.username == user.username).first()
)
session.query(Users).filter_by(id=1).update({'username': "xianbin"})
# 或
users = session.query(Users).filter_by(username="test").first()
users.username = "xianbin"
session.add(users)
session.delete(user) # Delete the user
session.commit() # Commit the change
# 或
session.query(Users).filter(Users.username == "test").delete()
session.commit()
源码
原生 SQL
from sqlalchemy import create_engine
from sqlalchemy import text
engine = create_engine('sqlite:///:memory:')
result = engine.execute(
text("SELECT * FROM users WHERE age >= :age"), {'age': 21})
for row in result:
print(row)
from sqlalchemy.orm import Session
session = Session(bind=engine)
result = session.execute(text("SELECT * FROM users WHERE age >= :age"), {'age': 21})
for row in result:
print(row)
with session.begin():
session.execute(
text("UPDATE users SET age = :new_age WHERE age = :old_age"),
{'new_age': 30, 'old_age': 20}
)
F&Q
ModuleNotFoundError: No module named ‘MySQLdb’
在 python2.x 中用 mysqldb，python3.x 中使用 pymysql 代替
import pymysql
pymysql.install_as_MySQLdb()
TimeoutError: QueuePool limit of size 5 overflow 10 reached, connection timed out, timeout 30
修改 pool_size 和 max_overflow 的大小
from sqlalchemy import create_engine
engine = create_engine(
"mysql://" + loadConfigVar("user") + ":" + loadConfigVar("password") + "@" + loadConfigVar("host") + "/" + loadConfigVar("schema"),
echo=False,
pool_size=20,
pool_recycle=60*30,
max_overflow=50)