SQLAlchemy in Flask

很多人更倾向于使用 SQLAlchemy 进行数据库操作。 在这种情况下,建议您使用 包的而不是模块的方式组织您的应用代码,并将所有的模型放置到一个单独的模块中 (大型应用)。 虽然这不是必要的,它有很大的意义。

使用 SQLAlchemy 有四种常用的方法,我们在下面列出了这几种方法的基本使用 框架: 我将在这里概述他们:

Flask-SQLAlchemy Extension

因为 SQLAlchemy 是一个常用的数据库抽象层和数据库关系映射包(ORM),并且需要 一点点设置才可以使用,因此存在一个 Flask 扩展帮助您操作它。 如果您想要快速 开始使用,那么我们建议您使用这种方法。

您可以从PyPI下载Flask-SQLAlchemy

声明性

SQLAlchemy 中的 declarative 扩展是最新的使用 SQLAlchemy 的方法。 它允许您 同时定义表和模型,就像 Django 一样工作。 除了以下文本,我推荐关于声明式扩展的官方文档。

下面是您应用程序的示例database.py

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

engine = create_engine('sqlite:////tmp/test.db', convert_unicode=True)
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 类的子类。 如果您好奇 为何我们在这里不用担心多线程的问题(就像我们在先前使用 g 对象操作 SQLite3 的例子一样):那是因为 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 '<User %r>' % (self.name)

您可以使用 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 u'admin'>]
>>> User.query.filter(User.name == 'admin').first()
<User u'admin'>

手动对象关系映射

手动实现 ORM (对象关系映射) 相比前面的显式调用方法,既有一些优点,也有一些缺点。 主要差别在于这里的数据表和模型是分开定义的,然后再将其映射起来。 它更灵活,但多一点类型。一般来说,它的工作方式类似于声明式方法,因此请确保将应用程序拆分为多个模块。

以下是您的应用程序的示例database.py模块:

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

engine = create_engine('sqlite:////tmp/test.db', convert_unicode=True)
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 '<User %r>' % (self.name)

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', convert_unicode=True)
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, u'admin', u'admin@localhost')

返回的结果也是字典样式的元组:

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

您也可以将 SQL 语句的字符串传入到 execute() 函数中:

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

有关SQLAlchemy的更多信息,请转到网站