flask-sqlalchemy-core 0.1.1

设置

import os
from flask_sqlalchemy_core import FlaskSQLAlchemy

DATABASE_URL = os.environ['DATABASE_URL']

db = FlaskSQLAlchemy(DATABASE_URL)

运行查询

from sqlalchemy import select

# Create your query here
query = select(...)

with db.connect() as conn:
    result = conn.execute(query)
    # Do something with the result...

事务

with db.transaction() as conn:
    result = conn.execute(query)

在“with”块成功退出时，事务将自动提交，如果抛出异常，则回滚。

嵌套事务

只需嵌套 with db.transaction(): 块。

这允许代码更易于重用，例如

def create_user(...):
    with db.transaction() as conn:
        # Create record in the users table
        conn.execute(...)
        # Other data for the user in some other table
        conn.execute(...)

def create_client(...):
    with db.transaction() as conn:
        # Create record in the clients table
        conn.execute(...)
        # ...other data for this client...
        conn.execute(...)

def setup_new_client(client_name, user_name):
    with db.transaction():
        create_user(user_name)
        create_client(client_name)

定义表

就像您通常那样做（创建一个Metadata实例，使用它来定义您的模式）。

创建模式

metadata.create_all(db.get_engine())

测试固定值

对于与pytest一起使用，将这些放在测试目录中的 conftest.py 文件中。

注意：您可能希望在测试期间更改您的 DATABASE_URL 环境变量，以避免覆盖当前的开发数据库。

import pytest

@pytest.fixture
def db(db_schema):
    with db.transaction(autocommit=False, rollback=True):
        # By wrapping execution in a transaction that automatically
        # gets rolled back, we can avoid having to recreate the whole
        # schema for every test function run.
        yield

@pytest.fixture(scope='session')
def db_schema():
    engine = db.get_engine()

    # Clean up, in case tables were left around from a previous run.
    # This can happen if the test process was abruptly killed.
    metadata.drop_all(engine)

    metadata.create_all(engine)

    yield

    metadata.drop_all(engine)

数据库迁移

使用 Alembic 创建数据库迁移。