Python SQLAlchemy 速查表

发表于 | 分类于 ，

Python 是广受欢迎的脚本语言，SQLAlchemy 是 Python 的数据库访问组件，它可以用 SQL Builder 和 ORM 两种方式访问数据库，SQL Builder 较 ORM 性能优越，ORM 在 访问关联数据 / 修改数据 上较 SQL Builder 便捷，可按场景需求选用。对逻辑问题，用 SQL 和 Python 都可解决，但本着尽量简化 SQL 操作的原则，本文仅介绍了 SQLAlchemy 的常用操作，适用于大部分编程场景。

准备数据

假设已创建以下表

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
29
30
31
32
33
34
35
-- posts
CREATE TABLE `posts` (
  `id` BIGINT(20) NOT NULL AUTO_INCREMENT,
  `title` VARCHAR(64) NOT NULL,
  `content` VARCHAR(1024) NOT NULL,
  `created_at` INT(11) DEFAULT NULL,
  `updated_at` INT(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=INNODB DEFAULT CHARSET=utf8mb4

-- comments
-- post 和 comment 是一对多关系
CREATE TABLE `comments` (
  `id` BIGINT(20) NOT NULL AUTO_INCREMENT,
  `post_id` BIGINT(20) NOT NULL,
  `content` VARCHAR(1024) NOT NULL,
  `created_at` INT(11) DEFAULT NULL,
  `updated_at` INT(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=INNODB DEFAULT CHARSET=utf8mb4

-- tags
CREATE TABLE `tags` (
  `id` BIGINT(20) NOT NULL AUTO_INCREMENT,
  `tag` VARCHAR(64) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=INNODB DEFAULT CHARSET=utf8mb4

-- post_tag
-- post_tag 定义了 post 和 tag 的多对多关系
CREATE TABLE `post_tag` (
  `post_id` BIGINT(20) NOT NULL,
  `tag_id` BIGINT(20) NOT NULL,
  PRIMARY KEY (`post_id`, `tag_id`)
) ENGINE=INNODB DEFAULT CHARSET=utf8mb4

数据库连接

连接 mysql 数据库

1
2
import sqlalchemy as sa
dbEngine = sa.create_engine('mysql://username:password@hostname/dbname')  # type: sa.engine.base.Engine

查看 sqlalchemy 版本

1
sa.__version__

SQL 方式访问数据库

定义数据库表 Model

此 Model 可用于 SQL Builder

1
2
3
4
5
6
7
8
9
10
11
dbMeta = sa.MetaData(bind=dbEngine)  # type: MetaData

TPost = sa.Table('posts', dbMeta, autoload=True)
TComment = sa.Table('comments', dbMeta, 
    sa.Column("post_id", sa.BigInteger, sa.ForeignKey("posts.id")),   # 定义外键
    autoload=True)
TTag = sa.Table('tags', dbMeta, autoload=True)
TPostTag = sa.Table('post_tag', dbMeta, 
    sa.Column("post_id", sa.BigInteger, sa.ForeignKey("posts.id"), primary_key=True),   # 定义外键
    sa.Column("tag_id", sa.BigInteger, sa.ForeignKey("tags.id"), primary_key=True),   # 定义外键
    autoload=True)

创建连接

1
dbconn = dbEngine.connect()

查询

1
2
3
4
5
6
7
8
9
10
11
12
13
stmt = sa.sql.select([TPost]) \
    .where(TPost.c.title == 'hello') \
    # LIKE
    .where(TPost.c.title.like('hel%')) \
    # IN
    .where(TPost.c.title.in_(['hello'])) \
    # NOT IN
    .where(~TPost.c.title.in_(['mike'])) \
    .where(sa.sql.and_(TPost.c.id > 5, TPost.c.id < 20)) \
    # IS NULL
    .where(sa.sql.or_(TPost.c.created_at < 1519818304, TPost.c.created_at == None)) \
    .order_by(sa.desc(TPost.c.created_at), TPost.c.title)
dbconn.execute(stmt).fetchall()     # output [(1, 'post title', 'post content', 1519818304, None)]

查询，直接用 SQL

1
2
r = dbconn.execute(sa.text('SELECT * FROM posts WHERE id=:id'), id=7)
r.fetchall()    # output [(1, 'post title', 'post content', 1519818304, None)]

插入

1
2
3
r = dbconn.execute(TPost.insert(), [{'title': 'new title', 'content':'new content'}, {'title': 'new title2', 'content':'new content2'}])
# 受影响的行 affected rows
r.rowcount      # output 2

修改

1
2
3
4
stmt = TPost.update().values(title='new title', content='new content').where(TPost.c.id == 1)
r = dbconn.execute(stmt)
# 受影响的行 affected rows
r.rowcount      # output 1 or 0

删除

1
2
3
dbconn.execute(TPost.delete().where(TPost.c.id == 1))
# 受影响的行 affected rows
r.rowcount      # output 1 or 0

ORM 方式访问数据库

定义 ORM Model

定义数据库表的 ORM Model，可用于 ORM 操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from sqlalchemy.ext.declarative import declarative_base
DBBase = declarative_base()

class OPost(DBBase):
    __table__ = TPost
    # 定义多对多关系
    tags = orm.relationship("OTag", secondary=TPostTag, back_populates="posts")

class OComment(DBBase):
    __table__ = TComment
    # 定义一对多关系
    post = orm.relationship("OPost", back_populates="comments")

# 定义一对多关系
OPost.comments = orm.relationship("OComment", cascade="all, delete-orphan", order_by=OComment.created_at, back_populates="post")

class OTag(DBBase):
    __table__ = TTag
    # 定义多对多关系
    posts = orm.relationship("OPost", secondary=TPostTag, back_populates="tags")

注意：其中 delete-orphan 让 Post 被删除时，不关联更新 Comment。参见官方文档 级联操作

创建 Session

session 管理查询状态并封装了 ORM，ORM 是一种在编程时比 SQL 更便捷的数据库操作方式，对数据库的增删操作尤其方便，但效率远不及 SQL 方式 session 需用 session maker 创建

1
2
3
4
5
6
7
8
9
# 得到 Session Maker
from sqlalchemy import orm
DBSession = orm.sessionmaker()
DBSession.configure(bind=dbEngine)
## 或
DBSession = orm.sessionmaker(bind=dbEngine)  # type: sessionmaker

# 创建 session 实例
dbsess = DBSession()

注意：每个 session 实例对其他实例对数据库的更改并不知情，如果需要得到最新数据，最简单的办法是创建新的 session

查询

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
posts = dbsess.query(OPost.id, OPost.title, OPost.created_at, OPost) \
    .filter(OPost.title == 'jacky') \
    # LIKE
    .filter(OPost.title.like('jack%')) \
    # IN
    .filter(OPost.title.in_(['jacky'])) \
    # NOT IN
    .filter(~OPost.title.in_(['mike'])) \
    .filter(sa.and_(OPost.id > 5, OPost.id < 20)) \
    # IS NULL
    .filter(sa.or_(OPost.created_at < 1519818304, OPost.created_at == None)) \
    .order_by(OPost.created_at.desc(), OPost.title)
    .all()

isinstance(posts, list)      # output True
row = posts[0]           # 返回结果集第一行
row[0]
row[1]
row[2]
post = row[3]                      # OPost Object
post.id
post.name
post.age
post.tags                 # 多对多关系
post.comments             # 一对多关系
post.__dict__             # 将 OPost Object 转换成 dict 类型

查询，直接用 SQL 子句

1
2
3
dbsess.query(OPost).filter(sa.text("created_at<:created_at and title like :title")) \
    .params(created_at=1519818304, title='%hello%') \
    .order_by(OPost.created_at).all()

查询，直接用 SQL

1
2
3
dbsess.query(OPost).from_statement( \
    sa.text("SELECT * FROM posts where title=:title")) \
    .params(title='hello world').all()

查询，统计

1
2
dbsess.query(OPost).filter(OPost.title.like("%a%")).count()
dbsess.query(sa.func.count(OComment.post_id), sa.func.max(OComment.created_at), OComment.post_id).group_by(OComment.post_id).all()

插入

1
2
3
4
post = OPost(title='hello world', content='to be or not to be? it\'s a problem.')
dbsess.add(post)
dbsess.commit()
post.id             # 数据库返回 mike 的 id（数据库自增id）

插入多行

1
2
3
4
5
dbsess.add_all([
    OPost(title='second post', content='bla bla ...'), 
    OPost(title='third post', content='ahaaha ah ...')
])
dbsess.commit()

修改

1
2
3
post = dbsess.query(OPost).filter(OPost.id == 1).first()
post.title += ' Stared'
dbsess.commit()

删除

1
2
3
post = dbsess.query(OPost).filter(OPost.id == 1).first()
dbsess.delete(post)
dbsess.commit()
相关文章