:ref:`change_4393_convertunicode`
- :ref:`FIXME` - FIXME - link to non-primary mapper deprecation
+ :ref:`change_4423`
:ticket:`4393`
New Features and Improvements - ORM
===================================
+.. _change_4423:
+
+Relationship to AliasedClass replaces the need for non primary mappers
+-----------------------------------------------------------------------
+
+The "non primary mapper" is a :func:`.mapper` created in the
+:ref:`classical_mapping` style, which acts as an additional mapper against an
+already mapped class against a different kind of selectable. The non primary
+mapper has its roots in the 0.1, 0.2 series of SQLAlchemy where it was
+anticipated that the :func:`.mapper` object was to be the primary query
+construction interface, before the :class:`.Query` object existed.
+
+With the advent of :class:`.Query` and later the :class:`.AliasedClass`
+construct, most use cases for the non primary mapper went away. This was a
+good thing since SQLAlchemy also moved away from "classical" mappings altogether
+around the 0.5 series in favor of the declarative system.
+
+One use case remained around for non primary mappers when it was realized that
+some very hard-to-define :func:`.relationship` configurations could be made
+possible when a non-primary mapper with an alternative selectable was made as
+the mapping target, rather than trying to construct a
+:paramref:`.relationship.primaryjoin` that encompassed all the complexity of a
+particular inter-object relationship.
+
+As this use case became more popular, its limitations became apparent,
+including that the non primary mapper is difficult to configure against a
+selectable that adds new columns, that the mapper does not inherit the
+relationships of the original mapping, that relationships which are configured
+explicitly on the non primary mapper do not function well with loader options,
+and that the non primary mapper also doesn't provide a fully functional
+namespace of column-based attributes which can be used in queries (which again,
+in the old 0.1 - 0.4 days, one would use :class:`.Table` objects directly with
+the ORM).
+
+The missing piece was to allow the :func:`.relationship` to refer directly
+to the :class:`.AliasedClass`. The :class:`.AliasedClass` already does
+everything we want the non primary mapper to do; it allows an existing mapped
+class to be loaded from an alternative selectable, it inherits all the
+attributes and relationships of the existing mapper, it works
+extremely well with loader options, and it provides a class-like
+object that can be mixed into queries just like the class itself.
+With this change, the recipes that
+were formerly for non primary mappers at :ref:`relationship_configure_joins`
+are changed to aliased class.
+
+At :ref:`relationship_aliased_class`, the original non primary mapper looked
+like::
+
+ j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
+
+ B_viacd = mapper(
+ B, j, non_primary=True, primary_key=[j.c.b_id],
+ properties={
+ "id": j.c.b_id, # so that 'id' looks the same as before
+ "c_id": j.c.c_id, # needed for disambiguation
+ "d_c_id": j.c.d_c_id, # needed for disambiguation
+ "b_id": [j.c.b_id, j.c.d_b_id],
+ "d_id": j.c.d_id,
+ }
+ )
+
+ A.b = relationship(B_viacd, primaryjoin=A.b_id == B_viacd.c.b_id)
+
+The properties were necessary in order to re-map the additional columns
+so that they did not conflict with the existing columns mapped to ``B``, as
+well as it was necessary to define a new primary key.
+
+With the new approach, all of this verbosity goes away, and the additional
+columns are referred towards directly when making the relationship::
+
+ j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
+
+ B_viacd = aliased(B, j, flat=True)
+
+ A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
+
+The non primary mapper is now deprecated with the eventual goal to be that
+classical mappings as a feature go away entirely. The Declarative API would
+become the single means of mapping which hopefully will allow internal
+improvements and simplifications, as well as a clearer documentation story.
+
+
+:ticket:`4423`
+
+
.. _change_4340:
selectin loading no longer uses JOIN for simple one-to-many
--- /dev/null
+.. change::
+ :tags: feature, orm
+ :tickets: 4423
+
+ Implemented a new feature whereby the :class:`.AliasedClass` construct can
+ now be used as the target of a :func:`.relationship`. This allows the
+ concept of "non primary mappers" to no longer be necessary, as the
+ :class:`.AliasedClass` is much easier to configure and automatically inherits
+ all the relationships of the mapped class, as well as preserves the
+ ability for loader options to work normally.
+
+ .. seealso::
+
+ :ref:`change_4423`
\ No newline at end of file
as well as support within declarative to specify complex conditions such
as joins involving class names as targets.
-.. _relationship_non_primary_mapper:
+.. _relationship_aliased_class:
-Relationship to Non Primary Mapper
+Relationship to Aliased Class
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+.. versionadded:: 1.3
+ The :class:`.AliasedClass` construct can now be specified as the
+ target of a :func:`.relationship`, replacing the previous approach
+ of using non-primary mappers, which had limitations such that they did
+ not inherit sub-relationships of the mapped entity as well as that they
+ required complex configuration against an alternate selectable. The
+ recipes in this section are now updated to use :class:`.AliasedClass`.
+
In the previous section, we illustrated a technique where we used
:paramref:`~.relationship.secondary` in order to place additional
tables within a join condition. There is one complex join case where
*directly*. In this case, the join from ``A`` to ``B`` may be
difficult to express with just a complex
:paramref:`~.relationship.primaryjoin` condition, as the intermediary
-tables may need special handling, and it is also not expressable with
+tables may need special handling, and it is also not expressible with
a :paramref:`~.relationship.secondary` object, since the
``A->secondary->B`` pattern does not support any references between
``A`` and ``B`` directly. When this **extremely advanced** case
arises, we can resort to creating a second mapping as a target for the
-relationship. This is where we use :func:`.mapper` in order to make a
+relationship. This is where we use :class:`.AliasedClass` in order to make a
mapping to a class that includes all the additional tables we need for
this join. In order to produce this mapper as an "alternative" mapping
-for our class, we use the :paramref:`~.mapper.non_primary` flag.
+for our class, we use the :func:`.aliased` function to produce the new
+construct, then use :func:`.relationship` against the object as though it
+were a plain mapped class.
Below illustrates a :func:`.relationship` with a simple join from ``A`` to
``B``, however the primaryjoin condition is augmented with two additional
# to it in the mapping multiple times.
j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
- # 2. Create a new mapper() to B, with non_primary=True.
- # Columns in the join with the same name must be
- # disambiguated within the mapping, using named properties.
- # we also have to make sure the primary key of the regular "B"
- # mapping is maintained.
- B_viacd = mapper(
- B, j, non_primary=True, primary_key=[j.c.b_id],
- properties={
- "id": j.c.b_id, # so that 'id' looks the same as before
- "c_id": j.c.c_id, # needed for disambiguation
- "d_c_id": j.c.d_c_id, # needed for disambiguation
- "b_id": [j.c.b_id, j.c.d_b_id],
- "d_id": j.c.d_id,
- }
- )
+ # 2. Create an AliasedClass to B
+ B_viacd = aliased(B, j, flat=True)
- A.b = relationship(B_viacd, primaryjoin=A.b_id == B_viacd.c.b_id)
+ A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
-In the above case, our non-primary mapper for ``B`` will emit for
-additional columns when we query; these can be ignored:
+With the above mapping, a simple join looks like:
.. sourcecode:: python+sql
{opensql}SELECT a.id AS a_id, a.b_id AS a_b_id
FROM a JOIN (b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) ON a.b_id = b.id
+.. _relationship_to_window_function:
+
Row-Limited Relationships with Window Functions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-Another interesting use case for non-primary mappers are situations where
+Another interesting use case for relationships to :class:`.AliasedClass`
+objects are situations where
the relationship needs to join to a specialized SELECT of any form. One
scenario is when the use of a window function is desired, such as to limit
how many rows should be returned for a relationship. The example below
).label('index')
]).alias()
- partitioned_b = mapper(B, partition, non_primary=True)
+ partitioned_b = aliased(B, partition)
A.partitioned_bs = relationship(
partitioned_b,
- primaryjoin=and_(partitioned_b.c.a_id == A.id, partitioned_b.c.index < 10)
+ primaryjoin=and_(partitioned_b.a_id == A.id, partition.c.index < 10)
)
We can use the above ``partitioned_bs`` relationship with most of the loader
# dict_ will be a dictproxy, which we can't write to, and we need to!
self.dict_ = dict(dict_)
self.classname = classname
- self.mapped_table = None
+ self.persist_selectable = None
self.properties = util.OrderedDict()
self.declared_columns = set()
self.column_copies = {}
elif self.inherits:
inherited_mapper = _declared_mapping_info(self.inherits)
inherited_table = inherited_mapper.local_table
- inherited_mapped_table = inherited_mapper.mapped_table
+ inherited_persist_selectable = inherited_mapper.persist_selectable
if table is None:
# single table inheritance.
)
inherited_table.append_column(c)
if (
- inherited_mapped_table is not None
- and inherited_mapped_table is not inherited_table
+ inherited_persist_selectable is not None
+ and inherited_persist_selectable is not inherited_table
):
- inherited_mapped_table._refresh_for_new_column(c)
+ inherited_persist_selectable._refresh_for_new_column(c)
def _prepare_mapper_arguments(self):
properties = self.properties
def set_parent(self, parent, init):
if self.map_column:
# implement the 'map_column' option.
- if self.key not in parent.mapped_table.c:
+ if self.key not in parent.persist_selectable.c:
raise sa_exc.ArgumentError(
"Can't compile synonym '%s': no column on table "
"'%s' named '%s'"
- % (self.name, parent.mapped_table.description, self.key)
+ % (
+ self.name,
+ parent.persist_selectable.description,
+ self.key,
+ )
)
elif (
- parent.mapped_table.c[self.key] in parent._columntoproperty
+ parent.persist_selectable.c[self.key]
+ in parent._columntoproperty
and parent._columntoproperty[
- parent.mapped_table.c[self.key]
+ parent.persist_selectable.c[self.key]
].key
== self.name
):
"%r for column %r"
% (self.key, self.name, self.name, self.key)
)
- p = properties.ColumnProperty(parent.mapped_table.c[self.key])
+ p = properties.ColumnProperty(
+ parent.persist_selectable.c[self.key]
+ )
parent._configure_property(self.name, p, init=init, setparent=True)
p._mapped_by_synonym = self.key
"""
return {}
- def setup(self, context, entity, path, adapter, **kwargs):
+ def setup(self, context, query_entity, path, adapter, **kwargs):
"""Called by Query for the purposes of constructing a SQL statement.
Each MapperProperty associated with the target mapper processes the
)
return strategy
- def setup(self, context, entity, path, adapter, **kwargs):
+ def setup(self, context, query_entity, path, adapter, **kwargs):
loader = self._get_context_loader(context, path)
if loader and loader.strategy:
strat = self._get_strategy(loader.strategy)
else:
strat = self.strategy
- strat.setup_query(context, entity, path, loader, adapter, **kwargs)
+ strat.setup_query(
+ context, query_entity, path, loader, adapter, **kwargs
+ )
def create_row_processor(
self, context, path, mapper, result, adapter, populators
def init_class_attribute(self, mapper):
pass
- def setup_query(self, context, entity, path, loadopt, adapter, **kwargs):
+ def setup_query(
+ self, context, query_entity, path, loadopt, adapter, **kwargs
+ ):
"""Establish column and other state for a given QueryContext.
This method fulfills the contract specified by MapperProperty.setup().
),
order_by=(
"1.1",
- "The :paramref:`.Mapper.order_by` parameter "
+ "The :paramref:`.mapper.order_by` parameter "
"is deprecated, and will be removed in a future release. "
"Use :meth:`.Query.order_by` to determine the ordering of a "
"result set.",
),
+ non_primary=(
+ "1.3",
+ "The :paramref:`.mapper.non_primary` parameter is deprecated, "
+ "and will be removed in a future release. The functionality "
+ "of non primary mappers is now better suited using the "
+ ":class:`.AliasedClass` construct, which can also be used "
+ "as the target of a :func:`.relationship` in 1.3.",
+ ),
)
def __init__(
self,
.. seealso::
- :attr:`~.Mapper.mapped_table`.
+ :attr:`~.Mapper.persist_selectable`.
"""
- mapped_table = None
+ persist_selectable = None
"""The :class:`.Selectable` to which this :class:`.Mapper` is mapped.
Typically an instance of :class:`.Table`, :class:`.Join`, or
:class:`.Alias`.
- The "mapped" table is the selectable that
- the mapper selects from during queries. For non-inheriting
- mappers, the mapped table is the same as the "local" table.
- For joined-table inheritance mappers, mapped_table references the
- full :class:`.Join` representing full rows for this particular
- subclass. For single-table inheritance mappers, mapped_table
- references the base table.
+ The :attr:`.Mapper.persist_selectable` is separate from
+ :attr:`.Mapper.selectable` in that the former represents columns
+ that are mapped on this class or its superclasses, whereas the
+ latter may be a "polymorphic" selectable that contains additional columns
+ which are in fact mapped on subclasses only.
+
+ "persist selectable" is the "thing the mapper writes to" and
+ "selectable" is the "thing the mapper selects from".
+
+ :attr:`.Mapper.persist_selectable` is also separate from
+ :attr:`.Mapper.local_table`, which represents the set of columns that
+ are locally mapped on this class directly.
+
.. seealso::
+ :attr:`~.Mapper.selectable`.
+
:attr:`~.Mapper.local_table`.
"""
which comprise the 'primary key' of the mapped table, from the
perspective of this :class:`.Mapper`.
- This list is against the selectable in :attr:`~.Mapper.mapped_table`. In
- the case of inheriting mappers, some columns may be managed by a
+ This list is against the selectable in :attr:`~.Mapper.persist_selectable`.
+ In the case of inheriting mappers, some columns may be managed by a
superclass mapper. For example, in the case of a :class:`.Join`, the
primary key is determined by all of the primary key columns across all
tables referenced by the :class:`.Join`.
c = None
"""A synonym for :attr:`~.Mapper.columns`."""
+ @property
+ @util.deprecated("1.3", "Use .persist_selectable")
+ def mapped_table(self):
+ return self.persist_selectable
+
@util.memoized_property
def _path_registry(self):
return PathRegistry.per_mapper(self)
# inherit_condition is optional.
if self.local_table is None:
self.local_table = self.inherits.local_table
- self.mapped_table = self.inherits.mapped_table
+ self.persist_selectable = self.inherits.persist_selectable
self.single = True
elif self.local_table is not self.inherits.local_table:
if self.concrete:
- self.mapped_table = self.local_table
+ self.persist_selectable = self.local_table
for mapper in self.iterate_to_root():
if mapper.polymorphic_on is not None:
mapper._requires_row_aliasing = True
self.inherit_condition = sql_util.join_condition(
self.inherits.local_table, self.local_table
)
- self.mapped_table = sql.join(
- self.inherits.mapped_table,
+ self.persist_selectable = sql.join(
+ self.inherits.persist_selectable,
self.local_table,
self.inherit_condition,
)
fks = util.to_set(self.inherit_foreign_keys)
self._inherits_equated_pairs = sql_util.criterion_as_pairs(
- self.mapped_table.onclause,
+ self.persist_selectable.onclause,
consider_as_foreign_keys=fks,
)
else:
- self.mapped_table = self.local_table
+ self.persist_selectable = self.local_table
if self.polymorphic_identity is not None and not self.concrete:
self._identity_class = self.inherits._identity_class
else:
self._all_tables = set()
self.base_mapper = self
- self.mapped_table = self.local_table
+ self.persist_selectable = self.local_table
if self.polymorphic_identity is not None:
self.polymorphic_map[self.polymorphic_identity] = self
self._identity_class = self.class_
- if self.mapped_table is None:
+ if self.persist_selectable is None:
raise sa_exc.ArgumentError(
- "Mapper '%s' does not have a mapped_table specified." % self
+ "Mapper '%s' does not have a persist_selectable specified."
+ % self
)
def _set_with_polymorphic(self, with_polymorphic):
instrumentation.unregister_class(self.class_)
def _configure_pks(self):
- self.tables = sql_util.find_tables(self.mapped_table)
+ self.tables = sql_util.find_tables(self.persist_selectable)
self._pks_by_table = {}
self._cols_by_table = {}
pk_cols = util.column_set(c for c in all_cols if c.primary_key)
# identify primary key columns which are also mapped by this mapper.
- tables = set(self.tables + [self.mapped_table])
+ tables = set(self.tables + [self.persist_selectable])
self._all_tables.update(tables)
for t in tables:
if t.primary_key and pk_cols.issuperset(t.primary_key):
# otherwise, see that we got a full PK for the mapped table
elif (
- self.mapped_table not in self._pks_by_table
- or len(self._pks_by_table[self.mapped_table]) == 0
+ self.persist_selectable not in self._pks_by_table
+ or len(self._pks_by_table[self.persist_selectable]) == 0
):
raise sa_exc.ArgumentError(
"Mapper %s could not assemble any primary "
"key columns for mapped table '%s'"
- % (self, self.mapped_table.description)
+ % (self, self.persist_selectable.description)
)
elif self.local_table not in self._pks_by_table and isinstance(
self.local_table, schema.Table
# that of the inheriting (unless concrete or explicit)
self.primary_key = self.inherits.primary_key
else:
- # determine primary key from argument or mapped_table pks -
+ # determine primary key from argument or persist_selectable pks -
# reduce to the minimal set of columns
if self._primary_key_argument:
primary_key = sql_util.reduce_columns(
[
- self.mapped_table.corresponding_column(c)
+ self.persist_selectable.corresponding_column(c)
for c in self._primary_key_argument
],
ignore_nonexistent_tables=True,
)
else:
primary_key = sql_util.reduce_columns(
- self._pks_by_table[self.mapped_table],
+ self._pks_by_table[self.persist_selectable],
ignore_nonexistent_tables=True,
)
raise sa_exc.ArgumentError(
"Mapper %s could not assemble any primary "
"key columns for mapped table '%s'"
- % (self, self.mapped_table.description)
+ % (self, self.persist_selectable.description)
)
self.primary_key = tuple(primary_key)
# create properties for each column in the mapped table,
# for those columns which don't already map to a property
- for column in self.mapped_table.columns:
+ for column in self.persist_selectable.columns:
if column in self._columntoproperty:
continue
# doesn't appear to be mapped. this means it can be 1.
# only present in the with_polymorphic selectable or
# 2. a totally standalone SQL expression which we'd
- # hope is compatible with this mapper's mapped_table
- col = self.mapped_table.corresponding_column(
+ # hope is compatible with this mapper's persist_selectable
+ col = self.persist_selectable.corresponding_column(
self.polymorphic_on
)
if col is None:
# for it. Just check that if it's directly a
# schema.Column and we have with_polymorphic, it's
# likely a user error if the schema.Column isn't
- # represented somehow in either mapped_table or
+ # represented somehow in either persist_selectable or
# with_polymorphic. Otherwise as of 0.7.4 we
# just go with it and assume the user wants it
# that way (i.e. a CASE statement)
# table is the same as the parent (i.e. single table
# inheritance), we can use it
if mapper.polymorphic_on is not None:
- if self.mapped_table is mapper.mapped_table:
+ if self.persist_selectable is mapper.persist_selectable:
self.polymorphic_on = mapper.polymorphic_on
else:
self.polymorphic_on = (
- self.mapped_table.corresponding_column
- )(mapper.polymorphic_on)
+ self.persist_selectable
+ ).corresponding_column(mapper.polymorphic_on)
# we can use the parent mapper's _set_polymorphic_identity
# directly; it ensures the polymorphic_identity of the
# instance's mapper is used so is portable to subclasses.
stack = deque([self])
while stack:
item = stack.popleft()
- if item.mapped_table is self.mapped_table:
+ if item.persist_selectable is self.persist_selectable:
identities.add(item.polymorphic_identity)
stack.extend(item._inheriting_mappers)
prop = self._property_from_column(key, prop)
if isinstance(prop, properties.ColumnProperty):
- col = self.mapped_table.corresponding_column(prop.columns[0])
+ col = self.persist_selectable.corresponding_column(prop.columns[0])
# if the column is not present in the mapped table,
# test if a column has been added after the fact to the
col = m.local_table.corresponding_column(prop.columns[0])
if col is not None:
for m2 in path:
- m2.mapped_table._reset_exported()
- col = self.mapped_table.corresponding_column(
+ m2.persist_selectable._reset_exported()
+ col = self.persist_selectable.corresponding_column(
prop.columns[0]
)
break
):
mapped_column = []
for c in columns:
- mc = self.mapped_table.corresponding_column(c)
+ mc = self.persist_selectable.corresponding_column(c)
if mc is None:
mc = self.local_table.corresponding_column(c)
if mc is not None:
# mapped table, this corresponds to adding a
# column after the fact to the local table.
# [ticket:1523]
- self.mapped_table._reset_exported()
- mc = self.mapped_table.corresponding_column(c)
+ self.persist_selectable._reset_exported()
+ mc = self.persist_selectable.corresponding_column(c)
if mc is None:
raise sa_exc.ArgumentError(
"When configuring property '%s' on %s, "
mapped tables.
"""
- from_obj = self.mapped_table
+ from_obj = self.persist_selectable
for m in mappers:
if m is self:
continue
@_memoized_configured_property
def _with_polymorphic_selectable(self):
if not self.with_polymorphic:
- return self.mapped_table
+ return self.persist_selectable
spec, selectable = self.with_polymorphic
if selectable is not None:
"""The :func:`.select` construct this :class:`.Mapper` selects from
by default.
- Normally, this is equivalent to :attr:`.mapped_table`, unless
+ Normally, this is equivalent to :attr:`.persist_selectable`, unless
the ``with_polymorphic`` feature is in use, in which case the
full "polymorphic" selectable is returned.
and ext_info.mapper.with_polymorphic
):
if (
- ext_info.mapper.mapped_table
+ ext_info.mapper.persist_selectable
not in self._polymorphic_adapters
):
self._mapper_loads_polymorphically_with(
if of_type:
right = of_type
else:
- right = onclause.property.mapper
+ right = onclause.property.entity
left = onclause._parententity
# be more liberal about auto-aliasing.
if right_mapper and (
right_mapper.with_polymorphic
- or isinstance(right_mapper.mapped_table, expression.Join)
+ or isinstance(right_mapper.persist_selectable, expression.Join)
):
for from_obj in self._from_obj or [l_info.selectable]:
if sql_util.selectables_overlap(
# as the ON clause
if not right_selectable.is_derived_from(
- right_mapper.mapped_table
+ right_mapper.persist_selectable
):
raise sa_exc.InvalidRequestError(
"Selectable '%s' is not derived from '%s'"
% (
right_selectable.description,
- right_mapper.mapped_table.description,
+ right_mapper.persist_selectable.description,
)
)
def process_query(self, query):
if isinstance(self.alias, util.string_types):
- alias = query._mapper_zero().mapped_table.alias(self.alias)
+ alias = query._mapper_zero().persist_selectable.alias(self.alias)
else:
alias = self.alias
query._from_obj_alias = sql_util.ColumnAdapter(alias)
of_type=self._of_type,
)
+ @util.memoized_property
+ def entity(self):
+ """The target entity referred to by this
+ :class:`.RelationshipProperty.Comparator`.
+
+ This is either a :class:`.Mapper` or :class:`.AliasedInsp`
+ object.
+
+ This is the "target" or "remote" side of the
+ :func:`.relationship`.
+
+ """
+ return self.property.entity
+
@util.memoized_property
def mapper(self):
"""The target :class:`.Mapper` referred to by this
def __clause_element__(self):
adapt_from = self._source_selectable()
if self._of_type:
- of_type = inspect(self._of_type).mapper
+ of_type_mapper = inspect(self._of_type).mapper
else:
- of_type = None
+ of_type_mapper = None
(
pj,
) = self.property._create_joins(
source_selectable=adapt_from,
source_polymorphic=True,
- of_type=of_type,
+ of_type_mapper=of_type_mapper,
)
if sj is not None:
return pj & sj
)
@util.memoized_property
- def mapper(self):
- """Return the targeted :class:`.Mapper` for this
- :class:`.RelationshipProperty`.
-
- This is a lazy-initializing static attribute.
+ def entity(self): # type: () -> Union[AliasedInsp, Mapper]
+ """Return the target mapped entity, which is an inspect() of the
+ class or aliased class tha is referred towards.
"""
if util.callable(self.argument) and not isinstance(
argument = self.argument
if isinstance(argument, type):
- mapper_ = mapperlib.class_mapper(argument, configure=False)
- elif isinstance(self.argument, mapperlib.Mapper):
- mapper_ = argument
+ return mapperlib.class_mapper(argument, configure=False)
+
+ try:
+ entity = inspect(argument)
+ except sa_exc.NoInspectionAvailable:
+ pass
else:
- raise sa_exc.ArgumentError(
- "relationship '%s' expects "
- "a class or a mapper argument (received: %s)"
- % (self.key, type(argument))
- )
- return mapper_
+ if hasattr(entity, "mapper"):
+ return entity
+
+ raise sa_exc.ArgumentError(
+ "relationship '%s' expects "
+ "a class or a mapper argument (received: %s)"
+ % (self.key, type(argument))
+ )
@util.memoized_property
- @util.deprecated("0.7", "Use .target")
- def table(self):
- """Return the selectable linked to this
- :class:`.RelationshipProperty` object's target
- :class:`.Mapper`.
+ def mapper(self):
+ """Return the targeted :class:`.Mapper` for this
+ :class:`.RelationshipProperty`.
+
+ This is a lazy-initializing static attribute.
+
"""
- return self.target
+ return self.entity.mapper
def do_init(self):
self._check_conflicts()
for x in util.to_column_set(self.remote_side)
)
- self.target = self.mapper.mapped_table
+ self.target = self.entity.persist_selectable
def _setup_join_conditions(self):
self._join_condition = jc = JoinCondition(
- parent_selectable=self.parent.mapped_table,
- child_selectable=self.mapper.mapped_table,
+ parent_persist_selectable=self.parent.persist_selectable,
+ child_persist_selectable=self.entity.persist_selectable,
parent_local_selectable=self.parent.local_table,
- child_local_selectable=self.mapper.local_table,
+ child_local_selectable=self.entity.local_table,
primaryjoin=self.primaryjoin,
secondary=self.secondary,
secondaryjoin=self.secondaryjoin,
and self.secondary.c.contains_column(c)
):
continue
- if not self.parent.mapped_table.c.contains_column(
+ if not self.parent.persist_selectable.c.contains_column(
c
) and not self.target.c.contains_column(c):
return False
source_selectable=None,
dest_polymorphic=False,
dest_selectable=None,
- of_type=None,
+ of_type_mapper=None,
):
if source_selectable is None:
if source_polymorphic and self.parent.with_polymorphic:
aliased = False
if dest_selectable is None:
+ dest_selectable = self.entity.selectable
if dest_polymorphic and self.mapper.with_polymorphic:
- dest_selectable = self.mapper._with_polymorphic_selectable
aliased = True
- else:
- dest_selectable = self.mapper.mapped_table
if self._is_self_referential and source_selectable is None:
dest_selectable = dest_selectable.alias()
else:
aliased = True
- dest_mapper = of_type or self.mapper
+ dest_mapper = of_type_mapper or self.mapper
single_crit = dest_mapper._single_table_criterion
aliased = aliased or (source_selectable is not None)
if source_selectable is None:
source_selectable = self.parent.local_table
if dest_selectable is None:
- dest_selectable = self.mapper.local_table
+ dest_selectable = self.entity.local_table
return (
primaryjoin,
secondaryjoin,
class JoinCondition(object):
def __init__(
self,
- parent_selectable,
- child_selectable,
+ parent_persist_selectable,
+ child_persist_selectable,
parent_local_selectable,
child_local_selectable,
primaryjoin=None,
support_sync=True,
can_be_synced_fn=lambda *c: True,
):
- self.parent_selectable = parent_selectable
+ self.parent_persist_selectable = parent_persist_selectable
self.parent_local_selectable = parent_local_selectable
- self.child_selectable = child_selectable
+ self.child_persist_selectable = child_persist_selectable
self.child_local_selectable = child_local_selectable
self.parent_equivalents = parent_equivalents
self.child_equivalents = child_equivalents
if self.secondary is not None:
if self.secondaryjoin is None:
self.secondaryjoin = join_condition(
- self.child_selectable,
+ self.child_persist_selectable,
self.secondary,
a_subset=self.child_local_selectable,
consider_as_foreign_keys=consider_as_foreign_keys,
)
if self.primaryjoin is None:
self.primaryjoin = join_condition(
- self.parent_selectable,
+ self.parent_persist_selectable,
self.secondary,
a_subset=self.parent_local_selectable,
consider_as_foreign_keys=consider_as_foreign_keys,
else:
if self.primaryjoin is None:
self.primaryjoin = join_condition(
- self.parent_selectable,
- self.child_selectable,
+ self.parent_persist_selectable,
+ self.child_persist_selectable,
a_subset=self.parent_local_selectable,
consider_as_foreign_keys=consider_as_foreign_keys,
)
comparisons where both columns are in both tables.
"""
- pt = self.parent_selectable
- mt = self.child_selectable
+ pt = self.parent_persist_selectable
+ mt = self.child_persist_selectable
result = [False]
def visit_binary(binary):
"""Return True if parent/child tables have some overlap."""
return selectables_overlap(
- self.parent_selectable, self.child_selectable
+ self.parent_persist_selectable, self.child_persist_selectable
)
def _annotate_remote(self):
if isinstance(left, expression.ColumnClause) and isinstance(
right, expression.ColumnClause
):
- if self.child_selectable.c.contains_column(
+ if self.child_persist_selectable.c.contains_column(
right
- ) and self.parent_selectable.c.contains_column(left):
+ ) and self.parent_persist_selectable.c.contains_column(left):
right = right._annotate({"remote": True})
elif (
check_entities
"""
def repl(element):
- if self.child_selectable.c.contains_column(element) and (
+ if self.child_persist_selectable.c.contains_column(element) and (
not self.parent_local_selectable.c.contains_column(element)
or self.child_local_selectable.c.contains_column(element)
):
[l for (l, r) in self._local_remote_pairs]
)
else:
- local_side = util.column_set(self.parent_selectable.c)
+ local_side = util.column_set(self.parent_persist_selectable.c)
def locals_(elem):
if "remote" not in elem._annotations and elem in local_side:
if self.secondaryjoin is not None:
self.direction = MANYTOMANY
else:
- parentcols = util.column_set(self.parent_selectable.c)
- targetcols = util.column_set(self.child_selectable.c)
+ parentcols = util.column_set(self.parent_persist_selectable.c)
+ targetcols = util.column_set(self.child_persist_selectable.c)
# fk collection which suggests ONETOMANY.
onetomany_fk = targetcols.intersection(self.foreign_key_columns)
if cls in self.__binds:
return self.__binds[cls]
if clause is None:
- clause = mapper.mapped_table
+ clause = mapper.persist_selectable
if clause is not None:
for t in sql_util.find_tables(clause, include_crud=True):
if isinstance(clause, sql.expression.ClauseElement) and clause.bind:
return clause.bind
- if mapper and mapper.mapped_table.bind:
- return mapper.mapped_table.bind
+ if mapper and mapper.persist_selectable.bind:
+ return mapper.persist_selectable.bind
context = []
if mapper is not None:
def setup_query(
self,
context,
- entity,
+ query_entity,
path,
loadopt,
adapter,
def setup_query(
self,
context,
- entity,
+ query_entity,
path,
loadopt,
adapter,
def setup_query(
self,
context,
- entity,
+ query_entity,
path,
loadopt,
adapter,
def setup_query(
self,
context,
- entity,
+ query_entity,
path,
loadopt,
adapter,
(("deferred", False), ("instrument", True))
).setup_query(
context,
- entity,
+ query_entity,
path,
loadopt,
adapter,
class AbstractRelationshipLoader(LoaderStrategy):
"""LoaderStratgies which deal with related objects."""
- __slots__ = "mapper", "target", "uselist"
+ __slots__ = "mapper", "target", "uselist", "entity"
def __init__(self, parent, strategy_key):
super(AbstractRelationshipLoader, self).__init__(parent, strategy_key)
self.mapper = self.parent_property.mapper
+ self.entity = self.parent_property.entity
self.target = self.parent_property.target
self.uselist = self.parent_property.uselist
"_lazywhere",
"_rev_lazywhere",
"use_get",
+ "is_aliased_class",
"_bind_to_col",
"_equated_columns",
"_rev_bind_to_col",
self._raise_always = self.strategy_opts["lazy"] == "raise"
self._raise_on_sql = self.strategy_opts["lazy"] == "raise_on_sql"
+ self.is_aliased_class = inspect(self.entity).is_aliased_class
+
join_condition = self.parent_property._join_condition
self._lazywhere, self._bind_to_col, self._equated_columns = (
join_condition.create_lazy_clause()
# determine if our "lazywhere" clause is the same as the mapper's
# get() clause. then we can just use mapper.get()
- self.use_get = not self.uselist and self.mapper._get_clause[0].compare(
- self._lazywhere,
- use_proxies=True,
- equivalents=self.mapper._equivalent_columns,
+ self.use_get = (
+ not self.is_aliased_class
+ and not self.uselist
+ and self.entity._get_clause[0].compare(
+ self._lazywhere,
+ use_proxies=True,
+ equivalents=self.mapper._equivalent_columns,
+ )
)
if self.use_get:
# does this, including how it decides what the correct
# identity_token would be for this identity.
instance = session.query()._identity_lookup(
- self.mapper,
+ self.entity,
primary_key_identity,
passive=passive,
lazy_loaded_from=state,
# lazy loaders. Currently the LRU cache is local to the LazyLoader,
# however add ourselves to the initial cache key just to future
# proof in case it moves
- q = self._bakery(lambda session: session.query(self.mapper), self)
+ q = self._bakery(lambda session: session.query(self.entity), self)
q.add_criteria(
lambda q: q._adapt_all_clauses()._with_invoke_all_eagers(False),
if with_poly_info is not None:
effective_entity = with_poly_info.entity
else:
- effective_entity = self.mapper
+ effective_entity = self.entity
subq_path = context.attributes.get(
("subquery_path", None), orm_util.PathRegistry.root
def setup_query(
self,
context,
- entity,
+ query_entity,
path,
loadopt,
adapter,
adapter,
add_to_collection,
) = self._setup_query_on_user_defined_adapter(
- context, entity, path, adapter, user_defined_adapter
+ context, query_entity, path, adapter, user_defined_adapter
)
else:
# if not via query option, check for
chained_from_outerjoin,
) = self._generate_row_adapter(
context,
- entity,
+ query_entity,
path,
loadopt,
adapter,
else:
with_polymorphic = None
- path = path[self.mapper]
+ path = path[self.entity]
loading._setup_entity_query(
context,
self.mapper,
- entity,
+ query_entity,
path,
clauses,
add_to_collection,
root_mapper, prop = path[-2:]
- # from .mapper import Mapper
- # from .interfaces import MapperProperty
- # assert isinstance(root_mapper, Mapper)
- # assert isinstance(prop, MapperProperty)
-
if alias is not None:
if isinstance(alias, str):
alias = prop.target.alias(alias)
# we need one unique AliasedClass per query per appearance of our
# entity in the query.
+ if inspect(self.entity).is_aliased_class:
+ alt_selectable = inspect(self.entity).selectable
+ else:
+ alt_selectable = None
+
key = ("joinedloader_ac", self)
if key not in context.attributes:
context.attributes[key] = idx = 0
if idx >= len(self._aliased_class_pool):
to_adapt = orm_util.AliasedClass(
- self.mapper, flat=True, use_mapper_path=True
+ self.mapper,
+ alias=alt_selectable.alias(flat=True)
+ if alt_selectable is not None
+ else None,
+ flat=True,
+ use_mapper_path=True,
)
+
# load up the .columns collection on the Alias() before
# the object becomes shared among threads. this prevents
# races for column identities.
def _create_eager_join(
self,
context,
- entity,
+ query_entity,
path,
adapter,
parentmapper,
):
if parentmapper is None:
- localparent = entity.mapper
+ localparent = query_entity.mapper
else:
localparent = parentmapper
and context.query._should_nest_selectable
)
- entity_key = None
+ query_entity_key = None
if (
- entity not in context.eager_joins
+ query_entity not in context.eager_joins
and not should_nest_selectable
and context.from_clause
):
indexes = sql_util.find_left_clause_that_matches_given(
- context.from_clause, entity.selectable
+ context.from_clause, query_entity.selectable
)
if len(indexes) > 1:
# for the eager load case, I can't reproduce this right
# now. For query.join() I can.
raise sa_exc.InvalidRequestError(
- "Can't identify which entity in which to joined eager "
- "load from. Please use an exact match when specifying "
- "the join path."
+ "Can't identify which query entity in which to joined "
+ "eager load from. Please use an exact match when "
+ "specifying the join path."
)
if indexes:
# key it to its list index in the eager_joins dict.
# Query._compile_context will adapt as needed and
# append to the FROM clause of the select().
- entity_key, default_towrap = indexes[0], clause
+ query_entity_key, default_towrap = indexes[0], clause
- if entity_key is None:
- entity_key, default_towrap = entity, entity.selectable
+ if query_entity_key is None:
+ query_entity_key, default_towrap = (
+ query_entity,
+ query_entity.selectable,
+ )
- towrap = context.eager_joins.setdefault(entity_key, default_towrap)
+ towrap = context.eager_joins.setdefault(
+ query_entity_key, default_towrap
+ )
if adapter:
if getattr(adapter, "aliased_class", None):
not chained_from_outerjoin
or not innerjoin
or innerjoin == "unnested"
- or entity.entity_zero.represents_outer_join
+ or query_entity.entity_zero.represents_outer_join
)
if attach_on_outside:
clauses.aliased_class,
onclause,
isouter=not innerjoin
- or entity.entity_zero.represents_outer_join
+ or query_entity.entity_zero.represents_outer_join
or (chained_from_outerjoin and isinstance(towrap, sql.Join)),
_left_memo=self.parent,
_right_memo=self.mapper,
path, towrap, clauses, onclause
)
- context.eager_joins[entity_key] = eagerjoin
+ context.eager_joins[query_entity_key] = eagerjoin
# send a hint to the Query as to where it may "splice" this join
- eagerjoin.stop_on = entity.selectable
+ eagerjoin.stop_on = query_entity.selectable
if not parentmapper:
# for parentclause that is the non-eager end of the join,
for col in sql_util._find_columns(
self.parent_property.primaryjoin
):
- if localparent.mapped_table.c.contains_column(col):
+ if localparent.persist_selectable.c.contains_column(col):
if adapter:
col = adapter.columns[col]
context.primary_columns.append(col)
self.mapper,
context,
result,
- our_path[self.mapper],
+ our_path[self.entity],
eager_adapter,
)
if with_poly_info is not None:
effective_entity = with_poly_info.entity
else:
- effective_entity = self.mapper
+ effective_entity = self.entity
if not path_w_prop.contains(context.attributes, "loader"):
if self.join_depth:
ent = inspect(existing_of_type)
else:
ent = path.entity
+
try:
# use getattr on the class to work around
# synonyms, hybrids, etc.
class AliasedClass(object):
r"""Represents an "aliased" form of a mapped class for usage with Query.
- The ORM equivalent of a :func:`sqlalchemy.sql.expression.alias`
+ The ORM equivalent of a :func:`~sqlalchemy.sql.expression.alias`
construct, this object mimics the mapped class using a
- __getattr__ scheme and maintains a reference to a
+ ``__getattr__`` scheme and maintains a reference to a
real :class:`~sqlalchemy.sql.expression.Alias` object.
- Usage is via the :func:`.orm.aliased` function, or alternatively
- via the :func:`.orm.with_polymorphic` function.
-
- Usage example::
+ A primary purpose of :class:`.AliasedClass` is to serve as an alternate
+ within a SQL statement generated by the ORM, such that an existing
+ mapped entity can be used in multiple contexts. A simple example::
# find all pairs of users with the same name
user_alias = aliased(User)
session.query(User, user_alias).\
join((user_alias, User.id > user_alias.id)).\
- filter(User.name==user_alias.name)
+ filter(User.name == user_alias.name)
+
+ :class:`.AliasedClass` is also capable of mapping an existing mapped
+ class to an entirely new selectable, provided this selectable is column-
+ compatible with the existing mapped selectable, and it can also be
+ configured in a mapping as the target of a :func:`.relationship`.
+ See the links below for examples.
+
+ The :class:`.AliasedClass` object is constructed typically using the
+ :func:`.orm.aliased` function. It also is produced with additional
+ configuration when using the :func:`.orm.with_polymorphic` function.
The resulting object is an instance of :class:`.AliasedClass`.
This object implements an attribute scheme which produces the
The resulting inspection object is an instance of :class:`.AliasedInsp`.
- See :func:`.aliased` and :func:`.with_polymorphic` for construction
- argument descriptions.
+
+ .. seealso::
+
+ :func:`.aliased`
+
+ :func:`.with_polymorphic`
+
+ :ref:`relationship_aliased_class`
+
+ :ref:`relationship_to_window_function`
+
"""
):
self.entity = entity
self.mapper = mapper
- self.selectable = selectable
+ self.selectable = (
+ self.persist_selectable
+ ) = self.local_table = selectable
self.name = name
self.with_polymorphic_mappers = with_polymorphic_mappers
self.polymorphic_on = polymorphic_on
else:
assert False, "mapper %s doesn't correspond to %s" % (mapper, self)
+ @util.memoized_property
+ def _get_clause(self):
+ onclause, replacemap = self.mapper._get_clause
+ return (
+ self._adapter.traverse(onclause),
+ {
+ self._adapter.traverse(col): param
+ for col, param in replacemap.items()
+ },
+ )
+
@util.memoized_property
def _memoized_values(self):
return {}
dest_selectable=adapt_to,
source_polymorphic=True,
dest_polymorphic=True,
- of_type=right_info.mapper,
+ of_type_mapper=right_info.mapper,
)
if sj is not None:
)
m2 = mapper(B, table2)
- m3 = mapper(A, table1, non_primary=True)
-
@profile_memory()
def go():
sess = create_session()
go()
metadata.drop_all()
- del m1, m2, m3
+ del m1, m2
assert_no_mappers()
def test_sessionmaker(self):
)
m2 = mapper(B, table2, _compiled_cache_size=50)
- m3 = mapper(A, table1, non_primary=True)
-
@profile_memory()
def go():
engine = engines.testing_engine(
go()
metadata.drop_all()
- del m1, m2, m3
+ del m1, m2
assert_no_mappers()
@testing.emits_warning("Compiled statement cache for.*")
)
mapper(B, table2)
- mapper(A, table1, non_primary=True)
-
sess = create_session()
a1 = A(col2="a1")
a2 = A(col2="a2")
--- /dev/null
+from sqlalchemy import testing
+from sqlalchemy.orm import mapper
+from .test_mutable import Foo
+from .test_mutable import (
+ MutableAssociationScalarPickleTest as _MutableAssociationScalarPickleTest,
+)
+from .test_mutable import (
+ MutableWithScalarJSONTest as _MutableWithScalarJSONTest,
+)
+
+
+class MutableIncludeNonPrimaryTest(_MutableWithScalarJSONTest):
+ @classmethod
+ def setup_mappers(cls):
+ foo = cls.tables.foo
+
+ mapper(Foo, foo)
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ mapper(
+ Foo, foo, non_primary=True, properties={"foo_bar": foo.c.data}
+ )
+
+
+class MutableAssocIncludeNonPrimaryTest(_MutableAssociationScalarPickleTest):
+ @classmethod
+ def setup_mappers(cls):
+ foo = cls.tables.foo
+
+ mapper(Foo, foo)
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ mapper(
+ Foo, foo, non_primary=True, properties={"foo_bar": foo.c.data}
+ )
self._test_non_mutable()
-class MutableIncludeNonPrimaryTest(MutableWithScalarJSONTest):
- @classmethod
- def setup_mappers(cls):
- foo = cls.tables.foo
-
- mapper(Foo, foo)
- mapper(Foo, foo, non_primary=True, properties={"foo_bar": foo.c.data})
-
-
class MutableColumnCopyJSONTest(_MutableDictTestBase, fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
)
-class MutableAssocIncludeNonPrimaryTest(MutableAssociationScalarPickleTest):
- @classmethod
- def setup_mappers(cls):
- foo = cls.tables.foo
-
- mapper(Foo, foo)
- mapper(Foo, foo, non_primary=True, properties={"foo_bar": foo.c.data})
-
-
class MutableAssociationScalarJSONTest(
_MutableDictTestBase, fixtures.MappedTest
):
--- /dev/null
+from sqlalchemy import and_
+from sqlalchemy import Column
+from sqlalchemy import ForeignKey
+from sqlalchemy import func
+from sqlalchemy import Integer
+from sqlalchemy import join
+from sqlalchemy import select
+from sqlalchemy import testing
+from sqlalchemy.orm import aliased
+from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import noload
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm import selectinload
+from sqlalchemy.orm import Session
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixtures
+from sqlalchemy.testing.assertsql import CompiledSQL
+from sqlalchemy.testing.fixtures import ComparableEntity
+
+
+class PartitionByFixture(fixtures.DeclarativeMappedTest):
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class A(Base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+
+ class B(Base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+ a_id = Column(ForeignKey("a.id"))
+ cs = relationship("C")
+
+ class C(Base):
+ __tablename__ = "c"
+ id = Column(Integer, primary_key=True)
+ b_id = Column(ForeignKey("b.id"))
+
+ partition = select(
+ [
+ B,
+ func.row_number()
+ .over(order_by=B.id, partition_by=B.a_id)
+ .label("index"),
+ ]
+ ).alias()
+
+ partitioned_b = aliased(B, alias=partition)
+
+ A.partitioned_bs = relationship(
+ partitioned_b,
+ primaryjoin=and_(
+ partitioned_b.a_id == A.id, partition.c.index < 10
+ ),
+ )
+
+ @classmethod
+ def insert_data(cls):
+ A, B, C = cls.classes("A", "B", "C")
+
+ s = Session(testing.db)
+ s.add_all([A(id=i) for i in range(1, 4)])
+ s.flush()
+ s.add_all(
+ [
+ B(a_id=i, cs=[C(), C()])
+ for i in range(1, 4)
+ for j in range(1, 21)
+ ]
+ )
+ s.commit()
+
+
+class AliasedClassRelationshipTest(
+ PartitionByFixture, testing.AssertsCompiledSQL
+):
+ # TODO: maybe make this more backend agnostic
+ __requires__ = ("window_functions",)
+ __dialect__ = "default"
+
+ def test_lazyload(self):
+ A, B, C = self.classes("A", "B", "C")
+
+ s = Session(testing.db)
+
+ def go():
+ for a1 in s.query(A): # 1 query
+ eq_(len(a1.partitioned_bs), 9) # 3 queries
+ for b in a1.partitioned_bs:
+ eq_(len(b.cs), 2) # 9 * 3 = 27 queries
+
+ self.assert_sql_count(testing.db, go, 31)
+
+ def test_join_one(self):
+ A, B, C = self.classes("A", "B", "C")
+
+ s = Session(testing.db)
+
+ q = s.query(A).join(A.partitioned_bs)
+ self.assert_compile(
+ q,
+ "SELECT a.id AS a_id FROM a JOIN "
+ "(SELECT b.id AS id, b.a_id AS a_id, row_number() "
+ "OVER (PARTITION BY b.a_id ORDER BY b.id) "
+ "AS index FROM b) AS anon_1 "
+ "ON anon_1.a_id = a.id AND anon_1.index < :index_1",
+ )
+
+ def test_join_two(self):
+ A, B, C = self.classes("A", "B", "C")
+
+ s = Session(testing.db)
+
+ q = s.query(A, A.partitioned_bs.entity).join(A.partitioned_bs)
+ self.assert_compile(
+ q,
+ "SELECT a.id AS a_id, anon_1.id AS anon_1_id, "
+ "anon_1.a_id AS anon_1_a_id "
+ "FROM a JOIN "
+ "(SELECT b.id AS id, b.a_id AS a_id, row_number() "
+ "OVER (PARTITION BY b.a_id ORDER BY b.id) "
+ "AS index FROM b) AS anon_1 "
+ "ON anon_1.a_id = a.id AND anon_1.index < :index_1",
+ )
+
+ def test_selectinload_w_noload_after(self):
+ A, B, C = self.classes("A", "B", "C")
+
+ s = Session(testing.db)
+
+ def go():
+ for a1 in s.query(A).options(
+ noload("*"), selectinload(A.partitioned_bs)
+ ):
+ for b in a1.partitioned_bs:
+ eq_(b.cs, [])
+
+ self.assert_sql_count(testing.db, go, 2)
+
+ def test_selectinload_w_joinedload_after(self):
+ A, B, C = self.classes("A", "B", "C")
+
+ s = Session(testing.db)
+
+ def go():
+ for a1 in s.query(A).options(
+ selectinload(A.partitioned_bs).joinedload("cs")
+ ):
+ for b in a1.partitioned_bs:
+ eq_(len(b.cs), 2)
+
+ self.assert_sql_count(testing.db, go, 2)
+
+
+class AltSelectableTest(
+ fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL
+):
+ __dialect__ = "default"
+
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class A(ComparableEntity, Base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ b_id = Column(ForeignKey("b.id"))
+
+ class B(ComparableEntity, Base):
+ __tablename__ = "b"
+
+ id = Column(Integer, primary_key=True)
+
+ class C(ComparableEntity, Base):
+ __tablename__ = "c"
+
+ id = Column(Integer, primary_key=True)
+ a_id = Column(ForeignKey("a.id"))
+
+ class D(ComparableEntity, Base):
+ __tablename__ = "d"
+
+ id = Column(Integer, primary_key=True)
+ c_id = Column(ForeignKey("c.id"))
+ b_id = Column(ForeignKey("b.id"))
+
+ # 1. set up the join() as a variable, so we can refer
+ # to it in the mapping multiple times.
+ j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
+
+ # 2. Create an AliasedClass to B
+ B_viacd = aliased(B, j, flat=True)
+
+ A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
+
+ @classmethod
+ def insert_data(cls):
+ A, B, C, D = cls.classes("A", "B", "C", "D")
+ sess = Session()
+
+ for obj in [
+ B(id=1),
+ A(id=1, b_id=1),
+ C(id=1, a_id=1),
+ D(id=1, c_id=1, b_id=1),
+ ]:
+ sess.add(obj)
+ sess.flush()
+ sess.commit()
+
+ def test_lazyload(self):
+ A, B = self.classes("A", "B")
+
+ sess = Session()
+ a1 = sess.query(A).first()
+
+ with self.sql_execution_asserter() as asserter:
+ # note this is many-to-one. use_get is unconditionally turned
+ # off for relationship to aliased class for now.
+ eq_(a1.b, B(id=1))
+
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT b.id AS b_id FROM b JOIN d ON d.b_id = b.id "
+ "JOIN c ON c.id = d.c_id WHERE :param_1 = b.id",
+ [{"param_1": 1}],
+ )
+ )
+
+ def test_joinedload(self):
+ A, B = self.classes("A", "B")
+
+ sess = Session()
+
+ with self.sql_execution_asserter() as asserter:
+ # note this is many-to-one. use_get is unconditionally turned
+ # off for relationship to aliased class for now.
+ a1 = sess.query(A).options(joinedload(A.b)).first()
+ eq_(a1.b, B(id=1))
+
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT a.id AS a_id, a.b_id AS a_b_id, b_1.id AS b_1_id "
+ "FROM a LEFT OUTER JOIN (b AS b_1 "
+ "JOIN d AS d_1 ON d_1.b_id = b_1.id "
+ "JOIN c AS c_1 ON c_1.id = d_1.c_id) ON a.b_id = b_1.id "
+ "LIMIT :param_1",
+ [{"param_1": 1}],
+ )
+ )
+
+ def test_selectinload(self):
+ A, B = self.classes("A", "B")
+
+ sess = Session()
+
+ with self.sql_execution_asserter() as asserter:
+ # note this is many-to-one. use_get is unconditionally turned
+ # off for relationship to aliased class for now.
+ a1 = sess.query(A).options(selectinload(A.b)).first()
+ eq_(a1.b, B(id=1))
+
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT a.id AS a_id, a.b_id AS a_b_id "
+ "FROM a LIMIT :param_1",
+ [{"param_1": 1}],
+ ),
+ CompiledSQL(
+ "SELECT a_1.id AS a_1_id, b.id AS b_id FROM a AS a_1 "
+ "JOIN (b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) "
+ "ON a_1.b_id = b.id WHERE a_1.id "
+ "IN ([EXPANDING_primary_keys]) ORDER BY a_1.id",
+ [{"primary_keys": [1]}],
+ ),
+ )
+
+ def test_join(self):
+ A, B = self.classes("A", "B")
+
+ sess = Session()
+
+ self.assert_compile(
+ sess.query(A).join(A.b),
+ "SELECT a.id AS a_id, a.b_id AS a_b_id "
+ "FROM a JOIN (b JOIN d ON d.b_id = b.id "
+ "JOIN c ON c.id = d.c_id) ON a.b_id = b.id",
+ )
l2 = Collection()
f1.attr = l2
eq_(canary, [adapter_1, f1.attr._sa_adapter, None])
+
+
+class NonPrimaryRelationshipLoaderTest(_fixtures.FixtureTest):
+ run_inserts = "once"
+ run_deletes = None
+
+ def test_selectload(self):
+ """tests lazy loading with two relationships simultaneously,
+ from the same table, using aliases. """
+
+ users, orders, User, Address, Order, addresses = (
+ self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses,
+ )
+
+ openorders = sa.alias(orders, "openorders")
+ closedorders = sa.alias(orders, "closedorders")
+
+ mapper(Address, addresses)
+
+ mapper(Order, orders)
+
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ open_mapper = mapper(Order, openorders, non_primary=True)
+ closed_mapper = mapper(Order, closedorders, non_primary=True)
+ mapper(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(Address, lazy=True),
+ open_orders=relationship(
+ open_mapper,
+ primaryjoin=sa.and_(
+ openorders.c.isopen == 1,
+ users.c.id == openorders.c.user_id,
+ ),
+ lazy="select",
+ ),
+ closed_orders=relationship(
+ closed_mapper,
+ primaryjoin=sa.and_(
+ closedorders.c.isopen == 0,
+ users.c.id == closedorders.c.user_id,
+ ),
+ lazy="select",
+ ),
+ ),
+ )
+
+ self._run_double_test(10)
+
+ def test_joinedload(self):
+ """Eager loading with two relationships simultaneously,
+ from the same table, using aliases."""
+
+ users, orders, User, Address, Order, addresses = (
+ self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses,
+ )
+
+ openorders = sa.alias(orders, "openorders")
+ closedorders = sa.alias(orders, "closedorders")
+
+ mapper(Address, addresses)
+ mapper(Order, orders)
+
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ open_mapper = mapper(Order, openorders, non_primary=True)
+ closed_mapper = mapper(Order, closedorders, non_primary=True)
+
+ mapper(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(
+ Address, lazy="joined", order_by=addresses.c.id
+ ),
+ open_orders=relationship(
+ open_mapper,
+ primaryjoin=sa.and_(
+ openorders.c.isopen == 1,
+ users.c.id == openorders.c.user_id,
+ ),
+ lazy="joined",
+ order_by=openorders.c.id,
+ ),
+ closed_orders=relationship(
+ closed_mapper,
+ primaryjoin=sa.and_(
+ closedorders.c.isopen == 0,
+ users.c.id == closedorders.c.user_id,
+ ),
+ lazy="joined",
+ order_by=closedorders.c.id,
+ ),
+ ),
+ )
+ self._run_double_test(1)
+
+ def test_selectin(self):
+
+ users, orders, User, Address, Order, addresses = (
+ self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses,
+ )
+
+ openorders = sa.alias(orders, "openorders")
+ closedorders = sa.alias(orders, "closedorders")
+
+ mapper(Address, addresses)
+ mapper(Order, orders)
+
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ open_mapper = mapper(Order, openorders, non_primary=True)
+ closed_mapper = mapper(Order, closedorders, non_primary=True)
+
+ mapper(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(
+ Address, lazy="selectin", order_by=addresses.c.id
+ ),
+ open_orders=relationship(
+ open_mapper,
+ primaryjoin=sa.and_(
+ openorders.c.isopen == 1,
+ users.c.id == openorders.c.user_id,
+ ),
+ lazy="selectin",
+ order_by=openorders.c.id,
+ ),
+ closed_orders=relationship(
+ closed_mapper,
+ primaryjoin=sa.and_(
+ closedorders.c.isopen == 0,
+ users.c.id == closedorders.c.user_id,
+ ),
+ lazy="selectin",
+ order_by=closedorders.c.id,
+ ),
+ ),
+ )
+
+ self._run_double_test(4)
+
+ def test_subqueryload(self):
+
+ users, orders, User, Address, Order, addresses = (
+ self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses,
+ )
+
+ openorders = sa.alias(orders, "openorders")
+ closedorders = sa.alias(orders, "closedorders")
+
+ mapper(Address, addresses)
+ mapper(Order, orders)
+
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ open_mapper = mapper(Order, openorders, non_primary=True)
+ closed_mapper = mapper(Order, closedorders, non_primary=True)
+
+ mapper(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(
+ Address, lazy="subquery", order_by=addresses.c.id
+ ),
+ open_orders=relationship(
+ open_mapper,
+ primaryjoin=sa.and_(
+ openorders.c.isopen == 1,
+ users.c.id == openorders.c.user_id,
+ ),
+ lazy="subquery",
+ order_by=openorders.c.id,
+ ),
+ closed_orders=relationship(
+ closed_mapper,
+ primaryjoin=sa.and_(
+ closedorders.c.isopen == 0,
+ users.c.id == closedorders.c.user_id,
+ ),
+ lazy="subquery",
+ order_by=closedorders.c.id,
+ ),
+ ),
+ )
+
+ self._run_double_test(4)
+
+ def _run_double_test(self, count):
+ User, Address, Order, Item = self.classes(
+ "User", "Address", "Order", "Item"
+ )
+ q = create_session().query(User).order_by(User.id)
+
+ def go():
+ eq_(
+ [
+ User(
+ id=7,
+ addresses=[Address(id=1)],
+ open_orders=[Order(id=3)],
+ closed_orders=[Order(id=1), Order(id=5)],
+ ),
+ User(
+ id=8,
+ addresses=[
+ Address(id=2),
+ Address(id=3),
+ Address(id=4),
+ ],
+ open_orders=[],
+ closed_orders=[],
+ ),
+ User(
+ id=9,
+ addresses=[Address(id=5)],
+ open_orders=[Order(id=4)],
+ closed_orders=[Order(id=2)],
+ ),
+ User(id=10),
+ ],
+ q.all(),
+ )
+
+ self.assert_sql_count(testing.db, go, count)
+
+ sess = create_session()
+ user = sess.query(User).get(7)
+
+ closed_mapper = User.closed_orders.entity
+ open_mapper = User.open_orders.entity
+ eq_(
+ [Order(id=1), Order(id=5)],
+ create_session()
+ .query(closed_mapper)
+ .with_parent(user, property="closed_orders")
+ .all(),
+ )
+ eq_(
+ [Order(id=3)],
+ create_session()
+ .query(open_mapper)
+ .with_parent(user, property="open_orders")
+ .all(),
+ )
+
+
+class NonPrimaryMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
+ __dialect__ = "default"
+
+ def test_non_primary_identity_class(self):
+ User = self.classes.User
+ users, addresses = self.tables.users, self.tables.addresses
+
+ class AddressUser(User):
+ pass
+
+ m1 = mapper(User, users, polymorphic_identity="user")
+ m2 = mapper(
+ AddressUser,
+ addresses,
+ inherits=User,
+ polymorphic_identity="address",
+ properties={"address_id": addresses.c.id},
+ )
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ m3 = mapper(AddressUser, addresses, non_primary=True)
+ assert m3._identity_class is m2._identity_class
+ eq_(
+ m2.identity_key_from_instance(AddressUser()),
+ m3.identity_key_from_instance(AddressUser()),
+ )
+
+ def test_illegal_non_primary(self):
+ users, Address, addresses, User = (
+ self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User,
+ )
+
+ mapper(User, users)
+ mapper(Address, addresses)
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ mapper(
+ User,
+ users,
+ non_primary=True,
+ properties={"addresses": relationship(Address)},
+ )
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ "Attempting to assign a new relationship 'addresses' "
+ "to a non-primary mapper on class 'User'",
+ configure_mappers,
+ )
+
+ def test_illegal_non_primary_2(self):
+ User, users = self.classes.User, self.tables.users
+
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ assert_raises_message(
+ sa.exc.InvalidRequestError,
+ "Configure a primary mapper first",
+ mapper,
+ User,
+ users,
+ non_primary=True,
+ )
+
+ def test_illegal_non_primary_3(self):
+ users, addresses = self.tables.users, self.tables.addresses
+
+ class Base(object):
+ pass
+
+ class Sub(Base):
+ pass
+
+ mapper(Base, users)
+ with testing.expect_deprecated(
+ "The mapper.non_primary parameter is deprecated"
+ ):
+ assert_raises_message(
+ sa.exc.InvalidRequestError,
+ "Configure a primary mapper first",
+ mapper,
+ Sub,
+ addresses,
+ non_primary=True,
+ )
sess.query(User).order_by(User.id).all(),
)
- def test_double(self):
+ def test_double_w_ac(self):
"""Eager loading with two relationships simultaneously,
from the same table, using aliases."""
- users, orders, User, Address, Order, addresses = (
+ (
+ users,
+ orders,
+ User,
+ Address,
+ Order,
+ addresses,
+ Item,
+ items,
+ order_items,
+ ) = (
self.tables.users,
self.tables.orders,
self.classes.User,
self.classes.Address,
self.classes.Order,
self.tables.addresses,
+ self.classes.Item,
+ self.tables.items,
+ self.tables.order_items,
)
- openorders = sa.alias(orders, "openorders")
- closedorders = sa.alias(orders, "closedorders")
-
mapper(Address, addresses)
- mapper(Order, orders)
+ mapper(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="joined",
+ order_by=items.c.id,
+ )
+ },
+ )
+ mapper(Item, items)
- open_mapper = mapper(Order, openorders, non_primary=True)
- closed_mapper = mapper(Order, closedorders, non_primary=True)
+ open_mapper = aliased(Order, orders)
+ closed_mapper = aliased(Order, orders)
mapper(
User,
open_orders=relationship(
open_mapper,
primaryjoin=sa.and_(
- openorders.c.isopen == 1,
- users.c.id == openorders.c.user_id,
+ open_mapper.isopen == 1,
+ users.c.id == open_mapper.user_id,
),
lazy="joined",
- order_by=openorders.c.id,
+ order_by=open_mapper.id,
),
closed_orders=relationship(
closed_mapper,
primaryjoin=sa.and_(
- closedorders.c.isopen == 0,
- users.c.id == closedorders.c.user_id,
+ closed_mapper.isopen == 0,
+ users.c.id == closed_mapper.user_id,
),
lazy="joined",
- order_by=closedorders.c.id,
+ order_by=closed_mapper.id,
),
),
)
- q = create_session().query(User).order_by(User.id)
+ self._run_double_test()
- def go():
- eq_(
- [
- User(
- id=7,
- addresses=[Address(id=1)],
- open_orders=[Order(id=3)],
- closed_orders=[Order(id=1), Order(id=5)],
- ),
- User(
- id=8,
- addresses=[
- Address(id=2),
- Address(id=3),
- Address(id=4),
- ],
- open_orders=[],
- closed_orders=[],
- ),
- User(
- id=9,
- addresses=[Address(id=5)],
- open_orders=[Order(id=4)],
- closed_orders=[Order(id=2)],
- ),
- User(id=10),
- ],
- q.all(),
- )
+ def test_double_w_ac_against_subquery(self):
+ """Eager loading with two relationships simultaneously,
+ from the same table, using aliases."""
- self.assert_sql_count(testing.db, go, 1)
+ (
+ users,
+ orders,
+ User,
+ Address,
+ Order,
+ addresses,
+ Item,
+ items,
+ order_items,
+ ) = (
+ self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses,
+ self.classes.Item,
+ self.tables.items,
+ self.tables.order_items,
+ )
+
+ mapper(Address, addresses)
+ mapper(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="joined",
+ order_by=items.c.id,
+ )
+ },
+ )
+ mapper(Item, items)
+
+ open_mapper = aliased(
+ Order, select([orders]).where(orders.c.isopen == 1).alias()
+ )
+ closed_mapper = aliased(
+ Order, select([orders]).where(orders.c.isopen == 0).alias()
+ )
+
+ mapper(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(
+ Address, lazy="joined", order_by=addresses.c.id
+ ),
+ open_orders=relationship(
+ open_mapper, lazy="joined", order_by=open_mapper.id
+ ),
+ closed_orders=relationship(
+ closed_mapper, lazy="joined", order_by=closed_mapper.id
+ ),
+ ),
+ )
+
+ self._run_double_test()
def test_double_same_mappers(self):
"""Eager loading with two relationships simultaneously,
),
),
)
+ self._run_double_test()
+
+ def _run_double_test(self, no_items=False):
+ User, Address, Order, Item = self.classes(
+ "User", "Address", "Order", "Item"
+ )
q = create_session().query(User).order_by(User.id)
+ def items(*ids):
+ if no_items:
+ return {}
+ else:
+ return {"items": [Item(id=id_) for id_ in ids]}
+
def go():
eq_(
[
User(
id=7,
addresses=[Address(id=1)],
- open_orders=[
- Order(
- id=3,
- items=[Item(id=3), Item(id=4), Item(id=5)],
- )
- ],
+ open_orders=[Order(id=3, **items(3, 4, 5))],
closed_orders=[
- Order(
- id=1,
- items=[Item(id=1), Item(id=2), Item(id=3)],
- ),
- Order(id=5, items=[Item(id=5)]),
+ Order(id=1, **items(1, 2, 3)),
+ Order(id=5, **items(5)),
],
),
User(
User(
id=9,
addresses=[Address(id=5)],
- open_orders=[
- Order(id=4, items=[Item(id=1), Item(id=5)])
- ],
- closed_orders=[
- Order(
- id=2,
- items=[Item(id=1), Item(id=2), Item(id=3)],
- )
- ],
+ open_orders=[Order(id=4, **items(1, 5))],
+ closed_orders=[Order(id=2, **items(1, 2, 3))],
),
User(id=10),
],
Order, backref="user", lazy="joined", order_by=orders.c.id
),
"max_order": relationship(
- mapper(Order, max_orders, non_primary=True),
- lazy="joined",
- uselist=False,
+ aliased(Order, max_orders), lazy="joined", uselist=False
),
},
)
)
self._assert_result(q)
- def test_splice_onto_np_mapper(self):
+ def test_splice_onto_ac(self):
A = self.classes.A
B = self.classes.B
C1 = self.classes.C1
weird_selectable = b_table.outerjoin(c1_table)
- b_np = mapper(
- B,
- weird_selectable,
- non_primary=True,
- properties=odict(
- [
- # note we need to make this fixed with lazy=False until
- # [ticket:3348] is resolved
- ("c1s", relationship(C1, lazy=False, innerjoin=True)),
- ("c_id", c1_table.c.id),
- ("b_value", b_table.c.value),
- ]
- ),
- )
+ b_np = aliased(B, weird_selectable, flat=True)
a_mapper = inspect(A)
a_mapper.add_property("bs_np", relationship(b_np))
q = s.query(A).options(joinedload("bs_np", innerjoin=False))
self.assert_compile(
q,
- "SELECT a.id AS a_id, c1_1.id AS c1_1_id, c1_1.b_id AS c1_1_b_id, "
- "c1_1.value AS c1_1_value, c1_2.id AS c1_2_id, "
- "b_1.value AS b_1_value, b_1.id AS b_1_id, "
- "b_1.a_id AS b_1_a_id, c1_2.b_id AS c1_2_b_id, "
- "c1_2.value AS c1_2_value "
- "FROM a LEFT OUTER JOIN "
- "(b AS b_1 LEFT OUTER JOIN c1 AS c1_2 ON b_1.id = c1_2.b_id "
- "JOIN c1 AS c1_1 ON b_1.id = c1_1.b_id) ON a.id = b_1.a_id",
+ "SELECT a.id AS a_id, b_1.id AS b_1_id, b_1.a_id AS b_1_a_id, "
+ "b_1.value AS b_1_value FROM a LEFT OUTER JOIN "
+ "(b AS b_1 LEFT OUTER JOIN c1 AS c1_1 ON b_1.id = c1_1.b_id) "
+ "ON a.id = b_1.a_id",
)
]:
q = s.query(crit)
mzero = q._entity_zero()
- is_(mzero.mapped_table, q._query_entity_zero().selectable)
+ is_(mzero.persist_selectable, q._query_entity_zero().selectable)
q = q.join(j)
self.assert_compile(q, exp)
insp = inspect(User)
is_(insp.local_table, user_table)
- def test_mapped_table(self):
+ def test_persist_selectable(self):
User = self.classes.User
user_table = self.tables.users
insp = inspect(User)
- is_(insp.mapped_table, user_table)
+ is_(insp.persist_selectable, user_table)
def test_mapper_selectable(self):
User = self.classes.User
mapper(Child, child)
derived = select([child]).alias()
- npc = mapper(Child, derived, non_primary=True)
+ npc = aliased(Child, derived)
cls.npc = npc
cls.derived = derived
mp.add_property("npc", relationship(npc))
from sqlalchemy import ForeignKeyConstraint
from sqlalchemy import Integer
from sqlalchemy import orm
+from sqlalchemy import select
from sqlalchemy import SmallInteger
from sqlalchemy import String
from sqlalchemy import testing
+from sqlalchemy.orm import aliased
from sqlalchemy.orm import attributes
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import create_session
list(q),
)
- def test_double(self):
- """tests lazy loading with two relationships simultaneously,
- from the same table, using aliases. """
+ def test_double_w_ac_against_subquery(self):
- users, orders, User, Address, Order, addresses = (
+ (
+ users,
+ orders,
+ User,
+ Address,
+ Order,
+ addresses,
+ Item,
+ items,
+ order_items,
+ ) = (
self.tables.users,
self.tables.orders,
self.classes.User,
self.classes.Address,
self.classes.Order,
self.tables.addresses,
+ self.classes.Item,
+ self.tables.items,
+ self.tables.order_items,
+ )
+
+ mapper(Address, addresses)
+
+ mapper(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="select",
+ order_by=items.c.id,
+ )
+ },
+ )
+ mapper(Item, items)
+
+ open_mapper = aliased(
+ Order, select([orders]).where(orders.c.isopen == 1).alias()
+ )
+ closed_mapper = aliased(
+ Order, select([orders]).where(orders.c.isopen == 0).alias()
+ )
+
+ mapper(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(Address, lazy=True),
+ open_orders=relationship(open_mapper, lazy="select"),
+ closed_orders=relationship(closed_mapper, lazy="select"),
+ ),
)
- openorders = sa.alias(orders, "openorders")
- closedorders = sa.alias(orders, "closedorders")
+ self._run_double_test()
+
+ def test_double_w_ac(self):
+
+ (
+ users,
+ orders,
+ User,
+ Address,
+ Order,
+ addresses,
+ Item,
+ items,
+ order_items,
+ ) = (
+ self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses,
+ self.classes.Item,
+ self.tables.items,
+ self.tables.order_items,
+ )
mapper(Address, addresses)
- mapper(Order, orders)
+ mapper(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="select",
+ order_by=items.c.id,
+ )
+ },
+ )
+ mapper(Item, items)
- open_mapper = mapper(Order, openorders, non_primary=True)
- closed_mapper = mapper(Order, closedorders, non_primary=True)
+ open_mapper = aliased(Order, orders)
+ closed_mapper = aliased(Order, orders)
mapper(
User,
users,
open_orders=relationship(
open_mapper,
primaryjoin=sa.and_(
- openorders.c.isopen == 1,
- users.c.id == openorders.c.user_id,
+ open_mapper.isopen == 1,
+ users.c.id == open_mapper.user_id,
),
lazy="select",
),
closed_orders=relationship(
closed_mapper,
primaryjoin=sa.and_(
- closedorders.c.isopen == 0,
- users.c.id == closedorders.c.user_id,
+ closed_mapper.isopen == 0,
+ users.c.id == closed_mapper.user_id,
),
lazy="select",
),
),
)
- q = create_session().query(User)
- assert [
- User(
- id=7,
- addresses=[Address(id=1)],
- open_orders=[Order(id=3)],
- closed_orders=[Order(id=1), Order(id=5)],
- ),
- User(
- id=8,
- addresses=[Address(id=2), Address(id=3), Address(id=4)],
- open_orders=[],
- closed_orders=[],
- ),
- User(
- id=9,
- addresses=[Address(id=5)],
- open_orders=[Order(id=4)],
- closed_orders=[Order(id=2)],
- ),
- User(id=10),
- ] == q.all()
+ self._run_double_test()
+
+ def _run_double_test(self, no_items=False):
+ User, Address, Order, Item = self.classes(
+ "User", "Address", "Order", "Item"
+ )
+ q = create_session().query(User).order_by(User.id)
+
+ def items(*ids):
+ if no_items:
+ return {}
+ else:
+ return {"items": [Item(id=id_) for id_ in ids]}
+
+ def go():
+ eq_(
+ [
+ User(
+ id=7,
+ addresses=[Address(id=1)],
+ open_orders=[Order(id=3, **items(3, 4, 5))],
+ closed_orders=[
+ Order(id=1, **items(1, 2, 3)),
+ Order(id=5, **items(5)),
+ ],
+ ),
+ User(
+ id=8,
+ addresses=[
+ Address(id=2),
+ Address(id=3),
+ Address(id=4),
+ ],
+ open_orders=[],
+ closed_orders=[],
+ ),
+ User(
+ id=9,
+ addresses=[Address(id=5)],
+ open_orders=[Order(id=4, **items(1, 5))],
+ closed_orders=[Order(id=2, **items(1, 2, 3))],
+ ),
+ User(id=10),
+ ],
+ q.all(),
+ )
+
+ if no_items:
+ self.assert_sql_count(testing.db, go, 10)
+ else:
+ self.assert_sql_count(testing.db, go, 15)
sess = create_session()
user = sess.query(User).get(7)
+
+ closed_mapper = User.closed_orders.entity
+ open_mapper = User.open_orders.entity
eq_(
[Order(id=1), Order(id=5)],
create_session()
assert n1.children[0] is n1._children[0] is n2
eq_(str(Node.parent == n2), ":param_1 = nodes.parent_id")
- def test_non_primary_identity_class(self):
- User = self.classes.User
- users, addresses = self.tables.users, self.tables.addresses
-
- class AddressUser(User):
- pass
-
- m1 = mapper(User, users, polymorphic_identity="user")
- m2 = mapper(
- AddressUser,
- addresses,
- inherits=User,
- polymorphic_identity="address",
- properties={"address_id": addresses.c.id},
- )
- m3 = mapper(AddressUser, addresses, non_primary=True)
- assert m3._identity_class is m2._identity_class
- eq_(
- m2.identity_key_from_instance(AddressUser()),
- m3.identity_key_from_instance(AddressUser()),
- )
-
def test_reassign_polymorphic_identity_warns(self):
User = self.classes.User
users = self.tables.users
polymorphic_identity="user",
)
- def test_illegal_non_primary(self):
- users, Address, addresses, User = (
- self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User,
- )
-
- mapper(User, users)
- mapper(Address, addresses)
- mapper(
- User,
- users,
- non_primary=True,
- properties={"addresses": relationship(Address)},
- )
- assert_raises_message(
- sa.exc.ArgumentError,
- "Attempting to assign a new relationship 'addresses' "
- "to a non-primary mapper on class 'User'",
- configure_mappers,
- )
-
- def test_illegal_non_primary_2(self):
- User, users = self.classes.User, self.tables.users
-
- assert_raises_message(
- sa.exc.InvalidRequestError,
- "Configure a primary mapper first",
- mapper,
- User,
- users,
- non_primary=True,
- )
-
- def test_illegal_non_primary_3(self):
- users, addresses = self.tables.users, self.tables.addresses
-
- class Base(object):
- pass
-
- class Sub(Base):
- pass
-
- mapper(Base, users)
- assert_raises_message(
- sa.exc.InvalidRequestError,
- "Configure a primary mapper first",
- mapper,
- Sub,
- addresses,
- non_primary=True,
- )
-
def test_prop_filters(self):
t = Table(
"person",
def test_join_targets_o2m_selfref(self):
joincond = self._join_fixture_o2m_selfref()
- left = select([joincond.parent_selectable]).alias("pj")
+ left = select([joincond.parent_persist_selectable]).alias("pj")
pj, sj, sec, adapter, ds = joincond.join_targets(
- left, joincond.child_selectable, True
+ left, joincond.child_persist_selectable, True
)
self.assert_compile(pj, "pj.id = selfref.sid")
+ self.assert_compile(pj, "pj.id = selfref.sid")
- right = select([joincond.child_selectable]).alias("pj")
+ right = select([joincond.child_persist_selectable]).alias("pj")
pj, sj, sec, adapter, ds = joincond.join_targets(
- joincond.parent_selectable, right, True
+ joincond.parent_persist_selectable, right, True
)
self.assert_compile(pj, "selfref.id = pj.sid")
+ self.assert_compile(pj, "selfref.id = pj.sid")
def test_join_targets_o2m_plain(self):
joincond = self._join_fixture_o2m()
pj, sj, sec, adapter, ds = joincond.join_targets(
- joincond.parent_selectable, joincond.child_selectable, False
+ joincond.parent_persist_selectable,
+ joincond.child_persist_selectable,
+ False,
)
self.assert_compile(pj, "lft.id = rgt.lid")
+ self.assert_compile(pj, "lft.id = rgt.lid")
def test_join_targets_o2m_left_aliased(self):
joincond = self._join_fixture_o2m()
- left = select([joincond.parent_selectable]).alias("pj")
+ left = select([joincond.parent_persist_selectable]).alias("pj")
pj, sj, sec, adapter, ds = joincond.join_targets(
- left, joincond.child_selectable, True
+ left, joincond.child_persist_selectable, True
)
self.assert_compile(pj, "pj.id = rgt.lid")
+ self.assert_compile(pj, "pj.id = rgt.lid")
def test_join_targets_o2m_right_aliased(self):
joincond = self._join_fixture_o2m()
- right = select([joincond.child_selectable]).alias("pj")
+ right = select([joincond.child_persist_selectable]).alias("pj")
pj, sj, sec, adapter, ds = joincond.join_targets(
- joincond.parent_selectable, right, True
+ joincond.parent_persist_selectable, right, True
)
self.assert_compile(pj, "lft.id = pj.lid")
+ self.assert_compile(pj, "lft.id = pj.lid")
def test_join_targets_o2m_composite_selfref(self):
joincond = self._join_fixture_o2m_composite_selfref()
- right = select([joincond.child_selectable]).alias("pj")
+ right = select([joincond.child_persist_selectable]).alias("pj")
pj, sj, sec, adapter, ds = joincond.join_targets(
- joincond.parent_selectable, right, True
+ joincond.parent_persist_selectable, right, True
)
self.assert_compile(
pj,
def test_join_targets_m2o_composite_selfref(self):
joincond = self._join_fixture_m2o_composite_selfref()
- right = select([joincond.child_selectable]).alias("pj")
+ right = select([joincond.child_persist_selectable]).alias("pj")
pj, sj, sec, adapter, ds = joincond.join_targets(
- joincond.parent_selectable, right, True
+ joincond.parent_persist_selectable, right, True
)
self.assert_compile(
pj,
Subset, common = self.classes.Subset, self.tables.common
subset_select = select([common.c.id, common.c.data]).alias()
- subset_mapper = mapper(Subset, subset_select)
+ mapper(Subset, subset_select)
sess = Session(bind=testing.db)
sess.add(Subset(data=1))
sess.flush()
eq_(sess.query(Subset).filter(Subset.data == 1).one(), Subset(data=1))
eq_(sess.query(Subset).filter(Subset.data != 1).first(), None)
- subset_select = sa.orm.class_mapper(Subset).mapped_table
+ subset_select = sa.orm.class_mapper(Subset).persist_selectable
eq_(
sess.query(Subset).filter(subset_select.c.data == 1).one(),
Subset(data=1),
from sqlalchemy import bindparam
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
+from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy.orm import aliased
sess.query(User).order_by(User.id).all(),
)
- def test_double(self):
- """Eager loading with two relationships simultaneously,
- from the same table, using aliases."""
+ def test_double_w_ac_against_subquery(self):
- users, orders, User, Address, Order, addresses = (
+ (
+ users,
+ orders,
+ User,
+ Address,
+ Order,
+ addresses,
+ Item,
+ items,
+ order_items,
+ ) = (
self.tables.users,
self.tables.orders,
self.classes.User,
self.classes.Address,
self.classes.Order,
self.tables.addresses,
+ self.classes.Item,
+ self.tables.items,
+ self.tables.order_items,
+ )
+
+ mapper(Address, addresses)
+ mapper(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="selectin",
+ order_by=items.c.id,
+ )
+ },
+ )
+ mapper(Item, items)
+
+ open_mapper = aliased(
+ Order, select([orders]).where(orders.c.isopen == 1).alias()
)
+ closed_mapper = aliased(
+ Order, select([orders]).where(orders.c.isopen == 0).alias()
+ )
+
+ mapper(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(
+ Address, lazy="selectin", order_by=addresses.c.id
+ ),
+ open_orders=relationship(
+ open_mapper, lazy="selectin", order_by=open_mapper.id
+ ),
+ closed_orders=relationship(
+ closed_mapper, lazy="selectin", order_by=closed_mapper.id
+ ),
+ ),
+ )
+
+ self._run_double_test()
- openorders = sa.alias(orders, "openorders")
- closedorders = sa.alias(orders, "closedorders")
+ def test_double_w_ac(self):
+
+ (
+ users,
+ orders,
+ User,
+ Address,
+ Order,
+ addresses,
+ Item,
+ items,
+ order_items,
+ ) = (
+ self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses,
+ self.classes.Item,
+ self.tables.items,
+ self.tables.order_items,
+ )
mapper(Address, addresses)
- mapper(Order, orders)
+ mapper(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="selectin",
+ order_by=items.c.id,
+ )
+ },
+ )
+ mapper(Item, items)
- open_mapper = mapper(Order, openorders, non_primary=True)
- closed_mapper = mapper(Order, closedorders, non_primary=True)
+ open_mapper = aliased(Order, orders)
+ closed_mapper = aliased(Order, orders)
mapper(
User,
open_orders=relationship(
open_mapper,
primaryjoin=sa.and_(
- openorders.c.isopen == 1,
- users.c.id == openorders.c.user_id,
+ open_mapper.isopen == 1,
+ users.c.id == open_mapper.user_id,
),
lazy="selectin",
- order_by=openorders.c.id,
+ order_by=open_mapper.id,
),
closed_orders=relationship(
closed_mapper,
primaryjoin=sa.and_(
- closedorders.c.isopen == 0,
- users.c.id == closedorders.c.user_id,
+ closed_mapper.isopen == 0,
+ users.c.id == closed_mapper.user_id,
),
lazy="selectin",
- order_by=closedorders.c.id,
+ order_by=closed_mapper.id,
),
),
)
- q = create_session().query(User).order_by(User.id)
-
- def go():
- eq_(
- [
- User(
- id=7,
- addresses=[Address(id=1)],
- open_orders=[Order(id=3)],
- closed_orders=[Order(id=1), Order(id=5)],
- ),
- User(
- id=8,
- addresses=[
- Address(id=2),
- Address(id=3),
- Address(id=4),
- ],
- open_orders=[],
- closed_orders=[],
- ),
- User(
- id=9,
- addresses=[Address(id=5)],
- open_orders=[Order(id=4)],
- closed_orders=[Order(id=2)],
- ),
- User(id=10),
- ],
- q.all(),
- )
-
- self.assert_sql_count(testing.db, go, 4)
+ self._run_double_test()
def test_double_same_mappers(self):
"""Eager loading with two relationships simultaneously,
),
),
)
+
+ self._run_double_test()
+
+ def _run_double_test(self, no_items=False):
+ User, Address, Order, Item = self.classes(
+ "User", "Address", "Order", "Item"
+ )
q = create_session().query(User).order_by(User.id)
+ def items(*ids):
+ if no_items:
+ return {}
+ else:
+ return {"items": [Item(id=id_) for id_ in ids]}
+
def go():
eq_(
[
User(
id=7,
addresses=[Address(id=1)],
- open_orders=[
- Order(
- id=3,
- items=[Item(id=3), Item(id=4), Item(id=5)],
- )
- ],
+ open_orders=[Order(id=3, **items(3, 4, 5))],
closed_orders=[
- Order(
- id=1,
- items=[Item(id=1), Item(id=2), Item(id=3)],
- ),
- Order(id=5, items=[Item(id=5)]),
+ Order(id=1, **items(1, 2, 3)),
+ Order(id=5, **items(5)),
],
),
User(
User(
id=9,
addresses=[Address(id=5)],
- open_orders=[
- Order(id=4, items=[Item(id=1), Item(id=5)])
- ],
- closed_orders=[
- Order(
- id=2,
- items=[Item(id=1), Item(id=2), Item(id=3)],
- )
- ],
+ open_orders=[Order(id=4, **items(1, 5))],
+ closed_orders=[Order(id=2, **items(1, 2, 3))],
),
User(id=10),
],
q.all(),
)
- self.assert_sql_count(testing.db, go, 6)
+ if no_items:
+ self.assert_sql_count(testing.db, go, 4)
+ else:
+ self.assert_sql_count(testing.db, go, 6)
def test_limit(self):
"""Limit operations combined with lazy-load relationships."""
order_by=orders.c.id,
),
"max_order": relationship(
- mapper(Order, max_orders, non_primary=True),
- lazy="selectin",
- uselist=False,
+ aliased(Order, max_orders), lazy="selectin", uselist=False
),
},
)
from sqlalchemy import ForeignKey
from sqlalchemy import inspect
from sqlalchemy import Integer
+from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy.orm import aliased
"1 FROM users",
)
- def test_double(self):
- """Eager loading with two relationships simultaneously,
- from the same table, using aliases."""
+ def test_double_w_ac_against_subquery(self):
- users, orders, User, Address, Order, addresses = (
+ (
+ users,
+ orders,
+ User,
+ Address,
+ Order,
+ addresses,
+ Item,
+ items,
+ order_items,
+ ) = (
self.tables.users,
self.tables.orders,
self.classes.User,
self.classes.Address,
self.classes.Order,
self.tables.addresses,
+ self.classes.Item,
+ self.tables.items,
+ self.tables.order_items,
+ )
+
+ mapper(Address, addresses)
+ mapper(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="subquery",
+ order_by=items.c.id,
+ )
+ },
+ )
+ mapper(Item, items)
+
+ open_mapper = aliased(
+ Order, select([orders]).where(orders.c.isopen == 1).alias()
)
+ closed_mapper = aliased(
+ Order, select([orders]).where(orders.c.isopen == 0).alias()
+ )
+
+ mapper(
+ User,
+ users,
+ properties=dict(
+ addresses=relationship(
+ Address, lazy="subquery", order_by=addresses.c.id
+ ),
+ open_orders=relationship(
+ open_mapper, lazy="subquery", order_by=open_mapper.id
+ ),
+ closed_orders=relationship(
+ closed_mapper, lazy="subquery", order_by=closed_mapper.id
+ ),
+ ),
+ )
+
+ self._run_double_test()
- openorders = sa.alias(orders, "openorders")
- closedorders = sa.alias(orders, "closedorders")
+ def test_double_w_ac(self):
+
+ (
+ users,
+ orders,
+ User,
+ Address,
+ Order,
+ addresses,
+ Item,
+ items,
+ order_items,
+ ) = (
+ self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses,
+ self.classes.Item,
+ self.tables.items,
+ self.tables.order_items,
+ )
mapper(Address, addresses)
- mapper(Order, orders)
+ mapper(
+ Order,
+ orders,
+ properties={
+ "items": relationship(
+ Item,
+ secondary=order_items,
+ lazy="subquery",
+ order_by=items.c.id,
+ )
+ },
+ )
+ mapper(Item, items)
- open_mapper = mapper(Order, openorders, non_primary=True)
- closed_mapper = mapper(Order, closedorders, non_primary=True)
+ open_mapper = aliased(Order, orders)
+ closed_mapper = aliased(Order, orders)
mapper(
User,
open_orders=relationship(
open_mapper,
primaryjoin=sa.and_(
- openorders.c.isopen == 1,
- users.c.id == openorders.c.user_id,
+ open_mapper.isopen == 1,
+ users.c.id == open_mapper.user_id,
),
lazy="subquery",
- order_by=openorders.c.id,
+ order_by=open_mapper.id,
),
closed_orders=relationship(
closed_mapper,
primaryjoin=sa.and_(
- closedorders.c.isopen == 0,
- users.c.id == closedorders.c.user_id,
+ closed_mapper.isopen == 0,
+ users.c.id == closed_mapper.user_id,
),
lazy="subquery",
- order_by=closedorders.c.id,
+ order_by=closed_mapper.id,
),
),
)
- q = create_session().query(User).order_by(User.id)
-
- def go():
- eq_(
- [
- User(
- id=7,
- addresses=[Address(id=1)],
- open_orders=[Order(id=3)],
- closed_orders=[Order(id=1), Order(id=5)],
- ),
- User(
- id=8,
- addresses=[
- Address(id=2),
- Address(id=3),
- Address(id=4),
- ],
- open_orders=[],
- closed_orders=[],
- ),
- User(
- id=9,
- addresses=[Address(id=5)],
- open_orders=[Order(id=4)],
- closed_orders=[Order(id=2)],
- ),
- User(id=10),
- ],
- q.all(),
- )
-
- self.assert_sql_count(testing.db, go, 4)
+ self._run_double_test()
def test_double_same_mappers(self):
"""Eager loading with two relationships simultaneously,
),
),
)
+ self._run_double_test()
+
+ def _run_double_test(self, no_items=False):
+ User, Address, Order, Item = self.classes(
+ "User", "Address", "Order", "Item"
+ )
q = create_session().query(User).order_by(User.id)
+ def items(*ids):
+ if no_items:
+ return {}
+ else:
+ return {"items": [Item(id=id_) for id_ in ids]}
+
def go():
eq_(
[
User(
id=7,
addresses=[Address(id=1)],
- open_orders=[
- Order(
- id=3,
- items=[Item(id=3), Item(id=4), Item(id=5)],
- )
- ],
+ open_orders=[Order(id=3, **items(3, 4, 5))],
closed_orders=[
- Order(
- id=1,
- items=[Item(id=1), Item(id=2), Item(id=3)],
- ),
- Order(id=5, items=[Item(id=5)]),
+ Order(id=1, **items(1, 2, 3)),
+ Order(id=5, **items(5)),
],
),
User(
User(
id=9,
addresses=[Address(id=5)],
- open_orders=[
- Order(id=4, items=[Item(id=1), Item(id=5)])
- ],
- closed_orders=[
- Order(
- id=2,
- items=[Item(id=1), Item(id=2), Item(id=3)],
- )
- ],
+ open_orders=[Order(id=4, **items(1, 5))],
+ closed_orders=[Order(id=2, **items(1, 2, 3))],
),
User(id=10),
],
q.all(),
)
- self.assert_sql_count(testing.db, go, 6)
+ if no_items:
+ self.assert_sql_count(testing.db, go, 4)
+ else:
+ self.assert_sql_count(testing.db, go, 6)
def test_limit(self):
"""Limit operations combined with lazy-load relationships."""
order_by=orders.c.id,
),
"max_order": relationship(
- mapper(Order, max_orders, non_primary=True),
- lazy="subquery",
- uselist=False,
+ aliased(Order, max_orders), lazy="subquery", uselist=False
),
},
)
mapper(Keyword, keywords)
- # note that we are breaking a rule here, making a second
- # mapper(Keyword, keywords) the reorganization of mapper construction
- # affected this, but was fixed again
-
mapper(
IKAssociation,
item_keywords,
primary_key=[item_keywords.c.item_id, item_keywords.c.keyword_id],
properties=dict(
keyword=relationship(
- mapper(Keyword, keywords, non_primary=True),
+ Keyword,
lazy="joined",
uselist=False,
# note here is a valid place where
@property
def window_functions(self):
return only_if(
- ["postgresql>=8.4", "mssql", "oracle"],
+ ["postgresql>=8.4", "mssql", "oracle", "sqlite>=3.25.0"],
"Backend does not support window functions",
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
