用法

在项目中使用 oslo.db

会话处理

会话处理是通过 oslo_db.sqlalchemy.enginefacade 系统实现的。该模块提供了一个函数装饰器以及上下文管理器方法，以提供 session.Session 以及 Connection 对象给一个函数或代码块。

两种调用方式都需要使用上下文对象。该对象可以是任何类，但当与装饰器形式一起使用时，需要特殊的工具化处理。

上下文管理器形式如下

from oslo_db.sqlalchemy import enginefacade


class MyContext(object):
    "User-defined context class."


def some_reader_api_function(context):
    with enginefacade.reader.using(context) as session:
        return session.query(SomeClass).all()


def some_writer_api_function(context, x, y):
    with enginefacade.writer.using(context) as session:
        session.add(SomeClass(x, y))


def run_some_database_calls():
    context = MyContext()

    results = some_reader_api_function(context)
    some_writer_api_function(context, 5, 10)

装饰器形式直接从用户定义的上下文中访问属性；上下文必须使用 oslo_db.sqlalchemy.enginefacade.transaction_context_provider() 装饰器进行装饰。每个函数必须接收上下文参数

from oslo_db.sqlalchemy import enginefacade

@enginefacade.transaction_context_provider
class MyContext(object):
    "User-defined context class."

@enginefacade.reader
def some_reader_api_function(context):
    return context.session.query(SomeClass).all()


@enginefacade.writer
def some_writer_api_function(context, x, y):
    context.session.add(SomeClass(x, y))


def run_some_database_calls():
    context = MyContext()

    results = some_reader_api_function(context)
    some_writer_api_function(context, 5, 10)

connection 修饰符可以在不需要 session.Session 对象时使用，例如，当首选 SQLAlchemy Core 时

@enginefacade.reader.connection
def _refresh_from_db(context, cache):
    sel = sa.select(table.c.id, table.c.name)
    res = context.connection.execute(sel).fetchall()
    cache.id_cache = {r[1]: r[0] for r in res}
    cache.str_cache = {r[0]: r[1] for r in res}

注意

在适当的读写块（无论是装饰器还是上下文管理器方法）的作用域内，必须访问 context.session 和 context.connection 属性。否则会引发 AttributeError。

装饰器形式也可以与类和实例方法一起使用，这些方法隐式地接收第一个位置参数

class DatabaseAccessLayer(object):

    @classmethod
    @enginefacade.reader
    def some_reader_api_function(cls, context):
        return context.session.query(SomeClass).all()

    @enginefacade.writer
    def some_writer_api_function(self, context, x, y):
        context.session.add(SomeClass(x, y))

注意

请注意，enginefacade 装饰器必须在 classmethod 之前 应用，否则您会在导入时收到 TypeError（因为 enginefacade 会尝试在描述符而不是绑定方法上使用 inspect.getargspec()，有关详细信息，请参阅 Python 语言参考的 数据模型 部分）。

两种方法的事务和连接范围都由系统透明地管理。连接的配置来自标准的 oslo_config.cfg.CONF 集合。可以使用 oslo_db.sqlalchemy.enginefacade.configure() 函数为 enginefacade 建立额外的配置，在开始使用数据库之前

from oslo_db.sqlalchemy import enginefacade

enginefacade.configure(
    sqlite_fk=True,
    max_retries=5,
    mysql_sql_mode='ANSI'
)

模型使用的基础类

from oslo_db.sqlalchemy import models


class ProjectSomething(models.TimestampMixin,
                       models.ModelBase):
    id = Column(Integer, primary_key=True)
    ...

DB API 后端支持

from oslo_config import cfg
from oslo_db import api as db_api


_BACKEND_MAPPING = {'sqlalchemy': 'project.db.sqlalchemy.api'}

IMPL = db_api.DBAPI.from_config(cfg.CONF, backend_mapping=_BACKEND_MAPPING)

def get_engine():
    return IMPL.get_engine()

def get_session():
    return IMPL.get_session()

# DB-API method
def do_something(somethind_id):
    return IMPL.do_something(somethind_id)

DB 迁移扩展

适用于 oslo_db.migration 的可用扩展。