Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/ext»

Browse Source
This commit is contained in:
Polina
2026-07-02 20:29:54 +00:00
parent ae4f5f5238
commit 3de8f46cb4
5 changed files with 3913 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

479
venv/Lib/site-packages/sqlalchemy/ext/horizontal_shard.py Normal file
View File

@@ -0,0 +1,479 @@
# ext/horizontal_shard.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
"""Horizontal sharding support.
Defines a rudimental 'horizontal sharding' system which allows a Session to
distribute queries and persistence operations across multiple databases.
For a usage example, see the :ref:`examples_sharding` example included in
the source distribution.
.. deepalchemy:: The horizontal sharding extension is an advanced feature,
involving a complex statement -> database interaction as well as
use of semi-public APIs for non-trivial cases. Simpler approaches to
referring to multiple database "shards", most commonly using a distinct
:class:`_orm.Session` per "shard", should always be considered first
before using this more complex and less-production-tested system.
"""
from __future__ import annotations
from typing import Any
from typing import Callable
from typing import Dict
from typing import Iterable
from typing import Optional
from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
from .. import event
from .. import exc
from .. import inspect
from .. import util
from ..orm import PassiveFlag
from ..orm._typing import OrmExecuteOptionsParameter
from ..orm.interfaces import ORMOption
from ..orm.mapper import Mapper
from ..orm.query import Query
from ..orm.session import _BindArguments
from ..orm.session import _PKIdentityArgument
from ..orm.session import Session
from ..util.typing import Protocol
from ..util.typing import Self
if TYPE_CHECKING:
from ..engine.base import Connection
from ..engine.base import Engine
from ..engine.base import OptionEngine
from ..engine.result import IteratorResult
from ..engine.result import Result
from ..orm import LoaderCallableStatus
from ..orm._typing import _O
from ..orm.bulk_persistence import BulkUDCompileState
from ..orm.context import QueryContext
from ..orm.session import _EntityBindKey
from ..orm.session import _SessionBind
from ..orm.session import ORMExecuteState
from ..orm.state import InstanceState
from ..sql import Executable
from ..sql._typing import _TP
from ..sql.elements import ClauseElement
__all__ = ["ShardedSession", "ShardedQuery"]
_T = TypeVar("_T", bound=Any)
ShardIdentifier = str
class ShardChooser(Protocol):
def __call__(
self,
mapper: Optional[Mapper[_T]],
instance: Any,
clause: Optional[ClauseElement],
) -> Any: ...
class IdentityChooser(Protocol):
def __call__(
self,
mapper: Mapper[_T],
primary_key: _PKIdentityArgument,
*,
lazy_loaded_from: Optional[InstanceState[Any]],
execution_options: OrmExecuteOptionsParameter,
bind_arguments: _BindArguments,
**kw: Any,
) -> Any: ...
class ShardedQuery(Query[_T]):
"""Query class used with :class:`.ShardedSession`.
.. legacy:: The :class:`.ShardedQuery` is a subclass of the legacy
:class:`.Query` class. The :class:`.ShardedSession` now supports
2.0 style execution via the :meth:`.ShardedSession.execute` method.
"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
assert isinstance(self.session, ShardedSession)
self.identity_chooser = self.session.identity_chooser
self.execute_chooser = self.session.execute_chooser
self._shard_id = None
def set_shard(self, shard_id: ShardIdentifier) -> Self:
"""Return a new query, limited to a single shard ID.
All subsequent operations with the returned query will
be against the single shard regardless of other state.
The shard_id can be passed for a 2.0 style execution to the
bind_arguments dictionary of :meth:`.Session.execute`::
results = session.execute(stmt, bind_arguments={"shard_id": "my_shard"})
""" # noqa: E501
return self.execution_options(_sa_shard_id=shard_id)
class ShardedSession(Session):
shard_chooser: ShardChooser
identity_chooser: IdentityChooser
execute_chooser: Callable[[ORMExecuteState], Iterable[Any]]
def __init__(
self,
shard_chooser: ShardChooser,
identity_chooser: Optional[IdentityChooser] = None,
execute_chooser: Optional[
Callable[[ORMExecuteState], Iterable[Any]]
] = None,
shards: Optional[Dict[str, Any]] = None,
query_cls: Type[Query[_T]] = ShardedQuery,
*,
id_chooser: Optional[
Callable[[Query[_T], Iterable[_T]], Iterable[Any]]
] = None,
query_chooser: Optional[Callable[[Executable], Iterable[Any]]] = None,
**kwargs: Any,
) -> None:
"""Construct a ShardedSession.
:param shard_chooser: A callable which, passed a Mapper, a mapped
instance, and possibly a SQL clause, returns a shard ID. This id
may be based off of the attributes present within the object, or on
some round-robin scheme. If the scheme is based on a selection, it
should set whatever state on the instance to mark it in the future as
participating in that shard.
:param identity_chooser: A callable, passed a Mapper and primary key
argument, which should return a list of shard ids where this
primary key might reside.
.. versionchanged:: 2.0 The ``identity_chooser`` parameter
supersedes the ``id_chooser`` parameter.
:param execute_chooser: For a given :class:`.ORMExecuteState`,
returns the list of shard_ids
where the query should be issued. Results from all shards returned
will be combined together into a single listing.
.. versionchanged:: 1.4 The ``execute_chooser`` parameter
supersedes the ``query_chooser`` parameter.
:param shards: A dictionary of string shard names
to :class:`~sqlalchemy.engine.Engine` objects.
"""
super().__init__(query_cls=query_cls, **kwargs)
event.listen(
self, "do_orm_execute", execute_and_instances, retval=True
)
self.shard_chooser = shard_chooser
if id_chooser:
_id_chooser = id_chooser
util.warn_deprecated(
"The ``id_chooser`` parameter is deprecated; "
"please use ``identity_chooser``.",
"2.0",
)
def _legacy_identity_chooser(
mapper: Mapper[_T],
primary_key: _PKIdentityArgument,
*,
lazy_loaded_from: Optional[InstanceState[Any]],
execution_options: OrmExecuteOptionsParameter,
bind_arguments: _BindArguments,
**kw: Any,
) -> Any:
q = self.query(mapper)
if lazy_loaded_from:
q = q._set_lazyload_from(lazy_loaded_from)
return _id_chooser(q, primary_key)
self.identity_chooser = _legacy_identity_chooser
elif identity_chooser:
self.identity_chooser = identity_chooser
else:
raise exc.ArgumentError(
"identity_chooser or id_chooser is required"
)
if query_chooser:
_query_chooser = query_chooser
util.warn_deprecated(
"The ``query_chooser`` parameter is deprecated; "
"please use ``execute_chooser``.",
"1.4",
)
if execute_chooser:
raise exc.ArgumentError(
"Can't pass query_chooser and execute_chooser "
"at the same time."
)
def _default_execute_chooser(
orm_context: ORMExecuteState,
) -> Iterable[Any]:
return _query_chooser(orm_context.statement)
if execute_chooser is None:
execute_chooser = _default_execute_chooser
if execute_chooser is None:
raise exc.ArgumentError(
"execute_chooser or query_chooser is required"
)
self.execute_chooser = execute_chooser
self.__shards: Dict[ShardIdentifier, _SessionBind] = {}
if shards is not None:
for k in shards:
self.bind_shard(k, shards[k])
def _identity_lookup(
self,
mapper: Mapper[_O],
primary_key_identity: Union[Any, Tuple[Any, ...]],
identity_token: Optional[Any] = None,
passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
lazy_loaded_from: Optional[InstanceState[Any]] = None,
execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Union[Optional[_O], LoaderCallableStatus]:
"""override the default :meth:`.Session._identity_lookup` method so
that we search for a given non-token primary key identity across all
possible identity tokens (e.g. shard ids).
.. versionchanged:: 1.4 Moved :meth:`.Session._identity_lookup` from
the :class:`_query.Query` object to the :class:`.Session`.
"""
if identity_token is not None:
obj = super()._identity_lookup(
mapper,
primary_key_identity,
identity_token=identity_token,
**kw,
)
return obj
else:
for shard_id in self.identity_chooser(
mapper,
primary_key_identity,
lazy_loaded_from=lazy_loaded_from,
execution_options=execution_options,
bind_arguments=dict(bind_arguments) if bind_arguments else {},
):
obj2 = super()._identity_lookup(
mapper,
primary_key_identity,
identity_token=shard_id,
lazy_loaded_from=lazy_loaded_from,
**kw,
)
if obj2 is not None:
return obj2
return None
def _choose_shard_and_assign(
self,
mapper: Optional[_EntityBindKey[_O]],
instance: Any,
**kw: Any,
) -> Any:
if instance is not None:
state = inspect(instance)
if state.key:
token = state.key[2]
assert token is not None
return token
elif state.identity_token:
return state.identity_token
assert isinstance(mapper, Mapper)
shard_id = self.shard_chooser(mapper, instance, **kw)
if instance is not None:
state.identity_token = shard_id
return shard_id
def connection_callable(
self,
mapper: Optional[Mapper[_T]] = None,
instance: Optional[Any] = None,
shard_id: Optional[ShardIdentifier] = None,
**kw: Any,
) -> Connection:
"""Provide a :class:`_engine.Connection` to use in the unit of work
flush process.
"""
if shard_id is None:
shard_id = self._choose_shard_and_assign(mapper, instance)
if self.in_transaction():
trans = self.get_transaction()
assert trans is not None
return trans.connection(mapper, shard_id=shard_id)
else:
bind = self.get_bind(
mapper=mapper, shard_id=shard_id, instance=instance
)
if isinstance(bind, Engine):
return bind.connect(**kw)
else:
assert isinstance(bind, Connection)
return bind
def get_bind(
self,
mapper: Optional[_EntityBindKey[_O]] = None,
*,
shard_id: Optional[ShardIdentifier] = None,
instance: Optional[Any] = None,
clause: Optional[ClauseElement] = None,
**kw: Any,
) -> _SessionBind:
if shard_id is None:
shard_id = self._choose_shard_and_assign(
mapper, instance=instance, clause=clause
)
assert shard_id is not None
return self.__shards[shard_id]
def bind_shard(
self, shard_id: ShardIdentifier, bind: Union[Engine, OptionEngine]
) -> None:
self.__shards[shard_id] = bind
class set_shard_id(ORMOption):
"""a loader option for statements to apply a specific shard id to the
primary query as well as for additional relationship and column
loaders.
The :class:`_horizontal.set_shard_id` option may be applied using
the :meth:`_sql.Executable.options` method of any executable statement::
stmt = (
select(MyObject)
.where(MyObject.name == "some name")
.options(set_shard_id("shard1"))
)
Above, the statement when invoked will limit to the "shard1" shard
identifier for the primary query as well as for all relationship and
column loading strategies, including eager loaders such as
:func:`_orm.selectinload`, deferred column loaders like :func:`_orm.defer`,
and the lazy relationship loader :func:`_orm.lazyload`.
In this way, the :class:`_horizontal.set_shard_id` option has much wider
scope than using the "shard_id" argument within the
:paramref:`_orm.Session.execute.bind_arguments` dictionary.
.. versionadded:: 2.0.0
"""
__slots__ = ("shard_id", "propagate_to_loaders")
def __init__(
self, shard_id: ShardIdentifier, propagate_to_loaders: bool = True
):
"""Construct a :class:`_horizontal.set_shard_id` option.
:param shard_id: shard identifier
:param propagate_to_loaders: if left at its default of ``True``, the
shard option will take place for lazy loaders such as
:func:`_orm.lazyload` and :func:`_orm.defer`; if False, the option
will not be propagated to loaded objects. Note that :func:`_orm.defer`
always limits to the shard_id of the parent row in any case, so the
parameter only has a net effect on the behavior of the
:func:`_orm.lazyload` strategy.
"""
self.shard_id = shard_id
self.propagate_to_loaders = propagate_to_loaders
def execute_and_instances(
orm_context: ORMExecuteState,
) -> Union[Result[_T], IteratorResult[_TP]]:
active_options: Union[
None,
QueryContext.default_load_options,
Type[QueryContext.default_load_options],
BulkUDCompileState.default_update_options,
Type[BulkUDCompileState.default_update_options],
]
if orm_context.is_select:
active_options = orm_context.load_options
elif orm_context.is_update or orm_context.is_delete:
active_options = orm_context.update_delete_options
else:
active_options = None
session = orm_context.session
assert isinstance(session, ShardedSession)
def iter_for_shard(
shard_id: ShardIdentifier,
) -> Union[Result[_T], IteratorResult[_TP]]:
bind_arguments = dict(orm_context.bind_arguments)
bind_arguments["shard_id"] = shard_id
orm_context.update_execution_options(identity_token=shard_id)
return orm_context.invoke_statement(bind_arguments=bind_arguments)
for orm_opt in orm_context._non_compile_orm_options:
# TODO: if we had an ORMOption that gets applied at ORM statement
# execution time, that would allow this to be more generalized.
# for now just iterate and look for our options
if isinstance(orm_opt, set_shard_id):
shard_id = orm_opt.shard_id
break
else:
if active_options and active_options._identity_token is not None:
shard_id = active_options._identity_token
elif "_sa_shard_id" in orm_context.execution_options:
shard_id = orm_context.execution_options["_sa_shard_id"]
elif "shard_id" in orm_context.bind_arguments:
shard_id = orm_context.bind_arguments["shard_id"]
else:
shard_id = None
if shard_id is not None:
return iter_for_shard(shard_id)
else:
partial = []
for shard_id in session.execute_chooser(orm_context):
result_ = iter_for_shard(shard_id)
partial.append(result_)
return partial[0].merge(*partial[1:])

1535
venv/Lib/site-packages/sqlalchemy/ext/hybrid.py Normal file
View File

File diff suppressed because it is too large Load Diff

364
venv/Lib/site-packages/sqlalchemy/ext/indexable.py Normal file
View File

@@ -0,0 +1,364 @@
# ext/indexable.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
"""Define attributes on ORM-mapped classes that have "index" attributes for
columns with :class:`_types.Indexable` types.
"index" means the attribute is associated with an element of an
:class:`_types.Indexable` column with the predefined index to access it.
The :class:`_types.Indexable` types include types such as
:class:`_types.ARRAY`, :class:`_types.JSON` and
:class:`_postgresql.HSTORE`.
The :mod:`~sqlalchemy.ext.indexable` extension provides
:class:`_schema.Column`-like interface for any element of an
:class:`_types.Indexable` typed column. In simple cases, it can be
treated as a :class:`_schema.Column` - mapped attribute.
Synopsis
========
Given ``Person`` as a model with a primary key and JSON data field.
While this field may have any number of elements encoded within it,
we would like to refer to the element called ``name`` individually
as a dedicated attribute which behaves like a standalone column::
from sqlalchemy import Column, JSON, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.indexable import index_property
Base = declarative_base()
class Person(Base):
__tablename__ = "person"
id = Column(Integer, primary_key=True)
data = Column(JSON)
name = index_property("data", "name")
Above, the ``name`` attribute now behaves like a mapped column. We
can compose a new ``Person`` and set the value of ``name``::
>>> person = Person(name="Alchemist")
The value is now accessible::
>>> person.name
'Alchemist'
Behind the scenes, the JSON field was initialized to a new blank dictionary
and the field was set::
>>> person.data
{'name': 'Alchemist'}
The field is mutable in place::
>>> person.name = "Renamed"
>>> person.name
'Renamed'
>>> person.data
{'name': 'Renamed'}
When using :class:`.index_property`, the change that we make to the indexable
structure is also automatically tracked as history; we no longer need
to use :class:`~.mutable.MutableDict` in order to track this change
for the unit of work.
Deletions work normally as well::
>>> del person.name
>>> person.data
{}
Above, deletion of ``person.name`` deletes the value from the dictionary,
but not the dictionary itself.
A missing key will produce ``AttributeError``::
>>> person = Person()
>>> person.name
AttributeError: 'name'
Unless you set a default value::
>>> class Person(Base):
... __tablename__ = "person"
...
... id = Column(Integer, primary_key=True)
... data = Column(JSON)
...
... name = index_property("data", "name", default=None) # See default
>>> person = Person()
>>> print(person.name)
None
The attributes are also accessible at the class level.
Below, we illustrate ``Person.name`` used to generate
an indexed SQL criteria::
>>> from sqlalchemy.orm import Session
>>> session = Session()
>>> query = session.query(Person).filter(Person.name == "Alchemist")
The above query is equivalent to::
>>> query = session.query(Person).filter(Person.data["name"] == "Alchemist")
Multiple :class:`.index_property` objects can be chained to produce
multiple levels of indexing::
from sqlalchemy import Column, JSON, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.indexable import index_property
Base = declarative_base()
class Person(Base):
__tablename__ = "person"
id = Column(Integer, primary_key=True)
data = Column(JSON)
birthday = index_property("data", "birthday")
year = index_property("birthday", "year")
month = index_property("birthday", "month")
day = index_property("birthday", "day")
Above, a query such as::
q = session.query(Person).filter(Person.year == "1980")
On a PostgreSQL backend, the above query will render as:
.. sourcecode:: sql
SELECT person.id, person.data
FROM person
WHERE person.data -> %(data_1)s -> %(param_1)s = %(param_2)s
Default Values
==============
:class:`.index_property` includes special behaviors for when the indexed
data structure does not exist, and a set operation is called:
* For an :class:`.index_property` that is given an integer index value,
the default data structure will be a Python list of ``None`` values,
at least as long as the index value; the value is then set at its
place in the list. This means for an index value of zero, the list
will be initialized to ``[None]`` before setting the given value,
and for an index value of five, the list will be initialized to
``[None, None, None, None, None]`` before setting the fifth element
to the given value. Note that an existing list is **not** extended
in place to receive a value.
* for an :class:`.index_property` that is given any other kind of index
value (e.g. strings usually), a Python dictionary is used as the
default data structure.
* The default data structure can be set to any Python callable using the
:paramref:`.index_property.datatype` parameter, overriding the previous
rules.
Subclassing
===========
:class:`.index_property` can be subclassed, in particular for the common
use case of providing coercion of values or SQL expressions as they are
accessed. Below is a common recipe for use with a PostgreSQL JSON type,
where we want to also include automatic casting plus ``astext()``::
class pg_json_property(index_property):
def __init__(self, attr_name, index, cast_type):
super(pg_json_property, self).__init__(attr_name, index)
self.cast_type = cast_type
def expr(self, model):
expr = super(pg_json_property, self).expr(model)
return expr.astext.cast(self.cast_type)
The above subclass can be used with the PostgreSQL-specific
version of :class:`_postgresql.JSON`::
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.postgresql import JSON
Base = declarative_base()
class Person(Base):
__tablename__ = "person"
id = Column(Integer, primary_key=True)
data = Column(JSON)
age = pg_json_property("data", "age", Integer)
The ``age`` attribute at the instance level works as before; however
when rendering SQL, PostgreSQL's ``->>`` operator will be used
for indexed access, instead of the usual index operator of ``->``::
>>> query = session.query(Person).filter(Person.age < 20)
The above query will render:
.. sourcecode:: sql
SELECT person.id, person.data
FROM person
WHERE CAST(person.data ->> %(data_1)s AS INTEGER) < %(param_1)s
""" # noqa
from __future__ import annotations
from typing import Any
from typing import Callable
from typing import cast
from typing import Optional
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
from .. import inspect
from ..ext.hybrid import hybrid_property
from ..orm.attributes import flag_modified
if TYPE_CHECKING:
from ..sql import SQLColumnExpression
from ..sql._typing import _HasClauseElement
__all__ = ["index_property"]
_T = TypeVar("_T")
class index_property(hybrid_property[_T]):
"""A property generator. The generated property describes an object
attribute that corresponds to an :class:`_types.Indexable`
column.
.. seealso::
:mod:`sqlalchemy.ext.indexable`
"""
_NO_DEFAULT_ARGUMENT = cast(_T, object())
def __init__(
self,
attr_name: str,
index: Union[int, str],
default: _T = _NO_DEFAULT_ARGUMENT,
datatype: Optional[Callable[[], Any]] = None,
mutable: bool = True,
onebased: bool = True,
):
"""Create a new :class:`.index_property`.
:param attr_name:
An attribute name of an `Indexable` typed column, or other
attribute that returns an indexable structure.
:param index:
The index to be used for getting and setting this value. This
should be the Python-side index value for integers.
:param default:
A value which will be returned instead of `AttributeError`
when there is not a value at given index.
:param datatype: default datatype to use when the field is empty.
By default, this is derived from the type of index used; a
Python list for an integer index, or a Python dictionary for
any other style of index. For a list, the list will be
initialized to a list of None values that is at least
``index`` elements long.
:param mutable: if False, writes and deletes to the attribute will
be disallowed.
:param onebased: assume the SQL representation of this value is
one-based; that is, the first index in SQL is 1, not zero.
"""
if mutable:
super().__init__(self.fget, self.fset, self.fdel, self.expr)
else:
super().__init__(self.fget, None, None, self.expr)
self.attr_name = attr_name
self.index = index
self.default = default
is_numeric = isinstance(index, int)
onebased = is_numeric and onebased
if datatype is not None:
self.datatype = datatype
else:
if is_numeric:
self.datatype = lambda: [None for x in range(index + 1)] # type: ignore[operator] # noqa: E501
else:
self.datatype = dict
self.onebased = onebased
def _fget_default(self, err: Optional[BaseException] = None) -> _T:
if self.default == self._NO_DEFAULT_ARGUMENT:
raise AttributeError(self.attr_name) from err
else:
return self.default
def fget(self, __instance: Any) -> _T:
attr_name = self.attr_name
column_value = getattr(__instance, attr_name)
if column_value is None:
return self._fget_default()
try:
value = column_value[self.index]
except (KeyError, IndexError) as err:
return self._fget_default(err)
else:
return value # type: ignore[no-any-return]
def fset(self, instance: Any, value: _T) -> None:
attr_name = self.attr_name
column_value = getattr(instance, attr_name, None)
if column_value is None:
column_value = self.datatype()
setattr(instance, attr_name, column_value)
column_value[self.index] = value
setattr(instance, attr_name, column_value)
if attr_name in inspect(instance).mapper.attrs:
flag_modified(instance, attr_name)
def fdel(self, instance: Any) -> None:
attr_name = self.attr_name
column_value = getattr(instance, attr_name)
if column_value is None:
raise AttributeError(self.attr_name)
try:
del column_value[self.index]
except KeyError as err:
raise AttributeError(self.attr_name) from err
else:
setattr(instance, attr_name, column_value)
flag_modified(instance, attr_name)
def expr(
self, model: Any
) -> Union[_HasClauseElement[_T], SQLColumnExpression[_T]]:
column = getattr(model, self.attr_name)
index = self.index
if self.onebased:
index += 1 # type: ignore[operator]
return column[index] # type: ignore[no-any-return]

450
venv/Lib/site-packages/sqlalchemy/ext/instrumentation.py Normal file
View File

@@ -0,0 +1,450 @@
# ext/instrumentation.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
"""Extensible class instrumentation.
The :mod:`sqlalchemy.ext.instrumentation` package provides for alternate
systems of class instrumentation within the ORM. Class instrumentation
refers to how the ORM places attributes on the class which maintain
data and track changes to that data, as well as event hooks installed
on the class.
.. note::
The extension package is provided for the benefit of integration
with other object management packages, which already perform
their own instrumentation. It is not intended for general use.
For examples of how the instrumentation extension is used,
see the example :ref:`examples_instrumentation`.
"""
import weakref
from .. import util
from ..orm import attributes
from ..orm import base as orm_base
from ..orm import collections
from ..orm import exc as orm_exc
from ..orm import instrumentation as orm_instrumentation
from ..orm import util as orm_util
from ..orm.instrumentation import _default_dict_getter
from ..orm.instrumentation import _default_manager_getter
from ..orm.instrumentation import _default_opt_manager_getter
from ..orm.instrumentation import _default_state_getter
from ..orm.instrumentation import ClassManager
from ..orm.instrumentation import InstrumentationFactory
INSTRUMENTATION_MANAGER = "__sa_instrumentation_manager__"
"""Attribute, elects custom instrumentation when present on a mapped class.
Allows a class to specify a slightly or wildly different technique for
tracking changes made to mapped attributes and collections.
Only one instrumentation implementation is allowed in a given object
inheritance hierarchy.
The value of this attribute must be a callable and will be passed a class
object. The callable must return one of:
- An instance of an :class:`.InstrumentationManager` or subclass
- An object implementing all or some of InstrumentationManager (TODO)
- A dictionary of callables, implementing all or some of the above (TODO)
- An instance of a :class:`.ClassManager` or subclass
This attribute is consulted by SQLAlchemy instrumentation
resolution, once the :mod:`sqlalchemy.ext.instrumentation` module
has been imported. If custom finders are installed in the global
instrumentation_finders list, they may or may not choose to honor this
attribute.
"""
def find_native_user_instrumentation_hook(cls):
"""Find user-specified instrumentation management for a class."""
return getattr(cls, INSTRUMENTATION_MANAGER, None)
instrumentation_finders = [find_native_user_instrumentation_hook]
"""An extensible sequence of callables which return instrumentation
implementations
When a class is registered, each callable will be passed a class object.
If None is returned, the
next finder in the sequence is consulted. Otherwise the return must be an
instrumentation factory that follows the same guidelines as
sqlalchemy.ext.instrumentation.INSTRUMENTATION_MANAGER.
By default, the only finder is find_native_user_instrumentation_hook, which
searches for INSTRUMENTATION_MANAGER. If all finders return None, standard
ClassManager instrumentation is used.
"""
class ExtendedInstrumentationRegistry(InstrumentationFactory):
"""Extends :class:`.InstrumentationFactory` with additional
bookkeeping, to accommodate multiple types of
class managers.
"""
_manager_finders = weakref.WeakKeyDictionary()
_state_finders = weakref.WeakKeyDictionary()
_dict_finders = weakref.WeakKeyDictionary()
_extended = False
def _locate_extended_factory(self, class_):
for finder in instrumentation_finders:
factory = finder(class_)
if factory is not None:
manager = self._extended_class_manager(class_, factory)
return manager, factory
else:
return None, None
def _check_conflicts(self, class_, factory):
existing_factories = self._collect_management_factories_for(
class_
).difference([factory])
if existing_factories:
raise TypeError(
"multiple instrumentation implementations specified "
"in %s inheritance hierarchy: %r"
% (class_.__name__, list(existing_factories))
)
def _extended_class_manager(self, class_, factory):
manager = factory(class_)
if not isinstance(manager, ClassManager):
manager = _ClassInstrumentationAdapter(class_, manager)
if factory != ClassManager and not self._extended:
# somebody invoked a custom ClassManager.
# reinstall global "getter" functions with the more
# expensive ones.
self._extended = True
_install_instrumented_lookups()
self._manager_finders[class_] = manager.manager_getter()
self._state_finders[class_] = manager.state_getter()
self._dict_finders[class_] = manager.dict_getter()
return manager
def _collect_management_factories_for(self, cls):
"""Return a collection of factories in play or specified for a
hierarchy.
Traverses the entire inheritance graph of a cls and returns a
collection of instrumentation factories for those classes. Factories
are extracted from active ClassManagers, if available, otherwise
instrumentation_finders is consulted.
"""
hierarchy = util.class_hierarchy(cls)
factories = set()
for member in hierarchy:
manager = self.opt_manager_of_class(member)
if manager is not None:
factories.add(manager.factory)
else:
for finder in instrumentation_finders:
factory = finder(member)
if factory is not None:
break
else:
factory = None
factories.add(factory)
factories.discard(None)
return factories
def unregister(self, class_):
super().unregister(class_)
if class_ in self._manager_finders:
del self._manager_finders[class_]
del self._state_finders[class_]
del self._dict_finders[class_]
def opt_manager_of_class(self, cls):
try:
finder = self._manager_finders.get(
cls, _default_opt_manager_getter
)
except TypeError:
# due to weakref lookup on invalid object
return None
else:
return finder(cls)
def manager_of_class(self, cls):
try:
finder = self._manager_finders.get(cls, _default_manager_getter)
except TypeError:
# due to weakref lookup on invalid object
raise orm_exc.UnmappedClassError(
cls, f"Can't locate an instrumentation manager for class {cls}"
)
else:
manager = finder(cls)
if manager is None:
raise orm_exc.UnmappedClassError(
cls,
f"Can't locate an instrumentation manager for class {cls}",
)
return manager
def state_of(self, instance):
if instance is None:
raise AttributeError("None has no persistent state.")
return self._state_finders.get(
instance.__class__, _default_state_getter
)(instance)
def dict_of(self, instance):
if instance is None:
raise AttributeError("None has no persistent state.")
return self._dict_finders.get(
instance.__class__, _default_dict_getter
)(instance)
orm_instrumentation._instrumentation_factory = _instrumentation_factory = (
ExtendedInstrumentationRegistry()
)
orm_instrumentation.instrumentation_finders = instrumentation_finders
class InstrumentationManager:
"""User-defined class instrumentation extension.
:class:`.InstrumentationManager` can be subclassed in order
to change
how class instrumentation proceeds. This class exists for
the purposes of integration with other object management
frameworks which would like to entirely modify the
instrumentation methodology of the ORM, and is not intended
for regular usage. For interception of class instrumentation
events, see :class:`.InstrumentationEvents`.
The API for this class should be considered as semi-stable,
and may change slightly with new releases.
"""
# r4361 added a mandatory (cls) constructor to this interface.
# given that, perhaps class_ should be dropped from all of these
# signatures.
def __init__(self, class_):
pass
def manage(self, class_, manager):
setattr(class_, "_default_class_manager", manager)
def unregister(self, class_, manager):
delattr(class_, "_default_class_manager")
def manager_getter(self, class_):
def get(cls):
return cls._default_class_manager
return get
def instrument_attribute(self, class_, key, inst):
pass
def post_configure_attribute(self, class_, key, inst):
pass
def install_descriptor(self, class_, key, inst):
setattr(class_, key, inst)
def uninstall_descriptor(self, class_, key):
delattr(class_, key)
def install_member(self, class_, key, implementation):
setattr(class_, key, implementation)
def uninstall_member(self, class_, key):
delattr(class_, key)
def instrument_collection_class(self, class_, key, collection_class):
return collections.prepare_instrumentation(collection_class)
def get_instance_dict(self, class_, instance):
return instance.__dict__
def initialize_instance_dict(self, class_, instance):
pass
def install_state(self, class_, instance, state):
setattr(instance, "_default_state", state)
def remove_state(self, class_, instance):
delattr(instance, "_default_state")
def state_getter(self, class_):
return lambda instance: getattr(instance, "_default_state")
def dict_getter(self, class_):
return lambda inst: self.get_instance_dict(class_, inst)
class _ClassInstrumentationAdapter(ClassManager):
"""Adapts a user-defined InstrumentationManager to a ClassManager."""
def __init__(self, class_, override):
self._adapted = override
self._get_state = self._adapted.state_getter(class_)
self._get_dict = self._adapted.dict_getter(class_)
ClassManager.__init__(self, class_)
def manage(self):
self._adapted.manage(self.class_, self)
def unregister(self):
self._adapted.unregister(self.class_, self)
def manager_getter(self):
return self._adapted.manager_getter(self.class_)
def instrument_attribute(self, key, inst, propagated=False):
ClassManager.instrument_attribute(self, key, inst, propagated)
if not propagated:
self._adapted.instrument_attribute(self.class_, key, inst)
def post_configure_attribute(self, key):
super().post_configure_attribute(key)
self._adapted.post_configure_attribute(self.class_, key, self[key])
def install_descriptor(self, key, inst):
self._adapted.install_descriptor(self.class_, key, inst)
def uninstall_descriptor(self, key):
self._adapted.uninstall_descriptor(self.class_, key)
def install_member(self, key, implementation):
self._adapted.install_member(self.class_, key, implementation)
def uninstall_member(self, key):
self._adapted.uninstall_member(self.class_, key)
def instrument_collection_class(self, key, collection_class):
return self._adapted.instrument_collection_class(
self.class_, key, collection_class
)
def initialize_collection(self, key, state, factory):
delegate = getattr(self._adapted, "initialize_collection", None)
if delegate:
return delegate(key, state, factory)
else:
return ClassManager.initialize_collection(
self, key, state, factory
)
def new_instance(self, state=None):
instance = self.class_.__new__(self.class_)
self.setup_instance(instance, state)
return instance
def _new_state_if_none(self, instance):
"""Install a default InstanceState if none is present.
A private convenience method used by the __init__ decorator.
"""
if self.has_state(instance):
return False
else:
return self.setup_instance(instance)
def setup_instance(self, instance, state=None):
self._adapted.initialize_instance_dict(self.class_, instance)
if state is None:
state = self._state_constructor(instance, self)
# the given instance is assumed to have no state
self._adapted.install_state(self.class_, instance, state)
return state
def teardown_instance(self, instance):
self._adapted.remove_state(self.class_, instance)
def has_state(self, instance):
try:
self._get_state(instance)
except orm_exc.NO_STATE:
return False
else:
return True
def state_getter(self):
return self._get_state
def dict_getter(self):
return self._get_dict
def _install_instrumented_lookups():
"""Replace global class/object management functions
with ExtendedInstrumentationRegistry implementations, which
allow multiple types of class managers to be present,
at the cost of performance.
This function is called only by ExtendedInstrumentationRegistry
and unit tests specific to this behavior.
The _reinstall_default_lookups() function can be called
after this one to re-establish the default functions.
"""
_install_lookups(
dict(
instance_state=_instrumentation_factory.state_of,
instance_dict=_instrumentation_factory.dict_of,
manager_of_class=_instrumentation_factory.manager_of_class,
opt_manager_of_class=_instrumentation_factory.opt_manager_of_class,
)
)
def _reinstall_default_lookups():
"""Restore simplified lookups."""
_install_lookups(
dict(
instance_state=_default_state_getter,
instance_dict=_default_dict_getter,
manager_of_class=_default_manager_getter,
opt_manager_of_class=_default_opt_manager_getter,
)
)
_instrumentation_factory._extended = False
def _install_lookups(lookups):
global instance_state, instance_dict
global manager_of_class, opt_manager_of_class
instance_state = lookups["instance_state"]
instance_dict = lookups["instance_dict"]
manager_of_class = lookups["manager_of_class"]
opt_manager_of_class = lookups["opt_manager_of_class"]
orm_base.instance_state = attributes.instance_state = (
orm_instrumentation.instance_state
) = instance_state
orm_base.instance_dict = attributes.instance_dict = (
orm_instrumentation.instance_dict
) = instance_dict
orm_base.manager_of_class = attributes.manager_of_class = (
orm_instrumentation.manager_of_class
) = manager_of_class
orm_base.opt_manager_of_class = orm_util.opt_manager_of_class = (
attributes.opt_manager_of_class
) = orm_instrumentation.opt_manager_of_class = opt_manager_of_class

1085
venv/Lib/site-packages/sqlalchemy/ext/mutable.py Normal file
View File

File diff suppressed because it is too large Load Diff