--- /dev/null
+.. change::
+ :tags: usecase, orm
+ :tickets: 7861
+
+ Added new attributes :attr:`.UpdateBase.returning_column_descriptions` and
+ :attr:`.UpdateBase.entity_description` to allow for inspection of ORM
+ attributes and entities that are installed as part of an :class:`.Insert`,
+ :class:`.Update`, or :class:`.Delete` construct. The
+ :attr:`.Select.column_descriptions` accessor is also now implemented for
+ Core-only selectables.
within the join expression.
plugin
+ plugin-enabled
plugin-specific
- "plugin-specific" generally indicates a function or method in
+ "plugin-enabled" or "plugin-specific" generally indicates a function or method in
SQLAlchemy Core which will behave differently when used in an ORM
context.
>>> conn.close()
ROLLBACK
+.. _queryguide_inspection:
+
+Inspecting entities and columns from ORM-enabled SELECT and DML statements
+==========================================================================
+
+The :func:`.select` construct, as well as the :func:`.insert`, :func:`.update`
+and :func:`.delete` constructs (for the latter DML constructs, as of SQLAlchemy
+1.4.33), all support the ability to inspect the entities in which these
+statements are created against, as well as the columns and datatypes that would
+be returned in a result set.
+
+For a :class:`.Select` object, this information is available from the
+:attr:`.Select.column_descriptions` attribute. This attribute operates in the
+same way as the legacy :attr:`.Query.column_descriptions` attribute. The format
+returned is a list of dictionaries::
+
+ >>> from pprint import pprint
+ >>> user_alias = aliased(User, name='user2')
+ >>> stmt = select(User, User.id, user_alias)
+ >>> pprint(stmt.column_descriptions)
+ [{'aliased': False,
+ 'entity': <class 'User'>,
+ 'expr': <class 'User'>,
+ 'name': 'User',
+ 'type': <class 'User'>},
+ {'aliased': False,
+ 'entity': <class 'User'>,
+ 'expr': <....InstrumentedAttribute object at ...>,
+ 'name': 'id',
+ 'type': Integer()},
+ {'aliased': True,
+ 'entity': <AliasedClass ...; User>,
+ 'expr': <AliasedClass ...; User>,
+ 'name': 'user2',
+ 'type': <class 'User'>}]
+
+
+When :attr:`.Select.column_descriptions` is used with non-ORM objects
+such as plain :class:`.Table` or :class:`.Column` objects, the entries
+will contain basic information about individual columns returned in all
+cases::
+
+ >>> stmt = select(user_table, address_table.c.id)
+ >>> pprint(stmt.column_descriptions)
+ [{'expr': Column('id', Integer(), table=<user_account>, primary_key=True, nullable=False),
+ 'name': 'id',
+ 'type': Integer()},
+ {'expr': Column('name', String(length=30), table=<user_account>),
+ 'name': 'name',
+ 'type': String(length=30)},
+ {'expr': Column('fullname', String(), table=<user_account>),
+ 'name': 'fullname',
+ 'type': String()},
+ {'expr': Column('id', Integer(), table=<address>, primary_key=True, nullable=False),
+ 'name': 'id_1',
+ 'type': Integer()}]
+
+.. versionchanged:: 1.4.33 The :attr:`.Select.column_descriptions` attribute now returns
+ a value when used against a :class:`.Select` that is not ORM-enabled. Previously,
+ this would raise ``NotImplementedError``.
+
+
+For :func:`.insert`, :func:`.update` and :func:`.delete` constructs, there are
+two separate attributes. One is :attr:`.UpdateBase.entity_description` which
+returns information about the primary ORM entity and database table which the
+DML construct would be affecting::
+
+ >>> from sqlalchemy import update
+ >>> stmt = update(User).values(name="somename").returning(User.id)
+ >>> pprint(stmt.entity_description)
+ {'entity': <class 'User'>,
+ 'expr': <class 'User'>,
+ 'name': 'User',
+ 'table': Table('user_account', ...),
+ 'type': <class 'User'>}
+
+.. tip:: The :attr:`.UpdateBase.entity_description` includes an entry
+ ``"table"`` which is actually the **table to be inserted, updated or
+ deleted** by the statement, which is **not** always the same as the SQL
+ "selectable" to which the class may be mapped. For example, in a
+ joined-table inheritance scenario, ``"table"`` will refer to the local table
+ for the given entity.
+
+The other is :attr:`.UpdateBase.returning_column_descriptions` which
+delivers information about the columns present in the RETURNING collection
+in a manner roughly similar to that of :attr:`.Select.column_descriptions`::
+
+ >>> pprint(stmt.returning_column_descriptions)
+ [{'aliased': False,
+ 'entity': <class 'User'>,
+ 'expr': <sqlalchemy.orm.attributes.InstrumentedAttribute ...>,
+ 'name': 'id',
+ 'type': Integer()}]
+
+.. versionadded:: 1.4.33 Added the :attr:`.UpdateBase.entity_description`
+ and :attr:`.UpdateBase.returning_column_descriptions` attributes.
+
+
.. _queryguide_additional:
Additional ORM API Constructs
from ..sql.base import CompileState
from ..sql.base import Options
from ..sql.dml import DeleteDMLState
+from ..sql.dml import InsertDMLState
from ..sql.dml import UpdateDMLState
from ..sql.elements import BooleanClauseList
from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
}
+class ORMDMLState:
+ @classmethod
+ def get_entity_description(cls, statement):
+ ext_info = statement.table._annotations["parententity"]
+ mapper = ext_info.mapper
+ if ext_info.is_aliased_class:
+ _label_name = ext_info.name
+ else:
+ _label_name = mapper.class_.__name__
+
+ return {
+ "name": _label_name,
+ "type": mapper.class_,
+ "expr": ext_info.entity,
+ "entity": ext_info.entity,
+ "table": mapper.local_table,
+ }
+
+ @classmethod
+ def get_returning_column_descriptions(cls, statement):
+ def _ent_for_col(c):
+ return c._annotations.get("parententity", None)
+
+ def _attr_for_col(c, ent):
+ if ent is None:
+ return c
+ proxy_key = c._annotations.get("proxy_key", None)
+ if not proxy_key:
+ return c
+ else:
+ return getattr(ent.entity, proxy_key, c)
+
+ return [
+ {
+ "name": c.key,
+ "type": c.type,
+ "expr": _attr_for_col(c, ent),
+ "aliased": ent.is_aliased_class,
+ "entity": ent.entity,
+ }
+ for c, ent in [
+ (c, _ent_for_col(c)) for c in statement._all_selected_columns
+ ]
+ ]
+
+
+@CompileState.plugin_for("orm", "insert")
+class ORMInsert(ORMDMLState, InsertDMLState):
+ pass
+
+
@CompileState.plugin_for("orm", "update")
-class BulkORMUpdate(UpdateDMLState, BulkUDCompileState):
+class BulkORMUpdate(ORMDMLState, UpdateDMLState, BulkUDCompileState):
@classmethod
def create_for_statement(cls, statement, compiler, **kw):
@CompileState.plugin_for("orm", "delete")
-class BulkORMDelete(DeleteDMLState, BulkUDCompileState):
+class BulkORMDelete(ORMDMLState, DeleteDMLState, BulkUDCompileState):
@classmethod
def create_for_statement(cls, statement, compiler, **kw):
self = cls.__new__(cls)
}
]
+ .. seealso::
+
+ This API is available using :term:`2.0 style` queries as well,
+ documented at:
+
+ * :ref:`queryguide_inspection`
+
+ * :attr:`.Select.column_descriptions`
+
"""
return _column_descriptions(self, legacy=True)
def __init__(self, statement, compiler, **kw):
raise NotImplementedError()
+ @classmethod
+ def get_entity_description(cls, statement):
+ return {"name": statement.table.name, "table": statement.table}
+
+ @classmethod
+ def get_returning_column_descriptions(cls, statement):
+ return [
+ {
+ "name": c.key,
+ "type": c.type,
+ "expr": c,
+ }
+ for c in statement._all_selected_columns
+ ]
+
@property
def dml_table(self):
return self.statement.table
self._hints = self._hints.union({(selectable, dialect_name): text})
return self
+ @property
+ def entity_description(self):
+ """Return a :term:`plugin-enabled` description of the table and/or entity
+ which this DML construct is operating against.
+
+ This attribute is generally useful when using the ORM, as an
+ extended structure which includes information about mapped
+ entities is returned. The section :ref:`queryguide_inspection`
+ contains more background.
+
+ For a Core statement, the structure returned by this accessor
+ is derived from the :attr:`.UpdateBase.table` attribute, and
+ refers to the :class:`.Table` being inserted, updated, or deleted::
+
+ >>> stmt = insert(user_table)
+ >>> stmt.entity_description
+ {
+ "name": "user_table",
+ "table": Table("user_table", ...)
+ }
+
+ .. versionadded:: 1.4.33
+
+ .. seealso::
+
+ :attr:`.UpdateBase.returning_column_descriptions`
+
+ :attr:`.Select.column_descriptions` - entity information for
+ a :func:`.select` construct
+
+ :ref:`queryguide_inspection` - ORM background
+
+ """
+ meth = DMLState.get_plugin_class(self).get_entity_description
+ return meth(self)
+
+ @property
+ def returning_column_descriptions(self):
+ """Return a :term:`plugin-enabled` description of the columns
+ which this DML construct is RETURNING against, in other words
+ the expressions established as part of :meth:`.UpdateBase.returning`.
+
+ This attribute is generally useful when using the ORM, as an
+ extended structure which includes information about mapped
+ entities is returned. The section :ref:`queryguide_inspection`
+ contains more background.
+
+ For a Core statement, the structure returned by this accessor is
+ derived from the same objects that are returned by the
+ :attr:`.UpdateBase.exported_columns` accessor::
+
+ >>> stmt = insert(user_table).returning(user_table.c.id, user_table.c.name)
+ >>> stmt.entity_description
+ [
+ {
+ "name": "id",
+ "type": Integer,
+ "expr": Column("id", Integer(), table=<user>, ...)
+ },
+ {
+ "name": "name",
+ "type": String(),
+ "expr": Column("name", String(), table=<user>, ...)
+ },
+ ]
+
+ .. versionadded:: 1.4.33
+
+ .. seealso::
+
+ :attr:`.UpdateBase.entity_description`
+
+ :attr:`.Select.column_descriptions` - entity information for
+ a :func:`.select` construct
+
+ :ref:`queryguide_inspection` - ORM background
+
+ """ # noqa E501
+ meth = DMLState.get_plugin_class(
+ self
+ ).get_returning_column_descriptions
+ return meth(self)
+
SelfValuesBase = typing.TypeVar("SelfValuesBase", bound="ValuesBase")
@classmethod
def get_column_descriptions(cls, statement):
- cls._plugin_not_implemented()
+ return [
+ {
+ "name": name,
+ "type": element.type,
+ "expr": element,
+ }
+ for _, name, _, element, _ in (
+ statement._generate_columns_plus_names(False)
+ )
+ ]
@classmethod
def from_statement(cls, statement, from_statement):
@property
def column_descriptions(self):
- """Return a 'column descriptions' structure which may be
- :term:`plugin-specific`.
+ """Return a :term:`plugin-enabled` 'column descriptions' structure
+ referring to the columns which are SELECTed by this statement.
+
+ This attribute is generally useful when using the ORM, as an
+ extended structure which includes information about mapped
+ entities is returned. The section :ref:`queryguide_inspection`
+ contains more background.
+
+ For a Core-only statement, the structure returned by this accessor
+ is derived from the same objects that are returned by the
+ :attr:`.Select.selected_columns` accessor, formatted as a list of
+ dictionaries which contain the keys ``name``, ``type`` and ``expr``,
+ which indicate the column expressions to be selected::
+
+ >>> stmt = select(user_table)
+ >>> stmt.column_descriptions
+ [
+ {
+ 'name': 'id',
+ 'type': Integer(),
+ 'expr': Column('id', Integer(), ...)},
+ {
+ 'name': 'name',
+ 'type': String(length=30),
+ 'expr': Column('name', String(length=30), ...)}
+ ]
+
+ .. versionchanged:: 1.4.33 The :attr:`.Select.column_descriptions`
+ attribute returns a structure for a Core-only set of entities,
+ not just ORM-only entities.
+
+ .. seealso::
+
+ :attr:`.UpdateBase.entity_description` - entity information for
+ an :func:`.insert`, :func:`.update`, or :func:`.delete`
+
+ :ref:`queryguide_inspection` - ORM background
"""
meth = SelectState.get_plugin_class(self).get_column_descriptions
from sqlalchemy import bindparam
from sqlalchemy import Column
+from sqlalchemy import delete
from sqlalchemy import exc
from sqlalchemy import func
from sqlalchemy import insert
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import union
+from sqlalchemy import update
from sqlalchemy import util
from sqlalchemy.orm import aliased
from sqlalchemy.orm import column_property
}
],
),
+ (
+ lambda user_alias: (user_alias,),
+ lambda User, user_alias: [
+ {
+ "name": None,
+ "type": User,
+ "aliased": True,
+ "expr": user_alias,
+ "entity": user_alias,
+ }
+ ],
+ ),
(
lambda User: (User.id,),
lambda User: [
},
],
),
+ (
+ lambda user_table: (user_table,),
+ lambda user_table: [
+ {
+ "name": "id",
+ "type": testing.eq_type_affinity(sqltypes.Integer),
+ "expr": user_table.c.id,
+ },
+ {
+ "name": "name",
+ "type": testing.eq_type_affinity(sqltypes.String),
+ "expr": user_table.c.name,
+ },
+ ],
+ ),
)
def test_column_descriptions(self, cols, expected):
User, Address = self.classes("User", "Address")
+ ua = aliased(User)
- cols = testing.resolve_lambda(cols, User=User, Address=Address)
- expected = testing.resolve_lambda(expected, User=User, Address=Address)
+ cols = testing.resolve_lambda(
+ cols,
+ User=User,
+ Address=Address,
+ user_alias=ua,
+ user_table=inspect(User).local_table,
+ )
+ expected = testing.resolve_lambda(
+ expected,
+ User=User,
+ Address=Address,
+ user_alias=ua,
+ user_table=inspect(User).local_table,
+ )
stmt = select(*cols)
-
eq_(stmt.column_descriptions, expected)
+ @testing.combinations(insert, update, delete, argnames="dml_construct")
+ @testing.combinations(
+ (
+ lambda User: User,
+ lambda User: (User.id, User.name),
+ lambda User, user_table: {
+ "name": "User",
+ "type": User,
+ "expr": User,
+ "entity": User,
+ "table": user_table,
+ },
+ lambda User: [
+ {
+ "name": "id",
+ "type": testing.eq_type_affinity(sqltypes.Integer),
+ "aliased": False,
+ "expr": User.id,
+ "entity": User,
+ },
+ {
+ "name": "name",
+ "type": testing.eq_type_affinity(sqltypes.String),
+ "aliased": False,
+ "expr": User.name,
+ "entity": User,
+ },
+ ],
+ ),
+ argnames="entity, cols, expected_entity, expected_returning",
+ )
+ def test_dml_descriptions(
+ self, dml_construct, entity, cols, expected_entity, expected_returning
+ ):
+ User, Address = self.classes("User", "Address")
+
+ lambda_args = dict(
+ User=User,
+ Address=Address,
+ user_table=inspect(User).local_table,
+ )
+ entity = testing.resolve_lambda(entity, **lambda_args)
+ cols = testing.resolve_lambda(cols, **lambda_args)
+ expected_entity = testing.resolve_lambda(
+ expected_entity, **lambda_args
+ )
+ expected_returning = testing.resolve_lambda(
+ expected_returning, **lambda_args
+ )
+
+ stmt = dml_construct(entity)
+ if cols:
+ stmt = stmt.returning(*cols)
+
+ eq_(stmt.entity_description, expected_entity)
+ eq_(stmt.returning_column_descriptions, expected_returning)
+
class ColumnsClauseFromsTest(QueryTest, AssertsCompiledSQL):
__dialect__ = "default"
from sqlalchemy import Boolean
from sqlalchemy import cast
from sqlalchemy import Column
+from sqlalchemy import delete
from sqlalchemy import exc
from sqlalchemy import exists
from sqlalchemy import false
from sqlalchemy import ForeignKey
from sqlalchemy import func
+from sqlalchemy import insert
from sqlalchemy import Integer
from sqlalchemy import join
from sqlalchemy import literal_column
from sqlalchemy import type_coerce
from sqlalchemy import TypeDecorator
from sqlalchemy import union
+from sqlalchemy import update
from sqlalchemy import util
from sqlalchemy.sql import Alias
from sqlalchemy.sql import annotation
):
__dialect__ = "default"
+ @testing.combinations(
+ (
+ (table1.c.col1, table1.c.col2),
+ [
+ {
+ "name": "col1",
+ "type": table1.c.col1.type,
+ "expr": table1.c.col1,
+ },
+ {
+ "name": "col2",
+ "type": table1.c.col2.type,
+ "expr": table1.c.col2,
+ },
+ ],
+ ),
+ (
+ (table1,),
+ [
+ {
+ "name": "col1",
+ "type": table1.c.col1.type,
+ "expr": table1.c.col1,
+ },
+ {
+ "name": "col2",
+ "type": table1.c.col2.type,
+ "expr": table1.c.col2,
+ },
+ {
+ "name": "col3",
+ "type": table1.c.col3.type,
+ "expr": table1.c.col3,
+ },
+ {
+ "name": "colx",
+ "type": table1.c.colx.type,
+ "expr": table1.c.colx,
+ },
+ ],
+ ),
+ (
+ (func.count(table1.c.col1),),
+ [
+ {
+ "name": "count",
+ "type": testing.eq_type_affinity(Integer),
+ "expr": testing.eq_clause_element(
+ func.count(table1.c.col1)
+ ),
+ }
+ ],
+ ),
+ (
+ (func.count(table1.c.col1), func.count(table1.c.col2)),
+ [
+ {
+ "name": "count",
+ "type": testing.eq_type_affinity(Integer),
+ "expr": testing.eq_clause_element(
+ func.count(table1.c.col1)
+ ),
+ },
+ {
+ "name": "count_1",
+ "type": testing.eq_type_affinity(Integer),
+ "expr": testing.eq_clause_element(
+ func.count(table1.c.col2)
+ ),
+ },
+ ],
+ ),
+ )
+ def test_core_column_descriptions(self, cols, expected):
+ stmt = select(*cols)
+ # reverse eq_ is so eq_clause_element works
+ eq_(expected, stmt.column_descriptions)
+
+ @testing.combinations(insert, update, delete, argnames="dml_construct")
+ @testing.combinations(
+ (
+ table1,
+ (table1.c.col1, table1.c.col2),
+ {"name": "table1", "table": table1},
+ [
+ {
+ "name": "col1",
+ "type": table1.c.col1.type,
+ "expr": table1.c.col1,
+ },
+ {
+ "name": "col2",
+ "type": table1.c.col2.type,
+ "expr": table1.c.col2,
+ },
+ ],
+ ),
+ (
+ table1,
+ (func.count(table1.c.col1),),
+ {"name": "table1", "table": table1},
+ [
+ {
+ "name": None,
+ "type": testing.eq_type_affinity(Integer),
+ "expr": testing.eq_clause_element(
+ func.count(table1.c.col1)
+ ),
+ },
+ ],
+ ),
+ (
+ table1,
+ None,
+ {"name": "table1", "table": table1},
+ [],
+ ),
+ argnames="entity, cols, expected_entity, expected_returning",
+ )
+ def test_dml_descriptions(
+ self, dml_construct, entity, cols, expected_entity, expected_returning
+ ):
+ stmt = dml_construct(entity)
+ if cols:
+ stmt = stmt.returning(*cols)
+
+ eq_(stmt.entity_description, expected_entity)
+ eq_(expected_returning, stmt.returning_column_descriptions)
+
def test_indirect_correspondence_on_labels(self):
# this test depends upon 'distance' to
# get the right result
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
