Flask 中的 SQLAlchemy¶

许多人喜欢使用 SQLAlchemy 进行数据库访问。在这种情况下，建议为你的 Flask 应用使用包而不是模块，并将模型放入单独的模块中（大型应用作为包）。虽然这不是必须的，但这样做很有意义。

有四种非常常见的使用 SQLAlchemy 的方法。我将在此概述每一种方法

Flask-SQLAlchemy 扩展¶

由于 SQLAlchemy 是一个常见的数据库抽象层和对象关系映射器，需要一些配置工作，因此有一个 Flask 扩展可以为你处理这些。如果你想快速入门，建议使用这个扩展。

声明式¶

SQLAlchemy 中的声明式扩展是最近使用 SQLAlchemy 的方法。它允许你一次性定义表和模型，类似于 Django 的工作方式。除了以下文本，我还建议阅读关于声明式扩展的官方文档。

这是你的应用程序的示例 database.py 模块

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker, declarative_base

engine = create_engine('sqlite:////tmp/test.db')
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()

def init_db():
    # import all modules here that might define models so that
    # they will be registered properly on the metadata.  Otherwise
    # you will have to import them first before calling init_db()
    import yourapplication.models
    Base.metadata.create_all(bind=engine)

要定义你的模型，只需子类化上面代码创建的 Base 类。如果你想知道为什么我们在这里不必关心线程（就像我们在上面的 SQLite3 示例中使用 g 对象那样）：那是因为 SQLAlchemy 已经通过 scoped_session 为我们做了这件事。

要在你的应用程序中以声明式方式使用 SQLAlchemy，你只需将以下代码放入你的应用程序模块中。Flask 将在请求结束或应用程序关闭时自动删除数据库会话

from yourapplication.database import db_session

@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

这是一个示例模型（例如，将其放入 models.py 中）

from sqlalchemy import Column, Integer, String
from yourapplication.database import Base

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    email = Column(String(120), unique=True)

    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

    def __repr__(self):
        return f'<User {self.name!r}>'

要创建数据库，你可以使用 init_db 函数

>>> from yourapplication.database import init_db
>>> init_db()

你可以像这样将条目插入数据库

>>> from yourapplication.database import db_session
>>> from yourapplication.models import User
>>> u = User('admin', 'admin@localhost')
>>> db_session.add(u)
>>> db_session.commit()

查询也很简单

>>> User.query.all()
[<User 'admin'>]
>>> User.query.filter(User.name == 'admin').first()
<User 'admin'>

手动对象关系映射¶

与上面的声明式方法相比，手动对象关系映射有一些优点和缺点。主要区别在于你分别定义表和类，并将它们映射在一起。它更灵活，但需要键入的内容更多。总的来说，它的工作方式类似于声明式方法，因此请确保也将你的应用程序拆分为包中的多个模块。

这是你的应用程序的示例 database.py 模块

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine('sqlite:////tmp/test.db')
metadata = MetaData()
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
def init_db():
    metadata.create_all(bind=engine)

与声明式方法一样，你需要在每次请求或应用程序上下文关闭后关闭会话。将此代码放入你的应用程序模块中

from yourapplication.database import db_session

@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

这是一个示例表和模型（将其放入 models.py 中）

from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from yourapplication.database import metadata, db_session

class User(object):
    query = db_session.query_property()

    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

    def __repr__(self):
        return f'<User {self.name!r}>'

users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50), unique=True),
    Column('email', String(120), unique=True)
)
mapper(User, users)

查询和插入的工作方式与上面的示例完全相同。

SQL 抽象层¶

如果你只想使用数据库系统（和 SQL）抽象层，你基本上只需要引擎

from sqlalchemy import create_engine, MetaData, Table

engine = create_engine('sqlite:////tmp/test.db')
metadata = MetaData(bind=engine)

然后你可以像上面的示例中那样在你的代码中声明表，或者自动加载它们

from sqlalchemy import Table

users = Table('users', metadata, autoload=True)

要插入数据，你可以使用 insert 方法。我们必须首先获取连接，以便我们可以使用事务

>>> con = engine.connect()
>>> con.execute(users.insert(), name='admin', email='admin@localhost')

SQLAlchemy 会自动为我们提交。

要查询你的数据库，你可以直接使用引擎或使用连接

>>> users.select(users.c.id == 1).execute().first()
(1, 'admin', 'admin@localhost')

这些结果也是类似字典的元组

>>> r = users.select(users.c.id == 1).execute().first()
>>> r['name']
'admin'

你还可以将 SQL 语句字符串传递给 execute() 方法

>>> engine.execute('select * from users where id = :1', [1]).first()
(1, 'admin', 'admin@localhost')

有关 SQLAlchemy 的更多信息，请访问网站。

Flask 中的 SQLAlchemy¶

Flask-SQLAlchemy 扩展¶

声明式¶

手动对象关系映射¶

SQL 抽象层¶

目录

导航