special keyword "ALGORITHM" in the middle, which was intended to be
optional but was not working correctly. The change allows view reflection
to work more completely on MySQL-compatible variants such as StarRocks.
- Pull request courtesy John Bodley.
\ No newline at end of file
+ Pull request courtesy John Bodley.
:members:
.. autoclass:: Composite
- :members:
-.. autodata:: CompositeProperty
+.. autoclass:: CompositeProperty
+ :members:
.. autoclass:: AttributeEventToken
:members:
.. autoclass:: InstrumentedAttribute
- :members: __get__, __set__, __delete__
- :undoc-members:
+ :members:
.. autoclass:: LoaderCallableStatus
:members:
.. autoclass:: MapperProperty
:members:
+.. autoclass:: MappedSQLExpression
+
.. autoclass:: InspectionAttrExtensionType
:members:
:inherited-members:
.. autoclass:: Relationship
- :members:
- :inherited-members:
.. autoclass:: RelationshipDirection
:members:
-.. autodata:: RelationshipProperty
+.. autoclass:: RelationshipProperty
+ :members:
.. autoclass:: Synonym
- :members:
- :inherited-members:
-.. autodata:: SynonymProperty
+.. autoclass:: SynonymProperty
+ :members:
.. autoclass:: QueryContext
:members:
.. autoclass:: QueryableAttribute
:members:
- :inherited-members:
.. autoclass:: UOWTransaction
:members:
target_collection = parent.target_collection
value_attr = parent.value_attr
prop = cast(
- "orm.Relationship[_T]",
+ "orm.RelationshipProperty[_T]",
orm.class_mapper(owning_class).get_property(target_collection),
)
# this was never asserted before but this should be made clear.
- if not isinstance(prop, orm.Relationship):
+ if not isinstance(prop, orm.RelationshipProperty):
raise NotImplementedError(
"association proxy to a non-relationship "
"intermediary is not supported"
for rel in mapper._props.values():
if (
- isinstance(rel, relationships.Relationship)
+ isinstance(rel, relationships.RelationshipProperty)
and rel._init_args.secondary._is_populated()
):
RELATIONSHIP,
{
"sqlalchemy.orm.relationships.Relationship",
+ "sqlalchemy.orm.relationships.RelationshipProperty",
"sqlalchemy.orm.Relationship",
+ "sqlalchemy.orm.RelationshipProperty",
+ },
+ ),
+ "RelationshipProperty": (
+ RELATIONSHIP,
+ {
+ "sqlalchemy.orm.relationships.Relationship",
+ "sqlalchemy.orm.relationships.RelationshipProperty",
+ "sqlalchemy.orm.Relationship",
+ "sqlalchemy.orm.RelationshipProperty",
},
),
"registry": (
"ColumnProperty": (
COLUMN_PROPERTY,
{
+ "sqlalchemy.orm.properties.MappedSQLExpression",
+ "sqlalchemy.orm.MappedSQLExpression",
+ "sqlalchemy.orm.properties.ColumnProperty",
+ "sqlalchemy.orm.ColumnProperty",
+ },
+ ),
+ "MappedSQLExpression": (
+ COLUMN_PROPERTY,
+ {
+ "sqlalchemy.orm.properties.MappedSQLExpression",
+ "sqlalchemy.orm.MappedSQLExpression",
"sqlalchemy.orm.properties.ColumnProperty",
"sqlalchemy.orm.ColumnProperty",
},
{
"sqlalchemy.orm.descriptor_props.Synonym",
"sqlalchemy.orm.Synonym",
+ "sqlalchemy.orm.descriptor_props.SynonymProperty",
+ "sqlalchemy.orm.SynonymProperty",
+ },
+ ),
+ "SynonymProperty": (
+ SYNONYM_PROPERTY,
+ {
+ "sqlalchemy.orm.descriptor_props.Synonym",
+ "sqlalchemy.orm.Synonym",
+ "sqlalchemy.orm.descriptor_props.SynonymProperty",
+ "sqlalchemy.orm.SynonymProperty",
},
),
"Composite": (
{
"sqlalchemy.orm.descriptor_props.Composite",
"sqlalchemy.orm.Composite",
+ "sqlalchemy.orm.descriptor_props.CompositeProperty",
+ "sqlalchemy.orm.CompositeProperty",
+ },
+ ),
+ "CompositeProperty": (
+ COMPOSITE_PROPERTY,
+ {
+ "sqlalchemy.orm.descriptor_props.Composite",
+ "sqlalchemy.orm.Composite",
+ "sqlalchemy.orm.descriptor_props.CompositeProperty",
+ "sqlalchemy.orm.CompositeProperty",
},
),
"MapperProperty": (
from ._orm_constructors import clear_mappers as clear_mappers
from ._orm_constructors import column_property as column_property
from ._orm_constructors import composite as composite
-from ._orm_constructors import CompositeProperty as CompositeProperty
from ._orm_constructors import contains_alias as contains_alias
from ._orm_constructors import create_session as create_session
from ._orm_constructors import deferred as deferred
from ._orm_constructors import outerjoin as outerjoin
from ._orm_constructors import query_expression as query_expression
from ._orm_constructors import relationship as relationship
-from ._orm_constructors import RelationshipProperty as RelationshipProperty
from ._orm_constructors import synonym as synonym
-from ._orm_constructors import SynonymProperty as SynonymProperty
from ._orm_constructors import with_loader_criteria as with_loader_criteria
from ._orm_constructors import with_polymorphic as with_polymorphic
from .attributes import AttributeEventToken as AttributeEventToken
from .decl_api import registry as registry
from .decl_api import synonym_for as synonym_for
from .descriptor_props import Composite as Composite
+from .descriptor_props import CompositeProperty as CompositeProperty
from .descriptor_props import Synonym as Synonym
+from .descriptor_props import SynonymProperty as SynonymProperty
from .dynamic import AppenderQuery as AppenderQuery
from .events import AttributeEvents as AttributeEvents
from .events import InstanceEvents as InstanceEvents
from .mapper import validates as validates
from .properties import ColumnProperty as ColumnProperty
from .properties import MappedColumn as MappedColumn
+from .properties import MappedSQLExpression as MappedSQLExpression
from .query import AliasOption as AliasOption
from .query import Query as Query
from .relationships import foreign as foreign
from .relationships import Relationship as Relationship
+from .relationships import RelationshipProperty as RelationshipProperty
from .relationships import remote as remote
from .scoping import scoped_session as scoped_session
from .session import close_all_sessions as close_all_sessions
from .interfaces import _AttributeOptions
from .properties import ColumnProperty
from .properties import MappedColumn
+from .properties import MappedSQLExpression
from .query import AliasOption
from .relationships import _RelationshipArgumentType
from .relationships import Relationship
+from .relationships import RelationshipProperty
from .session import Session
from .util import _ORMJoin
from .util import AliasedClass
_T = typing.TypeVar("_T")
-CompositeProperty = Composite
-"""Alias for :class:`_orm.Composite`."""
-
-RelationshipProperty = Relationship
-"""Alias for :class:`_orm.Relationship`."""
-
-SynonymProperty = Synonym
-"""Alias for :class:`_orm.Synonym`."""
-
-
@util.deprecated(
"1.4",
"The :class:`.AliasOption` object is not necessary "
expire_on_flush: bool = True,
info: Optional[_InfoType] = None,
doc: Optional[str] = None,
-) -> ColumnProperty[_T]:
+) -> MappedSQLExpression[_T]:
r"""Provide a column-level property for use with a mapping.
Column-based properties can normally be applied to the mapper's
expressions
"""
- return ColumnProperty(
+ return MappedSQLExpression(
column,
*additional_columns,
attribute_options=_AttributeOptions(
foreign_keys: Optional[_ORMColCollectionArgument] = None,
remote_side: Optional[_ORMColCollectionArgument] = None,
join_depth: Optional[int] = None,
- comparator_factory: Optional[Type[Relationship.Comparator[Any]]] = None,
+ comparator_factory: Optional[
+ Type[RelationshipProperty.Comparator[Any]]
+ ] = None,
single_parent: bool = False,
innerjoin: bool = False,
distinct_target_key: Optional[bool] = None,
"""
+
return Relationship(
argument,
secondary=secondary,
def dynamic_loader(
argument: Optional[_RelationshipArgumentType[Any]] = None, **kw: Any
-) -> Relationship[Any]:
+) -> RelationshipProperty[Any]:
"""Construct a dynamically-loading mapper property.
This is essentially the same as
from .interfaces import MapperProperty
from .interfaces import UserDefinedOption
from .mapper import Mapper
- from .relationships import Relationship
+ from .relationships import RelationshipProperty
from .state import InstanceState
from .util import AliasedClass
from .util import AliasedInsp
def prop_is_relationship(
prop: MapperProperty[Any],
- ) -> TypeGuard[Relationship[Any]]:
+ ) -> TypeGuard[RelationshipProperty[Any]]:
...
def is_collection_impl(
from . import exc as orm_exc
from . import interfaces
from ._typing import insp_is_aliased_class
+from .base import _DeclarativeMapped
from .base import ATTR_EMPTY
from .base import ATTR_WAS_SET
from .base import CALLABLES_OK
from .collections import CollectionAdapter
from .dynamic import DynamicAttributeImpl
from .interfaces import MapperProperty
- from .relationships import Relationship
+ from .relationships import RelationshipProperty
from .state import InstanceState
from .util import AliasedInsp
from ..event.base import _Dispatch
@inspection._self_inspects
class QueryableAttribute(
roles.ExpressionElementRole[_T],
- interfaces._MappedAttribute[_T],
+ _DeclarativeMapped[_T],
interfaces.InspectionAttr,
interfaces.PropComparator[_T],
roles.JoinTargetRole,
self, *clauses: _ColumnExpressionArgument[bool]
) -> interfaces.PropComparator[bool]:
if TYPE_CHECKING:
- assert isinstance(self.comparator, Relationship.Comparator)
+ assert isinstance(self.comparator, RelationshipProperty.Comparator)
exprs = tuple(
coercions.expect(roles.WhereHavingRole, clause)
__slots__ = ()
inherit_cache = True
+ """:meta private:"""
- # if not TYPE_CHECKING:
+ # hack to make __doc__ writeable on instances of
+ # InstrumentedAttribute, while still keeping classlevel
+ # __doc__ correct
- @property # type: ignore
+ @util.rw_hybridproperty # type: ignore
def __doc__(self) -> Optional[str]: # type: ignore
return self._doc
- @__doc__.setter
- def __doc__(self, value: Optional[str]) -> None:
+ @__doc__.setter # type: ignore
+ def __doc__(self, value: Optional[str]) -> None: # type: ignore
self._doc = value
+ @__doc__.classlevel # type: ignore
+ def __doc__(cls) -> Optional[str]: # type: ignore
+ return super().__doc__
+
def __set__(self, instance: object, value: Any) -> None:
self.impl.set(
instance_state(instance), instance_dict(instance), value, None
class_, key, comparator=comparator, parententity=parententity
)
- descriptor.__doc__ = doc
+ descriptor.__doc__ = doc # type: ignore
manager.instrument_attribute(key, descriptor)
return descriptor
class RelationshipDirection(Enum):
"""enumeration which indicates the 'direction' of a
- :class:`_orm.Relationship`.
+ :class:`_orm.RelationshipProperty`.
:class:`.RelationshipDirection` is accessible from the
:attr:`_orm.Relationship.direction` attribute of
- :class:`_orm.Relationship`.
+ :class:`_orm.RelationshipProperty`.
"""
...
-class _MappedAttribute(Mapped[_T], TypingOnly):
+class _MappedAttribute(Generic[_T], TypingOnly):
"""Mixin for attributes which should be replaced by mapper-assigned
attributes.
"""
__slots__ = ()
+
+
+class _DeclarativeMapped(Mapped[_T], _MappedAttribute[_T]):
+ """Mixin for :class:`.MapperProperty` subclasses that allows them to
+ be compatible with ORM-annotated declarative mappings.
+
+ """
+
+ __slots__ = ()
from . import attributes
from . import interfaces
-from .descriptor_props import Synonym
+from .descriptor_props import SynonymProperty
from .properties import ColumnProperty
from .util import class_mapper
from .. import exc
from ..util.typing import CallableReference
if TYPE_CHECKING:
- from .relationships import Relationship
+ from .relationships import RelationshipProperty
from ..sql.schema import MetaData
from ..sql.schema import Table
if desc.extension_type is interfaces.NotExtension.NOT_EXTENSION:
assert isinstance(desc, attributes.QueryableAttribute)
prop = desc.property
- if isinstance(prop, Synonym):
+ if isinstance(prop, SynonymProperty):
key = prop.name
elif not isinstance(prop, ColumnProperty):
raise exc.InvalidRequestError(
)
cls: Type[Any]
- prop: Relationship[Any]
+ prop: RelationshipProperty[Any]
fallback: Mapping[str, Any]
arg: str
favor_tables: bool
def __init__(
self,
cls: Type[Any],
- prop: Relationship[Any],
+ prop: RelationshipProperty[Any],
fallback: Mapping[str, Any],
arg: str,
favor_tables: bool = False,
def _resolver(
- cls: Type[Any], prop: Relationship[Any]
+ cls: Type[Any], prop: RelationshipProperty[Any]
) -> Tuple[
Callable[[str], Callable[[], Union[Type[Any], Table, _ModNS]]],
Callable[[str, bool], _class_resolver],
from .mapper import Mapper
from .properties import ColumnProperty
from .properties import MappedColumn
-from .relationships import Relationship
+from .relationships import RelationshipProperty
from .state import InstanceState
from .. import exc
from .. import inspection
@compat_typing.dataclass_transform(
field_descriptors=(
MappedColumn[Any],
- Relationship[Any],
+ RelationshipProperty[Any],
Composite[Any],
ColumnProperty[Any],
Synonym[Any],
@compat_typing.dataclass_transform(
field_descriptors=(
MappedColumn[Any],
- Relationship[Any],
+ RelationshipProperty[Any],
Composite[Any],
ColumnProperty[Any],
Synonym[Any],
from .attributes import QueryableAttribute
from .base import _is_mapped_class
from .base import InspectionAttr
-from .descriptor_props import Composite
-from .descriptor_props import Synonym
+from .descriptor_props import CompositeProperty
+from .descriptor_props import SynonymProperty
from .interfaces import _AttributeOptions
from .interfaces import _IntrospectsAnnotations
from .interfaces import _MappedAttribute
):
# detect a QueryableAttribute that's already mapped being
# assigned elsewhere in userland, turn into a synonym()
- value = Synonym(value.key)
+ value = SynonymProperty(value.key)
setattr(cls, k, value)
if (
del our_stuff[key]
for col in c.columns_to_assign:
- if not isinstance(c, Composite):
+ if not isinstance(c, CompositeProperty):
name_to_prop_key[col.name].add(key)
declared_columns.add(col)
elif isinstance(value, QueryableAttribute) and value.key != key:
# detect a QueryableAttribute that's already mapped being
# assigned elsewhere in userland, turn into a synonym()
- value = Synonym(value.key)
+ value = SynonymProperty(value.key)
mapped_cls.__mapper__.add_property(key, value)
else:
type.__setattr__(cls, key, value)
from . import attributes
from . import util as orm_util
+from .base import _DeclarativeMapped
from .base import LoaderCallableStatus
from .base import Mapped
from .base import PassiveFlag
] = weakref.WeakKeyDictionary()
-class Composite(
+class CompositeProperty(
_MapsColumns[_CC], _IntrospectsAnnotations, DescriptorProperty[_CC]
):
"""Defines a "composite" mapped attribute, representing a collection
of columns as one attribute.
- :class:`.Composite` is constructed using the :func:`.composite`
+ :class:`.CompositeProperty` is constructed using the :func:`.composite`
function.
- .. versionchanged:: 2.0 Renamed :class:`_orm.CompositeProperty`
- to :class:`_orm.Composite`. The old name
- :class:`_orm.CompositeProperty` remains as an alias.
-
.. seealso::
:ref:`mapper_composite`
group=False, *self._comparable_elements
)
- def __clause_element__(self) -> Composite.CompositeBundle[_PT]:
+ def __clause_element__(self) -> CompositeProperty.CompositeBundle[_PT]:
return self.expression
@util.memoized_property
- def expression(self) -> Composite.CompositeBundle[_PT]:
+ def expression(self) -> CompositeProperty.CompositeBundle[_PT]:
clauses = self.clauses._annotate(
{
"parententity": self._parententity,
"proxy_key": self.prop.key,
}
)
- return Composite.CompositeBundle(self.prop, clauses)
+ return CompositeProperty.CompositeBundle(self.prop, clauses)
def _bulk_update_tuples(
self, value: Any
return str(self.parent.class_.__name__) + "." + self.key
+class Composite(CompositeProperty[_T], _DeclarativeMapped[_T]):
+ """Declarative-compatible front-end for the :class:`.CompositeProperty`
+ class.
+
+ Public constructor is the :func:`_orm.composite` function.
+
+ .. versionchanged:: 2.0 Added :class:`_orm.Composite` as a Declarative
+ compatible subclass of :class:`_orm.CompositeProperty`.
+
+ .. seealso::
+
+ :ref:`mapper_composite`
+
+ """
+
+ inherit_cache = True
+ """:meta private:"""
+
+
class ConcreteInheritedProperty(DescriptorProperty[_T]):
"""A 'do nothing' :class:`.MapperProperty` that disables
an attribute on a concrete subclass that is only present
self.descriptor = NoninheritedConcreteProp()
-class Synonym(DescriptorProperty[_T]):
+class SynonymProperty(DescriptorProperty[_T]):
"""Denote an attribute name as a synonym to a mapped property,
in that the attribute will mirror the value and expression behavior
of another attribute.
:class:`.Synonym` is constructed using the :func:`_orm.synonym`
function.
- .. versionchanged:: 2.0 Renamed :class:`_orm.SynonymProperty`
- to :class:`_orm.Synonym`. The old name
- :class:`_orm.SynonymProperty` remains as an alias.
-
.. seealso::
:ref:`synonyms` - Overview of synonyms
p._mapped_by_synonym = self.key
self.parent = parent
+
+
+class Synonym(SynonymProperty[_T], _DeclarativeMapped[_T]):
+ """Declarative front-end for the :class:`.SynonymProperty` class.
+
+ Public constructor is the :func:`_orm.synonym` function.
+
+ .. versionchanged:: 2.0 Added :class:`_orm.Synonym` as a Declarative
+ compatible subclass for :class:`_orm.SynonymProperty`
+
+ .. seealso::
+
+ :ref:`synonyms` - Overview of synonyms
+
+ """
+
+ inherit_cache = True
+ """:meta private:"""
@log.class_logger
-@relationships.Relationship.strategy_for(lazy="dynamic")
+@relationships.RelationshipProperty.strategy_for(lazy="dynamic")
class DynaLoader(strategies.AbstractRelationshipLoader, log.Identified):
def init_class_attribute(self, mapper):
self.is_class_level = True
# by typing tools
@inspection._self_inspects
class MapperProperty(
- HasCacheKey, _MappedAttribute[_T], InspectionAttrInfo, util.MemoizedSlots
+ HasCacheKey,
+ _MappedAttribute[_T],
+ InspectionAttrInfo,
+ util.MemoizedSlots,
):
"""Represent a particular class attribute mapped by :class:`_orm.Mapper`.
from ._typing import _RegistryType
from .decl_api import registry
from .dependency import DependencyProcessor
- from .descriptor_props import Composite
- from .descriptor_props import Synonym
+ from .descriptor_props import CompositeProperty
+ from .descriptor_props import SynonymProperty
from .events import MapperEvents
from .instrumentation import ClassManager
from .path_registry import CachingEntityRegistry
from .properties import ColumnProperty
- from .relationships import Relationship
+ from .relationships import RelationshipProperty
from .state import InstanceState
from ..engine import Row
from ..engine import RowMapping
@HasMemoized.memoized_attribute
@util.preload_module("sqlalchemy.orm.descriptor_props")
- def synonyms(self) -> util.ReadOnlyProperties[Synonym[Any]]:
+ def synonyms(self) -> util.ReadOnlyProperties[SynonymProperty[Any]]:
"""Return a namespace of all :class:`.Synonym`
properties maintained by this :class:`_orm.Mapper`.
"""
descriptor_props = util.preloaded.orm_descriptor_props
- return self._filter_properties(descriptor_props.Synonym)
+ return self._filter_properties(descriptor_props.SynonymProperty)
@property
def entity_namespace(self):
@HasMemoized.memoized_attribute
@util.preload_module("sqlalchemy.orm.relationships")
- def relationships(self) -> util.ReadOnlyProperties[Relationship[Any]]:
+ def relationships(
+ self,
+ ) -> util.ReadOnlyProperties[RelationshipProperty[Any]]:
"""A namespace of all :class:`.Relationship` properties
maintained by this :class:`_orm.Mapper`.
"""
return self._filter_properties(
- util.preloaded.orm_relationships.Relationship
+ util.preloaded.orm_relationships.RelationshipProperty
)
@HasMemoized.memoized_attribute
@util.preload_module("sqlalchemy.orm.descriptor_props")
- def composites(self) -> util.ReadOnlyProperties[Composite[Any]]:
+ def composites(self) -> util.ReadOnlyProperties[CompositeProperty[Any]]:
"""Return a namespace of all :class:`.Composite`
properties maintained by this :class:`_orm.Mapper`.
"""
return self._filter_properties(
- util.preloaded.orm_descriptor_props.Composite
+ util.preloaded.orm_descriptor_props.CompositeProperty
)
def _filter_properties(
from ._typing import _InternalEntityType
from .interfaces import MapperProperty
from .mapper import Mapper
- from .relationships import Relationship
+ from .relationships import RelationshipProperty
from .util import AliasedInsp
from ..sql.cache_key import _CacheKeyTraversalType
from ..sql.elements import BindParameter
self.has_entity = prop._links_to_entity
if prop._is_relationship:
if TYPE_CHECKING:
- assert isinstance(prop, Relationship)
+ assert isinstance(prop, RelationshipProperty)
self.entity = prop.entity
self.mapper = prop.mapper
else:
from . import attributes
from . import strategy_options
-from .descriptor_props import Composite
+from .base import _DeclarativeMapped
+from .descriptor_props import CompositeProperty
from .descriptor_props import ConcreteInheritedProperty
-from .descriptor_props import Synonym
+from .descriptor_props import SynonymProperty
from .interfaces import _AttributeOptions
from .interfaces import _DEFAULT_ATTRIBUTE_OPTIONS
from .interfaces import _IntrospectsAnnotations
from .interfaces import MapperProperty
from .interfaces import PropComparator
from .interfaces import StrategizedProperty
-from .relationships import Relationship
+from .relationships import RelationshipProperty
from .. import exc as sa_exc
from .. import ForeignKey
from .. import log
__all__ = [
"ColumnProperty",
- "Composite",
+ "CompositeProperty",
"ConcreteInheritedProperty",
- "Relationship",
- "Synonym",
+ "RelationshipProperty",
+ "SynonymProperty",
]
_IntrospectsAnnotations,
log.Identified,
):
- """Describes an object attribute that corresponds to a table column.
+ """Describes an object attribute that corresponds to a table column
+ or other column expression.
Public constructor is the :func:`_orm.column_property` function.
strategy_wildcard_key = strategy_options._COLUMN_TOKEN
inherit_cache = True
+ """:meta private:"""
+
_links_to_entity = False
columns: List[NamedColumn[Any]]
return str(self.parent.class_.__name__) + "." + self.key
+class MappedSQLExpression(ColumnProperty[_T], _DeclarativeMapped[_T]):
+ """Declarative front-end for the :class:`.ColumnProperty` class.
+
+ Public constructor is the :func:`_orm.column_property` function.
+
+ .. versionchanged:: 2.0 Added :class:`_orm.MappedSQLExpression` as
+ a Declarative compatible subclass for :class:`_orm.ColumnProperty`.
+
+ .. seealso::
+
+ :class:`.MappedColumn`
+
+ """
+
+ inherit_cache = True
+ """:meta private:"""
+
+
class MappedColumn(
SQLCoreOperations[_T],
_IntrospectsAnnotations,
_MapsColumns[_T],
+ _DeclarativeMapped[_T],
):
"""Maps a single :class:`_schema.Column` on a class.
for prop in mapper.iterate_properties:
if (
- isinstance(prop, relationships.Relationship)
+ isinstance(prop, relationships.RelationshipProperty)
and prop.mapper is entity_zero.mapper # type: ignore
):
property = prop # type: ignore # noqa: A001
from . import strategy_options
from ._typing import insp_is_aliased_class
from ._typing import is_has_collection_adapter
+from .base import _DeclarativeMapped
from .base import _is_mapped_class
from .base import class_mapper
from .base import LoaderCallableStatus
@log.class_logger
-class Relationship(
+class RelationshipProperty(
_IntrospectsAnnotations, StrategizedProperty[_T], log.Identified
):
"""Describes an object property that holds a single item or list
:ref:`relationship_config_toplevel`
- .. versionchanged:: 2.0 Renamed :class:`_orm.RelationshipProperty`
- to :class:`_orm.Relationship`. The old name
- :class:`_orm.RelationshipProperty` remains as an alias.
-
"""
strategy_wildcard_key = strategy_options._RELATIONSHIP_TOKEN
inherit_cache = True
+ """:meta private:"""
_links_to_entity = True
_is_relationship = True
remote_side: Optional[_ORMColCollectionArgument] = None,
join_depth: Optional[int] = None,
comparator_factory: Optional[
- Type[Relationship.Comparator[Any]]
+ Type[RelationshipProperty.Comparator[Any]]
] = None,
single_parent: bool = False,
innerjoin: bool = False,
_local_remote_pairs: Optional[_ColumnPairs] = None,
_legacy_inactive_history_style: bool = False,
):
- super(Relationship, self).__init__(attribute_options=attribute_options)
+ super().__init__(attribute_options=attribute_options)
self.uselist = uselist
self.argument = argument
self.omit_join = omit_join
self.local_remote_pairs = _local_remote_pairs
self.load_on_pending = load_on_pending
- self.comparator_factory = comparator_factory or Relationship.Comparator
+ self.comparator_factory = (
+ comparator_factory or RelationshipProperty.Comparator
+ )
util.set_creation_order(self)
if info is not None:
self.strategy_key = (("lazy", self.lazy),)
- self._reverse_property: Set[Relationship[Any]] = set()
+ self._reverse_property: Set[RelationshipProperty[Any]] = set()
if overlaps:
self._overlaps = set(re.split(r"\s*,\s*", overlaps)) # type: ignore # noqa: E501
class Comparator(util.MemoizedSlots, PropComparator[_PT]):
"""Produce boolean, comparison, and other operators for
- :class:`.Relationship` attributes.
+ :class:`.RelationshipProperty` attributes.
See the documentation for :class:`.PropComparator` for a brief
overview of ORM level operator definition.
"_extra_criteria",
)
- prop: RODescriptorReference[Relationship[_PT]]
+ prop: RODescriptorReference[RelationshipProperty[_PT]]
_of_type: Optional[_EntityType[_PT]]
def __init__(
self,
- prop: Relationship[_PT],
+ prop: RelationshipProperty[_PT],
parentmapper: _InternalEntityType[Any],
adapt_to_entity: Optional[AliasedInsp[Any]] = None,
of_type: Optional[_EntityType[_PT]] = None,
extra_criteria: Tuple[ColumnElement[bool], ...] = (),
):
- """Construction of :class:`.Relationship.Comparator`
+ """Construction of :class:`.RelationshipProperty.Comparator`
is internal to the ORM's attribute mechanics.
"""
def adapt_to_entity(
self, adapt_to_entity: AliasedInsp[Any]
- ) -> Relationship.Comparator[Any]:
+ ) -> RelationshipProperty.Comparator[Any]:
return self.__class__(
self.prop,
self._parententity,
entity: _InternalEntityType[_PT]
"""The target entity referred to by this
- :class:`.Relationship.Comparator`.
+ :class:`.RelationshipProperty.Comparator`.
This is either a :class:`_orm.Mapper` or :class:`.AliasedInsp`
object.
mapper: Mapper[_PT]
"""The target :class:`_orm.Mapper` referred to by this
- :class:`.Relationship.Comparator`.
+ :class:`.RelationshipProperty.Comparator`.
This is the "target" or "remote" side of the
:func:`_orm.relationship`.
"""
- return Relationship.Comparator(
+ return RelationshipProperty.Comparator(
self.prop,
self._parententity,
adapt_to_entity=self._adapt_to_entity,
for clause in util.coerce_generator_arg(criteria)
)
- return Relationship.Comparator(
+ return RelationshipProperty.Comparator(
self.prop,
self._parententity,
adapt_to_entity=self._adapt_to_entity,
else:
return _orm_annotate(self.__negated_contains_or_equals(other))
- def _memoized_attr_property(self) -> Relationship[_PT]:
+ def _memoized_attr_property(self) -> RelationshipProperty[_PT]:
self.prop.parent._check_configure()
return self.prop
@staticmethod
def _check_sync_backref(
- rel_a: Relationship[Any], rel_b: Relationship[Any]
+ rel_a: RelationshipProperty[Any], rel_b: RelationshipProperty[Any]
) -> None:
if rel_a.viewonly and rel_b.sync_backref:
raise sa_exc.InvalidRequestError(
def _add_reverse_property(self, key: str) -> None:
other = self.mapper.get_property(key, _configure_mappers=False)
- if not isinstance(other, Relationship):
+ if not isinstance(other, RelationshipProperty):
raise sa_exc.InvalidRequestError(
"back_populates on relationship '%s' refers to attribute '%s' "
"that is not a relationship. The back_populates parameter "
@util.memoized_property
def mapper(self) -> Mapper[_T]:
"""Return the targeted :class:`_orm.Mapper` for this
- :class:`.Relationship`.
+ :class:`.RelationshipProperty`.
"""
return self.entity.mapper
self._post_init()
self._generate_backref()
self._join_condition._warn_for_conflicting_sync_targets()
- super(Relationship, self).do_init()
+ super().do_init()
self._lazy_strategy = cast(
"LazyLoader", self._get_strategy((("lazy", "select"),))
)
@property
def cascade(self) -> CascadeOptions:
"""Return the current cascade setting for this
- :class:`.Relationship`.
+ :class:`.RelationshipProperty`.
"""
return self._cascade
def _columns_are_mapped(self, *cols: ColumnElement[Any]) -> bool:
"""Return True if all columns in the given collection are
- mapped by the tables referenced by this :class:`.Relationship`.
+ mapped by the tables referenced by this :class:`.RelationshipProperty`.
"""
kwargs.setdefault("passive_updates", self.passive_updates)
kwargs.setdefault("sync_backref", self.sync_backref)
self.back_populates = backref_key
- relationship = Relationship(
+ relationship = RelationshipProperty(
parent,
self.secondary,
primaryjoin=pj,
primaryjoin: ColumnElement[bool]
secondaryjoin: Optional[ColumnElement[bool]]
secondary: Optional[FromClause]
- prop: Relationship[Any]
+ prop: RelationshipProperty[Any]
synchronize_pairs: _ColumnPairs
secondary_synchronize_pairs: _ColumnPairs
child_persist_selectable: FromClause,
parent_local_selectable: FromClause,
child_local_selectable: FromClause,
+ *,
primaryjoin: Optional[ColumnElement[bool]] = None,
secondary: Optional[FromClause] = None,
secondaryjoin: Optional[ColumnElement[bool]] = None,
local_remote_pairs: Optional[_ColumnPairs] = None,
remote_side: Any = None,
self_referential: Any = False,
- prop: Optional[Relationship[Any]] = None,
+ prop: RelationshipProperty[Any],
support_sync: bool = True,
can_be_synced_fn: Callable[..., bool] = lambda *c: True,
):
+
self.parent_persist_selectable = parent_persist_selectable
self.parent_local_selectable = parent_local_selectable
self.child_persist_selectable = child_persist_selectable
self._log_joins()
def _log_joins(self) -> None:
- if self.prop is None:
- return
log = self.prop.logger
log.info("%s setup primary join %s", self.prop, self.primaryjoin)
log.info("%s setup secondary join %s", self.prop, self.secondaryjoin)
)
def _annotate_parentmapper(self) -> None:
- if self.prop is None:
- return
-
def parentmappers_(element: _CE, **kw: Any) -> Optional[_CE]:
if "remote" in element._annotations:
return element._annotate({"parentmapper": self.prop.mapper})
_track_overlapping_sync_targets: weakref.WeakKeyDictionary[
ColumnElement[Any],
- weakref.WeakKeyDictionary[Relationship[Any], ColumnElement[Any]],
+ weakref.WeakKeyDictionary[
+ RelationshipProperty[Any], ColumnElement[Any]
+ ],
] = weakref.WeakKeyDictionary()
def _warn_for_conflicting_sync_targets(self) -> None:
def __call__(self, c: ClauseElement) -> bool:
return self.name in c._annotations
+
+
+class Relationship(RelationshipProperty[_T], _DeclarativeMapped[_T]):
+ """Declarative front-end for the :class:`.RelationshipProperty` class.
+
+ Public constructor is the :func:`_orm.relationship` function.
+
+ .. seealso::
+
+ :ref:`relationship_config_toplevel`
+
+ .. versionchanged:: 2.0 Added :class:`_orm.Relationship` as a Declarative
+ compatible subclass for :class:`_orm.RelationshipProperty`.
+
+ """
+
+ inherit_cache = True
+ """:meta private:"""
from ..sql.selectable import Select
if TYPE_CHECKING:
- from .relationships import Relationship
+ from .relationships import RelationshipProperty
from ..sql.elements import ColumnElement
@log.class_logger
-@relationships.Relationship.strategy_for(do_nothing=True)
+@relationships.RelationshipProperty.strategy_for(do_nothing=True)
class DoNothingLoader(LoaderStrategy):
"""Relationship loader that makes no change to the object's state.
@log.class_logger
-@relationships.Relationship.strategy_for(lazy="noload")
-@relationships.Relationship.strategy_for(lazy=None)
+@relationships.RelationshipProperty.strategy_for(lazy="noload")
+@relationships.RelationshipProperty.strategy_for(lazy=None)
class NoLoader(AbstractRelationshipLoader):
"""Provide loading behavior for a :class:`.Relationship`
with "lazy=None".
@log.class_logger
-@relationships.Relationship.strategy_for(lazy=True)
-@relationships.Relationship.strategy_for(lazy="select")
-@relationships.Relationship.strategy_for(lazy="raise")
-@relationships.Relationship.strategy_for(lazy="raise_on_sql")
-@relationships.Relationship.strategy_for(lazy="baked_select")
+@relationships.RelationshipProperty.strategy_for(lazy=True)
+@relationships.RelationshipProperty.strategy_for(lazy="select")
+@relationships.RelationshipProperty.strategy_for(lazy="raise")
+@relationships.RelationshipProperty.strategy_for(lazy="raise_on_sql")
+@relationships.RelationshipProperty.strategy_for(lazy="baked_select")
class LazyLoader(
AbstractRelationshipLoader, util.MemoizedSlots, log.Identified
):
_rev_lazywhere: ColumnElement[bool]
_rev_bind_to_col: Dict[str, ColumnElement[Any]]
- parent_property: Relationship[Any]
+ parent_property: RelationshipProperty[Any]
def __init__(
- self, parent: Relationship[Any], strategy_key: Tuple[Any, ...]
+ self, parent: RelationshipProperty[Any], strategy_key: Tuple[Any, ...]
):
super(LazyLoader, self).__init__(parent, strategy_key)
self._raise_always = self.strategy_opts["lazy"] == "raise"
)
-@relationships.Relationship.strategy_for(lazy="immediate")
+@relationships.RelationshipProperty.strategy_for(lazy="immediate")
class ImmediateLoader(PostLoader):
__slots__ = ()
@log.class_logger
-@relationships.Relationship.strategy_for(lazy="subquery")
+@relationships.RelationshipProperty.strategy_for(lazy="subquery")
class SubqueryLoader(PostLoader):
__slots__ = ("join_depth",)
@log.class_logger
-@relationships.Relationship.strategy_for(lazy="joined")
-@relationships.Relationship.strategy_for(lazy=False)
+@relationships.RelationshipProperty.strategy_for(lazy="joined")
+@relationships.RelationshipProperty.strategy_for(lazy=False)
class JoinedLoader(AbstractRelationshipLoader):
"""Provide loading behavior for a :class:`.Relationship`
using joined eager loading.
@log.class_logger
-@relationships.Relationship.strategy_for(lazy="selectin")
+@relationships.RelationshipProperty.strategy_for(lazy="selectin")
class SelectInLoader(PostLoader, util.MemoizedSlots):
__slots__ = (
"join_depth",
from .context import ORMCompileState
from .mapper import Mapper
from .query import Query
- from .relationships import Relationship
+ from .relationships import RelationshipProperty
from ..engine import Row
from ..engine import RowMapping
from ..sql._typing import _CE
if isinstance(onclause, attributes.QueryableAttribute):
if TYPE_CHECKING:
- assert isinstance(onclause.comparator, Relationship.Comparator)
+ assert isinstance(
+ onclause.comparator, RelationshipProperty.Comparator
+ )
on_selectable = onclause.comparator._source_selectable()
prop = onclause.property
_extra_criteria += onclause._extra_criteria
.. versionadded:: 1.2
"""
- prop_t: Relationship[Any]
+ prop_t: RelationshipProperty[Any]
if isinstance(prop, str):
raise sa_exc.ArgumentError(
from .langhelpers import quoted_token_parser as quoted_token_parser
from .langhelpers import ro_memoized_property as ro_memoized_property
from .langhelpers import ro_non_memoized_property as ro_non_memoized_property
+from .langhelpers import rw_hybridproperty as rw_hybridproperty
from .langhelpers import safe_reraise as safe_reraise
from .langhelpers import set_creation_order as set_creation_order
from .langhelpers import string_or_unprintable as string_or_unprintable
def clsname_as_plain_name(cls: Type[Any]) -> str:
return " ".join(
- n.lower() for n in re.findall(r"([A-Z][a-z]+)", cls.__name__)
+ n.lower() for n in re.findall(r"([A-Z][a-z]+|SQL)", cls.__name__)
)
return self
+class rw_hybridproperty(Generic[_T]):
+ def __init__(self, func: Callable[..., _T]):
+ self.func = func
+ self.clslevel = func
+ self.setfn: Optional[Callable[..., Any]] = None
+
+ def __get__(self, instance: Any, owner: Any) -> _T:
+ if instance is None:
+ clsval = self.clslevel(owner)
+ return clsval
+ else:
+ return self.func(instance)
+
+ def __set__(self, instance: Any, value: Any) -> None:
+ assert self.setfn is not None
+ self.setfn(instance, value)
+
+ def setter(self, func: Callable[..., Any]) -> rw_hybridproperty[_T]:
+ self.setfn = func
+ return self
+
+ def classlevel(self, func: Callable[..., Any]) -> rw_hybridproperty[_T]:
+ self.clslevel = func
+ return self
+
+
class hybridmethod(Generic[_T]):
"""Decorate a function as cls- or instance- level."""
from sqlalchemy.orm import deferred
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import registry
-from sqlalchemy.orm import Relationship
from sqlalchemy.orm import relationship
+from sqlalchemy.orm import RelationshipProperty
from sqlalchemy.orm.decl_api import declared_attr
from sqlalchemy.orm.interfaces import MapperProperty
from sqlalchemy.sql.schema import ForeignKey
return relationship("A", back_populates="bs")
@declared_attr
- def a3(cls) -> Relationship["A"]:
+ def a3(cls) -> RelationshipProperty["A"]:
return relationship("A", back_populates="bs")
@declared_attr
- def c1(cls) -> Relationship[C]:
+ def c1(cls) -> RelationshipProperty[C]:
return relationship(C, back_populates="bs")
@declared_attr
from sqlalchemy import String
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import registry
-from sqlalchemy.orm import Relationship
from sqlalchemy.orm import relationship
+from sqlalchemy.orm import RelationshipProperty
reg: registry = registry()
# EXPECTED: Can't infer type from @declared_attr on function 'some_relationship' # noqa
@declared_attr
- # EXPECTED_MYPY: Missing type parameters for generic type "Relationship"
- def some_relationship(cls) -> Relationship:
+ # EXPECTED_MYPY: Missing type parameters for generic type "RelationshipProperty"
+ def some_relationship(cls) -> RelationshipProperty:
return relationship("Bar")
assert ASub.brap.property is A.data.property
assert isinstance(
- ASub.brap.original_property, descriptor_props.Synonym
+ ASub.brap.original_property, descriptor_props.SynonymProperty
)
def test_alt_name_attr_subclass_relationship_inline(self):
assert ASub.brap.property is A.b.property
assert isinstance(
- ASub.brap.original_property, descriptor_props.Synonym
+ ASub.brap.original_property, descriptor_props.SynonymProperty
)
ASub(brap=B())
A.brap = A.data
assert A.brap.property is A.data.property
- assert isinstance(A.brap.original_property, descriptor_props.Synonym)
+ assert isinstance(
+ A.brap.original_property, descriptor_props.SynonymProperty
+ )
def test_alt_name_attr_subclass_relationship_attrset(
self, require_metaclass
id = Column("id", Integer, primary_key=True)
assert A.brap.property is A.b.property
- assert isinstance(A.brap.original_property, descriptor_props.Synonym)
+ assert isinstance(
+ A.brap.original_property, descriptor_props.SynonymProperty
+ )
A(brap=B())
def test_eager_order_by(self):
from sqlalchemy.orm import load_only
from sqlalchemy.orm import reconstructor
from sqlalchemy.orm import registry
-from sqlalchemy.orm import Relationship
from sqlalchemy.orm import relationship
+from sqlalchemy.orm import RelationshipProperty
from sqlalchemy.orm import Session
from sqlalchemy.orm import synonym
from sqlalchemy.orm.persistence import _sort_states
# NOTE: this API changed in 0.8, previously __clause_element__()
# gave the parent selecatable, now it gives the
# primaryjoin/secondaryjoin
- class MyFactory(Relationship.Comparator):
+ class MyFactory(RelationshipProperty.Comparator):
__hash__ = None
def __eq__(self, other):
self._source_selectable().c.user_id
) == func.foobar(other.id)
- class MyFactory2(Relationship.Comparator):
+ class MyFactory2(RelationshipProperty.Comparator):
__hash__ = None
def __eq__(self, other):
lambda: (joinedload(Keyword.id).joinedload(Item.keywords),),
'Can\'t apply "joined loader" strategy to property "Keyword.id", '
'which is a "column property"; this loader strategy is intended '
- 'to be used with a "relationship".',
+ 'to be used with a "relationship property".',
)
def test_option_against_wrong_multi_entity_type_attr_two(self):
[Keyword, Item],
lambda: (joinedload(Keyword.keywords).joinedload(Item.keywords),),
'Can\'t apply "joined loader" strategy to property '
- '"Keyword.keywords", which is a "column property"; this loader '
- 'strategy is intended to be used with a "relationship".',
+ '"Keyword.keywords", which is a "mapped sql expression"; '
+ "this loader "
+ 'strategy is intended to be used with a "relationship property".',
)
def test_option_against_wrong_multi_entity_type_attr_three(self):
Column("x", Integer),
Column("y", Integer),
)
+
cls.right = Table(
"rgt",
m,
Column("x", Integer),
Column("y", Integer),
)
+
+ from sqlalchemy.orm import registry
+
+ reg = registry()
+
+ cls.relationship = relationship("Otherwise")
+
+ @reg.mapped
+ class Whatever:
+ __table__ = cls.left
+
+ foo = cls.relationship
+
+ @reg.mapped
+ class Otherwise:
+ __table__ = cls.right
+
+ reg.configure()
+
cls.right_multi_fk = Table(
"rgt_multi_fk",
m,
self.three_tab_b,
self.three_tab_a,
self.three_tab_b,
+ prop=self.relationship,
support_sync=False,
can_be_synced_fn=_can_sync,
primaryjoin=and_(
self.m2mright,
self.m2mleft,
self.m2mright,
+ prop=self.relationship,
secondary=self.m2msecondary,
**kw,
)
self.m2mleft,
self.m2mright,
self.m2mleft,
+ prop=self.relationship,
secondary=self.m2msecondary,
primaryjoin=j1.secondaryjoin_minus_local,
secondaryjoin=j1.primaryjoin_minus_local,
def _join_fixture_o2m(self, **kw):
return relationships.JoinCondition(
- self.left, self.right, self.left, self.right, **kw
+ self.left,
+ self.right,
+ self.left,
+ self.right,
+ prop=self.relationship,
+ **kw,
)
def _join_fixture_m2o(self, **kw):
return relationships.JoinCondition(
- self.right, self.left, self.right, self.left, **kw
+ self.right,
+ self.left,
+ self.right,
+ self.left,
+ prop=self.relationship,
+ **kw,
)
def _join_fixture_o2m_selfref(self, **kw):
return relationships.JoinCondition(
- self.selfref, self.selfref, self.selfref, self.selfref, **kw
+ self.selfref,
+ self.selfref,
+ self.selfref,
+ self.selfref,
+ prop=self.relationship,
+ **kw,
)
def _join_fixture_m2o_selfref(self, **kw):
self.selfref,
self.selfref,
self.selfref,
+ prop=self.relationship,
remote_side=set([self.selfref.c.id]),
**kw,
)
self.composite_selfref,
self.composite_selfref,
self.composite_selfref,
+ prop=self.relationship,
**kw,
)
self.composite_selfref,
self.composite_selfref,
self.composite_selfref,
+ prop=self.relationship,
remote_side=set(
[
self.composite_selfref.c.id,
self.composite_selfref,
self.composite_selfref,
self.composite_selfref,
+ prop=self.relationship,
primaryjoin=and_(
self.composite_selfref.c.group_id
== func.foo(self.composite_selfref.c.group_id),
self.composite_selfref,
self.composite_selfref,
self.composite_selfref,
+ prop=self.relationship,
primaryjoin=and_(
self.composite_selfref.c.group_id
== func.foo(self.composite_selfref.c.group_id),
self.composite_selfref,
self.composite_selfref,
self.composite_selfref,
+ prop=self.relationship,
primaryjoin=and_(
remote(self.composite_selfref.c.group_id)
== func.foo(self.composite_selfref.c.group_id),
self.right,
self.left,
self.right,
+ prop=self.relationship,
primaryjoin=(self.left.c.x + self.left.c.y)
== relationships.remote(
relationships.foreign(self.right.c.x * self.right.c.y)
self.right,
self.left,
self.right,
+ prop=self.relationship,
primaryjoin=(self.left.c.x + self.left.c.y)
== relationships.foreign(self.right.c.x * self.right.c.y),
**kw,
self.right,
self.left,
self.right,
+ prop=self.relationship,
primaryjoin=(self.left.c.x + self.left.c.y)
== (self.right.c.x * self.right.c.y),
**kw,
right,
self.base_w_sub_rel,
self.rel_sub,
+ prop=self.relationship,
primaryjoin=self.base_w_sub_rel.c.sub_id == self.rel_sub.c.id,
**kw,
)
self.base,
self.sub_w_base_rel,
self.base,
+ prop=self.relationship,
primaryjoin=self.sub_w_base_rel.c.base_id == self.base.c.id,
)
right,
self.sub,
self.sub_w_base_rel,
+ prop=self.relationship,
primaryjoin=self.sub_w_base_rel.c.base_id == self.base.c.id,
)
right,
self.sub,
self.sub_w_sub_rel,
+ prop=self.relationship,
primaryjoin=self.sub.c.id == self.sub_w_sub_rel.c.sub_id,
)
right,
self.right_w_base_rel,
self.right_w_base_rel,
+ prop=self.relationship,
)
def _join_fixture_m2o_sub_to_joined_sub_func(self, **kw):
right,
self.right_w_base_rel,
self.right_w_base_rel,
+ prop=self.relationship,
primaryjoin=self.right_w_base_rel.c.base_id
== func.foo(self.base.c.id),
)
left = self.base.join(self.sub, self.base.c.id == self.sub.c.id)
# see test_relationships->AmbiguousJoinInterpretedAsSelfRef
- return relationships.JoinCondition(left, self.sub, left, self.sub)
+ return relationships.JoinCondition(
+ left,
+ self.sub,
+ left,
+ self.sub,
+ prop=self.relationship,
+ )
def _join_fixture_o2m_to_annotated_func(self, **kw):
return relationships.JoinCondition(
self.right,
self.left,
self.right,
+ prop=self.relationship,
primaryjoin=self.left.c.id == foreign(func.foo(self.right.c.lid)),
**kw,
)
self.right,
self.left,
self.right,
+ prop=self.relationship,
primaryjoin=self.left.c.id == func.foo(self.right.c.lid),
consider_as_foreign_keys={self.right.c.lid},
**kw,
self.composite_multi_ref,
self.composite_target,
self.composite_multi_ref,
+ prop=self.relationship,
consider_as_foreign_keys={
self.composite_multi_ref.c.uid2,
self.composite_multi_ref.c.oid,
self.right,
self.left,
self.right,
+ prop=self.relationship,
primaryjoin=and_(
self.left.c.id == self.right.c.lid, self.left.c.x == 5
),
self.purely_single_col,
self.purely_single_col,
self.purely_single_col,
+ prop=self.relationship,
support_sync=False,
primaryjoin=self.purely_single_col.c.path.like(
remote(foreign(self.purely_single_col.c.path.concat("%")))
self.purely_single_col,
self.purely_single_col,
self.purely_single_col,
+ prop=self.relationship,
support_sync=False,
primaryjoin=remote(self.purely_single_col.c.path).like(
foreign(self.purely_single_col.c.path.concat("%"))
self.selfref,
self.selfref,
self.selfref,
+ prop=self.relationship,
support_sync=False,
primaryjoin=fn(
# we're putting a do-nothing annotation on
exc.SAWarning,
"Non-simple column elements in "
"primary join condition for property "
- r"None - consider using remote\(\) "
+ r"Whatever.foo - consider using remote\(\) "
"annotations to mark the remote side.",
fn,
)
self._assert_raises_no_relevant_fks(
self._join_fixture_compound_expression_1_non_annotated,
r"lft.x \+ lft.y = rgt.x \* rgt.y",
- "None",
+ "Whatever.foo",
"primary",
)
assert_raises_message(
exc.AmbiguousForeignKeysError,
"Could not determine join condition between "
- "parent/child tables on relationship None - "
+ "parent/child tables on relationship Whatever.foo - "
"there are multiple foreign key paths linking "
"the tables. Specify the 'foreign_keys' argument, "
"providing a list of those columns which "
self.right_multi_fk,
self.left,
self.right_multi_fk,
+ prop=self.relationship,
)
def test_determine_join_no_fks_o2m(self):
self._assert_raises_no_join(
relationships.JoinCondition,
- "None",
+ "Whatever.foo",
None,
self.left,
self.selfref,
self.left,
self.selfref,
+ prop=self.relationship,
)
def test_determine_join_ambiguous_fks_m2m(self):
self._assert_raises_ambig_join(
relationships.JoinCondition,
- "None",
+ "Whatever.foo",
self.m2msecondary_ambig_fks,
self.m2mleft,
self.m2mright,
self.m2mleft,
self.m2mright,
+ prop=self.relationship,
secondary=self.m2msecondary_ambig_fks,
)
def test_determine_join_no_fks_m2m(self):
self._assert_raises_no_join(
relationships.JoinCondition,
- "None",
+ "Whatever.foo",
self.m2msecondary_no_fks,
self.m2mleft,
self.m2mright,
self.m2mleft,
self.m2mright,
+ prop=self.relationship,
secondary=self.m2msecondary_no_fks,
)
self.m2mright,
self.m2mleft,
self.m2mright,
+ prop=self.relationship,
secondary=self.m2msecondary_ambig_fks,
consider_as_foreign_keys={
self.m2msecondary_ambig_fks.c.lid1,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
