Flask 使用 SQLAlchemy

2023-10-18

字数统计: 1.9k字 | 阅读时长≈ 8分

参考文章：[Python web开发实战]

ORM 简介

随着项目越来越大，采用原生 SQL 的方式在代码中会出现大量的 SQL 语句，那么问题就出现了：

SQL 语句重复利用率不高，越复杂的SQL语句条件越多，代码越长。你会看到很多相近的 SQL 语句。

很多 SQL 语句是在业务逻辑中拼接出来的，如果有数据库需要更改，就要求开发人员非常了解这些逻辑，否则就很容易漏掉对某些 SQL 语句的修改。
写 SQL 时容易忽略 web 安全问题，给未来造成隐患。

ORM，全称 Object Relational Mapping，中文叫做对象关系映射，通过它我们可以直接使用 Python 的类的方式做数据库开发，而不再直接写原生的 SQL 语句（甚至不需要 SQL 的基础）。通过把表映射成类，把行作为实例，把字段作为属性，ORM 在执行对象操作的时候会把对应的操作转换成数据原生语句的方式来完成数据库开发工作。

ORM 有如下优点：

易用性。使用这种 ORM 数据库抽象封装方式做开发，可以有效减少出现重复SQL语句的概率，写出来的模型也更直观，清晰
性能损耗小，ORM 转换成底层数据库操作指令确实会有一些开销。
设计灵活，可以很轻松地写出复杂的查询
可移植。比如 SQLAlchemy，它支持多个关系型数据库引擎，包括流行的 MySQL，PostgreSQL 和 SQLite。可以近乎无痛地换掉数据库，只需要改很少的配置项即可。

使用 SQLAlchemy

SQLAlchemy 是最流行的关系型数据库的 ORM 框架，它由 Mako 的作者 Mike Bayer 创建。

安装 SQLAlchemy

1	pip install SQLAlchemy

连接数据库

首先需要连接到数据库：

In [5]: from sqlalchemy import create_engine                                                                         

In [6]: engine = create_engine('sqlite://', echo=False)                                                               

In [7]: with engine.connect() as con: 
   ...:     rs = con.execute('SELECT 1') 
   ...:     print(rs.fetchone()) 
   ...:                                                                                                               
(1,)

create_engine 传入了一个数据库的 URL，sqlite:// 表示使用了一个SQLite 的内存型数据库。URL 的格式如下：

1	dialect+driver://username:password@host:port/database

dialect 是数据库的实现，比如 MySQL，PostgreSQL， SQLite。driver 是 Python对应的驱动。如果不指定，会选择默认的驱动。比如 MySQL 的默认驱动是 MySQLdb：

1	engine = create_engine('mysql+mysqldb://scott:tiger@localhost/foo')

因为 mysqldb 是默认的驱动，所以可以不写驱动部分：

1	engine = create_engine('mysql://scott:tiger@localhost/foo')

执行 SQL 时也可以更简略的这样写：

1
2
3

rs = engine.execute('select 1')
rs.fetchone()
(1,)

如果你需要更详细的输出，可以设置 echo=True

使用原生 SQL

我们把之前的例子改写成使用 SQLAlchemy

示例一: `raw_sql.py`

from sqlalchemy import create_engine
from consts import DB_URL


eng = create_engine(DB_URL)
with eng.connect() as conn:
    conn.execute('drop table if exists users')
    conn.execute('create table users(ID INT PRIMARY KEY AUTO_INCREMENT, '
                 'Name VARCHAR(25))')
    conn.execute('insert into users(Name) values("xiaoming")')
    conn.execute('insert into users(Name) values("wanglang")')
    rs = conn.execute('select * from users')
    for row in rs:
        print(row)

它和之前的 MySQLdb 例子的不同之处在于，结果通过返回值获取，不再需要执行 fetchone 或者 fetchall 也能获取到。

使用表达式

SQLAlchemy 支持使用表达式的方式来操作数据库，这种方式和 Ruby On Rails 中的 Active Record 模式很像。

示例三: `exp_sql.py`

from sqlalchemy import (
    create_engine, Table, MetaData, Column, Integer, String, tuple_
)
from sqlalchemy.sql import select, asc, and_
from consts import DB_URL


eng = create_engine(DB_URL)
meta = MetaData(eng)

users = Table(
    'users', meta,
    Column('ID', Integer, primary_key=True, autoincrement=True),
    Column('Name', String(50), nullable=False),
)

if users.exists():
    users.drop()
    users.create()
else:
    users.create()


def executes(s):
    print('-' * 20)
    rs = con.execute(s)
    for row in rs:
        print(row['ID'], row['Name'])


with eng.connect() as con:
    for username in ('xiaoming', 'wanglang', 'lilei'):
        user = users.insert().values(Name=username)
        con.execute(user)

    stm = select([users]).limit(1)
    executes(stm)

    k = [(2, )]
    stm = select([users]).where(tuple_(users.c.ID).in_(k))
    executes(stm)

    stm = select([users]).where(and_(users.c.ID > 2,
                                     users.c.ID < 4))
    executes(stm)

    stm = select([users]).order_by(asc(users.c.Name))
    executes(stm)

    stm = select([users]).where(users.c.Name.like('%min%'))
    executes(stm)

分析一下这个例子：

使用 users.create() 的方式来创建表，如果需要创建的表比较多，也可以选择使用 meta.create_all(eng)。
代码中的 stm 变量就是一个生成好的 SQL 语句。比如 “stm=select([users]).order_by(asc(users.Name))”, stm 的结果是：
1
SELECT users.`ID`, users.`Name` FROM users ORDER BY users.`Name` ASC

使用 ORM

ORM 是基于 SQLAlchemy 表达式语言的，看一个例子：
示例四：orm_sql.py

from sqlalchemy import create_engine, Column, Integer, String, Sequence
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import and_, or_
from sqlalchemy.orm import sessionmaker
from consts import DB_URL


eng = create_engine(DB_URL)
Base = declarative_base()


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, Sequence('user_id_seq'), primary_key=True, autoincrement=True)
    name = Column(String(50))


Base.metadata.drop_all(bind=eng)    # 删除表
Base.metadata.create_all(bind=eng)  # 创建表

Session = sessionmaker(bind=eng)
session = Session()

session.add_all([User(name=username) for username in ('xiaoming', 'wanglang', 'lilei')])
session.commit()


def get_result(rs):
    print('-' * 20)
    for user in rs:
        print(user.name)


rs = session.query(User).all()
get_result(rs)

rs = session.query(User).filter(User.id.in_([2, ]))
get_result(rs)

rs = session.query(User).filter(and_(User.id > 2, User.id < 4))
get_result(rs)

rs = session.query(User).filter(or_(User.id == 2, User.id == 4))
get_result(rs)

rs = session.query(User).filter(User.name.like('%min%'))
get_result(rs)

user = session.query(User).filter_by(name='xiaoming').first()
get_result([user])

分析一下这个例子：

定义的 User 类会生成一张表，__tablename__ 的值就是表名；
通过 sessionmaker 创建一个会话，会话提供了事务控制的支持。模型实例对象本身独立存在，如果要让其修改（创建）生效，需要把它们加入某个会话；如果不希望对其生效就从会话中去掉由 session 管理的实例对象。执行 session.commit() 时修改被提交到数据库，执行 session.rollback() 可以回滚变更。

例子里在执行开始前会先删除表，但实际工作中请不要这样用。

还可以通过 sqlalchemy.txt 写复杂的条件语句来操作数据库，如下所示：
示例五：text_sql.py

rs = session.query(User).filter(
    text('id > 2 and id < 4')).order_by(text('id')).all()
get_result(rs)

rs = session.query(User).filter(text('id<:value and name=:name')).params(
    value=3, name='xiaoming').all()
get_result(rs)

rs = session.query(User).from_statement(
    text('Select * from users where name=:name')).params(name='wanglang').all()
get_result(rs)

数据库关联

InnoDB 类型的表可以使用外键进行多表关联，保证数据的一致性和实现一些级联操作。我们看一个例子：
rel_sql.py

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from consts import DB_URL

eng = create_engine(DB_URL)
Base = declarative_base()


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50))


class Address(Base):
    __tablename__ = 'addresses'

    id = Column(Integer, primary_key=True, autoincrement=True)
    email_address = Column(String(128), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    # user 字段关联到 User 表
    user = relationship('User', back_populates='addresses')


User.addresses = relationship("Address", order_by=Address.id, back_populates="user")

Base.metadata.drop_all(bind=eng)
Base.metadata.create_all(bind=eng)

Session = sessionmaker(bind=eng)
session = Session()

user = User(name='xiaoming')
user.address = [Address(email_address='a@gmail.com', user_id=user.id),
                Address(email_address='b@gmail.com', user_id=user.id)]
session.add(user)
session.commit()

Address 表的 user_id 字段其实就是 User 的 id 字段，使用 ForeignKey 关联之后，就不需要在 Address 上独立存储一份 user_id 数据。

虽然使用外键可以降低开发成本，减少数据量，但是在用户量大，并发度高的时候，不推荐使用外键来关联，数据的一致性和完整性问题可以通过事务来保证。