SQLAlchemyProxy

类源码

db_proxy = SQLAlchemyProxy[_SQLAlchemyLiteDB_co, _ModelLite_co](
    db: _SQLAlchemyLiteDB_co,
    model_class: type[_ModelLite_co],
)

flask_sqlalchemy.SQLAlchemy的代理类。

该类是针对SQLAlchemy-Lite的封装。透过调整flask_sqlalchemy_lite.SQLAlchemy，来模拟flask_sqlalchemy.SQLAlchemy的用法。

须知并非所有来自flask_sqlalchemy.SQLAlchemy都能精准复刻。准确来说，该代理只完成了以下工作：

按照给定的初始化参数model_class，提供一个Model类，其__tablename__属性可以从model_class自动推断。
提供一个假的Query类。该Query类其实就是sqlalchemy.orm.Query。换言之，其不会提供任何额外的功能。
提供一个假的Table类。这个假类实质上是一个创建sqlalchemy.Table实例的工厂函数。使用该函数的好处是，用户可以略过提供metadata的步骤。
像flask_sqlalchemy.SQLAlchemy那样，提供一个域内会话（scoped session）。该域内会话由该代理类的实例直接管理。
提供flask_sqlalchemy.SQLAlchemy的主要、基本的功能。用法完全等价，但具体的实现是透过flask_sqlalchemy_lite完成的。
任何无法复刻的功能，都会抛出NotImplementedError。

若用户希望从Flask SQLAlchemy迁移到Flask SQLAlchemy-Lite，又不想过多地改动与对象关系映射（ORM）有关的代码，则可以考虑使用该封装。还有一种可用的场景，是用户主要使用Flask SQLAlchemy开发、却需要在某些特殊的部署条件下，使用Flask SQLAlchemy Lite。

别名

该类可以按以下方式之一获取

import flask_sqlalchemy_compat as fsc


fsc.SQLAlchemyProxy
fsc.flask_sa_api.SQLAlchemyProxy

参数

参数类型必选说明

参数	类型	必选	说明
`db`	`_SQLAlchemyLiteDB_co`		用来导出session和engine实例的数据库扩展。由`flask_sqlachemy_lite`提供。封装该参数后，可以在参数来自lite版的情况下、模拟`flask_sqlalchemy`的行为。
`model_class`	`_ModelLite_co`		基模型的类。该`model_class`参数是必需的。该参数将会用来提供一些与metadata有关的功能。

db

_SQLAlchemyLiteDB_co

用来导出session和engine实例的数据库扩展。

由flask_sqlachemy_lite提供。封装该参数后，可以在参数来自lite版的情况下、模拟flask_sqlalchemy的行为。

model_class

_ModelLite_co

基模型的类。该model_class参数是必需的。

该参数将会用来提供一些与metadata有关的功能。

其中，_SQLAlchemyLiteDB_co是fsc.protocols.SQLAlchemyLiteProtocol的TypeVar。
其中，_ModelLite_co是sa.orm.DeclarativeBase的TypeVar。

方法

`init_app`

db_proxy.init_app(
    app: Flask,
)

按照Flask.config创建engine，并将数据库扩展注册到应用中。

输入

参数	类型	必选	说明
`app`	`Flask`		要注册到的应用。

`create_all`

db_proxy.create_all(
    bind_key: str | None | Sequence[str | None] = "__all__",
)

透过调用metadata.create_all()，在数据表不存在的情况下，在数据库中、为某个、某些、或全部bind_key创建数据表。该操作不会更新那些已经存在的数据表。如果用户需要这种更新，应当使用迁移（migration）库。

该方法需要在启动的Flask应用上下文之内调用。

输入

参数类型必选说明

参数	类型	必选	说明
`bind_key`	`str \| None \| Sequence[str \| None]`		一个或多个、要设置metadata的engine的名称。如果该值为`__all__`，则会向配置中的所有engine设置metadata。

bind_key

str | None | Sequence[str | None]

一个或多个、要设置metadata的engine的名称。

如果该值为__all__，则会向配置中的所有engine设置metadata。

`drop_all`

db_proxy.create_all(
    bind_key: str | None | Sequence[str | None] = "__all__",
)

透过调用metadata.drop_all()、为某个、某些、或全部bind_key删除所有数据表。

该方法需要在启动的Flask应用上下文之内调用。

输入

参数类型必选说明

参数	类型	必选	说明
`bind_key`	`str \| None \| Sequence[str \| None]`		一个或多个、要删去数据表的engine的名称。如果该值为`__all__`，则会令配置中的所有engine删去其数据表。

bind_key

str | None | Sequence[str | None]

一个或多个、要删去数据表的engine的名称。

如果该值为__all__，则会令配置中的所有engine删去其数据表。

`reflect`

db_proxy.reflect(
    bind_key: str | None | Sequence[str | None] = "__all__",
)

透过调用metadata.reflect()，为某个、某些、或全部bind_key从数据库本身加载数据表的定义。

该方法需要在启动的Flask应用上下文之内调用。

输入

参数类型必选说明

参数	类型	必选	说明
`bind_key`	`str \| None \| Sequence[str \| None]`		一个或多个、要反射数据表的engine的名称。如果该值为`__all__`，则会令配置中的所有engine反射其数据表。

bind_key

str | None | Sequence[str | None]

一个或多个、要反射数据表的engine的名称。

如果该值为__all__，则会令配置中的所有engine反射其数据表。

`dynamic_loader`

val: RelationshipProperty[Any] = db_proxy.dynamic_loader(
    argument: _RelationshipArgumentType[Any] | None = None,
    **kw: Any,
)

和sa.orm.relationship相同。

用来构建动态载入的mapper属性。

本质上，它和使用relationship的时候、设置lazy="dynamic"是一样的。

dynamic_loader(SomeClass)  # 等同于

relationship(SomeClass, lazy="dynamic")

输入

参数	类型	必选	说明
`argument`	`_RelationshipArgumentType[Any]`		第一个要传递到`sa.orm.relationship(...)`的参数。
`**kw`	`Any`		其他要传递到`sa.orm.relationship(...)`的参数。

输出

参数	类型	说明
`val`	`RelationshipProperty[Any]`	动态载入的关系属性。

`get_engine`

engine: sa.engine.Engine = db_proxy.reflect(
    bind_key: str | None = None,
)

和self.engines[bind_key]相同。

按照指定的bind_key来获取当前应用对应关联的engine。该方法需要在启动的Flask应用上下文之内调用。

输入

参数	类型	必选	说明
`bind_key`	`str \| None`		engine的名字。

输出

参数	类型	说明
`engine`	`sa.engine.Engine`	按照名字获取到的engine。

`get_or_404`

val: T = db_proxy.get_or_404[T](
    entity: _EntityBindKey[T],
    ident: _PKIdentityArgument,
    *,
    description: str | None = None,
    **kwargs: Any,
)

与session.get()相若，但不会返回None，而是会抛出404 Not Found错误、并拒绝请求。

除了该方法的特有参数description，其他的参数、用法与session.get()相同。

输入

参数	类型	说明
`entity`	`_EntityBindKey[T]`	要检索的模型类、或实体语句。
`ident`	`_PKIdentityArgument`	要检索的主键。
`description`	`str \| None`	用来在抛出错误时、显示的定制信息。
`**kwargs`	`Any`	其他传递到`session.get()`的参数。

输出

参数	类型	说明
`val`	`T`	检索到的模型实体。

其中，T是某个Model的TypeVar。该Model可以是一个sa.orm.DeclarativeBase的子类。
其中，_EntityBindKey[T]通常是Type[T]。
其中，_PKIdentityArgument是Any | tuple[Any]的别名。

`first_or_404`

val: T = db_proxy.first_or_404[T](
    statement: sa.sql.Select[tuple[T]],
    *,
    description: str | None = None,
)
val: tuple[T, T2, ...] = db_proxy.first_or_404[T, T2, ...](
    statement: sa.sql.Select[tuple[T, T2, ...]],
    *,
    description: str | None = None,
)

与Result.scalar()相若，但不会返回None，而是会抛出404 Not Found错误、并拒绝请求。

输入

参数	类型	必选	说明
`statement`	`sa.sql.Select[Tuple[T, ...]]`		要执行的`select`语句。
`description`	`str \| None`		用来在抛出错误时、显示的定制信息。

输出

参数	类型	说明
`val`	`T, ...`	当`statement`检索到时，所返回的第一个结果。

其中，T, T2, ...是sa.select语句中指定的类型。

`one_or_404`

val: T = db_proxy.one_or_404[T](
    statement: sa.sql.Select[tuple[T]],
    *,
    description: str | None = None,
)
val: tuple[T, T2, ...] = db_proxy.one_or_404[T, T2, ...](
    statement: sa.sql.Select[tuple[T, T2, ...]],
    *,
    description: str | None = None,
)

与Result.scalar_one()相若，但不会抛出NoResultFound或MultipleResultsFound，而是会抛出404 Not Found错误、并拒绝请求。

输入

参数	类型	必选	说明
`statement`	`sa.sql.Select[Tuple[T, ...]]`		要执行的`select`语句。
`description`	`str \| None`		用来在抛出错误时、显示的定制信息。

输出

参数	类型	说明
`val`	`T, ...`	当`statement`检索到时，所返回的第一个、且唯一一个结果。

其中，T, T2, ...是sa.select语句中指定的类型。

`paginate`

db_proxy.paginate(*args: Any, **kwargs: Any)

危险

该代理类并未实现该方法。

`relationship`

val: _RelationshipDeclared[Any] = db_proxy.relationship(
    argument: _RelationshipArgumentType[Any] | None = None,
    secondary: _RelationshipSecondaryArgument | None = None,
    *,
    uselist: bool | None = None,
    collection_class: type[Collection[Any]] | Callable[[], Collection[Any]] = None,
    primaryjoin: _RelationshipJoinConditionArgument | None = None,
    secondaryjoin: _RelationshipJoinConditionArgument | None = None,
    back_populates: str | None = None,
    order_by: _ORMOrderByArgument = False,
    backref: ORMBackrefArgument | None = None,
    overlaps: str | None = None,
    post_update: bool = False,
    cascade: str = "save-update, merge",
    viewonly: bool = False,
    init: _NoArg | bool = _NoArg.NO_ARG,
    repr: _NoArg | bool = _NoArg.NO_ARG,
    default: _NoArg | _T = _NoArg.NO_ARG,
    default_factory: _NoArg | Callable[[], _T] = _NoArg.NO_ARG,
    compare: _NoArg | bool = _NoArg.NO_ARG,
    kw_only: _NoArg | bool = _NoArg.NO_ARG,
    hash: _NoArg | bool | None = _NoArg.NO_ARG,
    lazy: _LazyLoadArgumentType = "select",
    passive_deletes: Literal["all"] | bool = False,
    passive_updates: bool = True,
    active_history: bool = False,
    enable_typechecks: bool = True,
    foreign_keys: _ORMColCollectionArgument | None = None,
    remote_side: _ORMColCollectionArgument | None = None,
    join_depth: int | None = None,
    comparator_factory: type[RelationshipProperty.Comparator[Any]] | None = None,
    single_parent: bool = False,
    innerjoin: bool = False,
    distinct_target_key: bool | None = None,
    load_on_pending: bool = False,
    query_class: type[Query[Any]] | None = None,
    info: _InfoType | None = None,
    omit_join: Literal[None, False] = None,
    sync_backref: bool | None = None,
    **kw: Any,
)

与sa.orm.relationship相同，用来提供两个映射类之间的关系。

详情请参见官方文档。

属性

`db`

db: _SQLAlchemyLiteDB_co = db_proxy.db

Flask SQLAlchemy扩展所提供的db实例。

`Model`

db: type[_ModelLite_co] = db_proxy.Model

Model类。定义SQLAlchemy模型时，充当基类。该值与初始化参数model_class是同一值，但是为了支持额外功能、经过了修改。

`Table`

table_factory: Callable[[...], sa.Table] = db_proxy.Table

sa.Table的简化版。

相比sa.Table，唯一的不同是，参数metadata可以隐式配置。用户使用该工厂函数构造数据表时，可以不设置该参数。

`Query`

query_class: sa.orm.Query = db_proxy.Query

和sa.orm.Query相同。

ORM层的SQL语句构造器。

`session`

sess: sa.orm.scoped_session[sa.orm.Session] = db_proxy.session

和self.db.session用法相若。

当前应用上下文的默认会话。在离开应用上下文时，会自动关闭。

`metadatas`

metadatas: dict[str | None, sa.MetaData] = db_proxy.metadatas

metadata的兼容接口。

由于flask_sqlalchemy_lite并未支持bind_key，目前metdatas总是实现为仅含一个成员的字典，其形式如下：

{None: self.metadata}

`metadata`

metadata: sa.MetaData = db_proxy.metadata

未配置bind key的情况下，Model和Table的默认metadata。

`engines`

engines: MutableMapping[str | None, sa.engine.Engine] = db_proxy.engines

配置在当前应用下的所有engine。

为确保该值和flask_sqlalchemy用法兼容。所返回的值添加了一个额外项：self.engines[None] is self.engines["default"]。

`engine`

engine: sa.engine.Engine = db_proxy.engine

配置在当前应用下的默认engine。

运算符

`getattr`

obj: Any = getattr(
    db_proxy,
    any_attr: str
)

当需要向db_proxy获取非显式定义的属性（方法）时调用。

会依次向sa.orm和sa搜索属性。若所要的属性不存在于sa.orm，则在sa中搜索。

特殊情况包含：

属性	值
`db_proxy.event`	`sa.event`
`db_proxy.relation`	`sa.orm.relationship`

例如，

db_proxy.select

即等同于调用sa.select。

输入

参数	类型	必选	说明
`any_attr`	`str`		要获取的属性名

输出

参数	类型	说明
`obj`	`Any`	在`sa.orm`或`sa`找到的属性。

别名​

参数​

方法​

init_app​

输入​

create_all​

输入​

drop_all​

输入​

reflect​

输入​

dynamic_loader​

输入​

输出​

get_engine​

输入​

输出​

get_or_404​

输入​

输出​

first_or_404​

输入​

输出​

one_or_404​

输入​

输出​

paginate​

relationship​

属性​

db​

Model​

Table​

Query​

session​

metadatas​

metadata​

engines​

engine​

运算符​

__getattr__​

输入​

输出​

别名

参数

方法

`init_app`

输入

`create_all`

输入

`drop_all`

输入

`reflect`

输入

`dynamic_loader`

输入

输出

`get_engine`

输入

输出

`get_or_404`

输入

输出

`first_or_404`

输入

输出

`one_or_404`

输入

输出

`paginate`

`relationship`

属性

`db`

`Model`

`Table`

`Query`

`session`

`metadatas`

`metadata`

`engines`

`engine`

运算符

`getattr`

输入

输出