Sqlalchemy 使用介绍

Sqlalchemy Python ORM(Object Relational Mapping) 使用介绍

介绍

Sqlalchemy API 分新旧两种：

• 1.3 及之前的老 API
• 1.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 配合使用
  • echo=False 关闭日志
• 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)