_T = TypeVar("_T", bound=Any)
-class aggregate_order_by(expression.ColumnElement[_T]):
- """Represent a PostgreSQL aggregate order by expression.
+class aggregate_order_by(elements.AggregateOrderBy[_T]):
+ """Represent an aggregate order by expression.
+
+ .. deprecated:: 2.0.X
+ The PostgreSQL-specific :class:`aggregate_order_by` is deprecated.
+ Please use :func:`sqlalchemy.sql.expression.aggregate_order_by` instead,
+ which is now available as core functionality.
E.g.::
.. versionchanged:: 1.2.13 - the ORDER BY argument may be multiple terms
+ .. versionchanged:: 2.0.X - moved to core functionality as :func:`sqlalchemy.sql.expression.aggregate_order_by`
+
.. seealso::
+ :func:`sqlalchemy.sql.expression.aggregate_order_by`
+
:class:`_functions.array_agg`
"""
- __visit_name__ = "aggregate_order_by"
-
- stringify_dialect = "postgresql"
- _traverse_internals: _TraverseInternalsType = [
- ("target", InternalTraversal.dp_clauseelement),
- ("type", InternalTraversal.dp_type),
- ("order_by", InternalTraversal.dp_clauseelement),
- ]
-
- @overload
- def __init__(
- self,
- target: ColumnElement[_T],
- *order_by: _ColumnExpressionArgument[Any],
- ): ...
-
- @overload
- def __init__(
- self,
- target: _ColumnExpressionArgument[_T],
- *order_by: _ColumnExpressionArgument[Any],
- ): ...
-
- def __init__(
- self,
- target: _ColumnExpressionArgument[_T],
- *order_by: _ColumnExpressionArgument[Any],
- ):
- self.target: ClauseElement = coercions.expect(
- roles.ExpressionElementRole, target
- )
- self.type = self.target.type
-
- _lob = len(order_by)
- self.order_by: ClauseElement
- if _lob == 0:
- raise TypeError("at least one ORDER BY element is required")
- elif _lob == 1:
- self.order_by = coercions.expect(
- roles.ExpressionElementRole, order_by[0]
- )
- else:
- self.order_by = elements.ClauseList(
- *order_by, _literal_as_text_role=roles.ExpressionElementRole
- )
-
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> ClauseElement:
- return self
-
- def get_children(self, **kwargs: Any) -> Iterable[ClauseElement]:
- return self.target, self.order_by
-
- def _copy_internals(
- self, clone: _CloneCallableType = elements._clone, **kw: Any
- ) -> None:
- self.target = clone(self.target, **kw)
- self.order_by = clone(self.order_by, **kw)
-
- @property
- def _from_objects(self) -> List[FromClause]:
- return self.target._from_objects + self.order_by._from_objects
-
-
class ExcludeConstraint(ColumnCollectionConstraint):
"""A table-level EXCLUDE constraint.
from .ddl import DDL as DDL
from .ddl import DDLElement as DDLElement
from .ddl import ExecutableDDLElement as ExecutableDDLElement
+from .expression import aggregate_order_by as aggregate_order_by
from .expression import Alias as Alias
from .expression import alias as alias
from .expression import all_ as all_
from . import roles
from .base import _NoArg
from .coercions import _document_text_coercion
+from .elements import AggregateOrderBy
from .elements import BindParameter
from .elements import BooleanClauseList
from .elements import Case
"""
return WithinGroup(element, *order_by)
+
+
+def aggregate_order_by(
+ target: _ColumnExpressionArgument[_T], *order_by: _ColumnExpressionArgument[Any]
+) -> AggregateOrderBy[_T]:
+ """Produce an :class:`.AggregateOrderBy` object.
+
+ Used for aggregating functions that support ordering, typically used within
+ aggregate functions like :func:`.func.array_agg` or :func:`.func.string_agg`.
+
+ E.g.::
+
+ from sqlalchemy import func, select
+ from sqlalchemy.sql import aggregate_order_by
+
+ expr = func.array_agg(aggregate_order_by(table.c.a, table.c.b.desc()))
+ stmt = select(expr)
+
+ would represent the expression:
+
+ .. sourcecode:: sql
+
+ SELECT array_agg(a ORDER BY b DESC) FROM table;
+
+ Similarly::
+
+ expr = func.string_agg(
+ table.c.a, aggregate_order_by(literal_column("','"), table.c.a)
+ )
+ stmt = select(expr)
+
+ Would represent:
+
+ .. sourcecode:: sql
+
+ SELECT string_agg(a, ',' ORDER BY a) FROM table;
+
+ The ORDER BY argument may be multiple terms.
+
+ .. versionadded:: 2.0.X Moved from PostgreSQL-specific to core functionality
+
+ .. seealso::
+
+ :class:`.AggregateOrderBy`
+
+ :class:`_functions.array_agg`
+
+ """
+ return AggregateOrderBy(target, *order_by)
funcfilter.criterion._compiler_dispatch(self, **kwargs),
)
+ def visit_aggregate_order_by(self, aggregate_order_by, **kwargs):
+ return "%s ORDER BY %s" % (
+ aggregate_order_by.target._compiler_dispatch(self, **kwargs),
+ aggregate_order_by.order_by._compiler_dispatch(self, **kwargs),
+ )
+
def visit_extract(self, extract, **kwargs):
field = self.extract_map.get(extract.field, extract.field)
return "EXTRACT(%s FROM %s)" % (
)
+class AggregateOrderBy(ColumnElement[_T]):
+ """Represent an aggregate ORDER BY expression.
+
+ E.g.::
+
+ from sqlalchemy import func, select
+ from sqlalchemy.sql import aggregate_order_by
+
+ expr = func.array_agg(aggregate_order_by(table.c.a, table.c.b.desc()))
+ stmt = select(expr)
+
+ would represent the expression:
+
+ .. sourcecode:: sql
+
+ SELECT array_agg(a ORDER BY b DESC) FROM table;
+
+ Similarly::
+
+ expr = func.string_agg(
+ table.c.a, aggregate_order_by(literal_column("','"), table.c.a)
+ )
+ stmt = select(expr)
+
+ Would represent:
+
+ .. sourcecode:: sql
+
+ SELECT string_agg(a, ',' ORDER BY a) FROM table;
+
+ The ORDER BY argument may be multiple terms.
+
+ .. versionadded:: 2.0.X Moved from PostgreSQL-specific to core functionality
+
+ .. seealso::
+
+ :class:`_functions.array_agg`
+
+ """
+
+ __visit_name__ = "aggregate_order_by"
+
+ _traverse_internals: _TraverseInternalsType = [
+ ("target", InternalTraversal.dp_clauseelement),
+ ("type", InternalTraversal.dp_type),
+ ("order_by", InternalTraversal.dp_clauseelement),
+ ]
+
+ @overload
+ def __init__(
+ self,
+ target: ColumnElement[_T],
+ *order_by: _ColumnExpressionArgument[Any],
+ ): ...
+
+ @overload
+ def __init__(
+ self,
+ target: _ColumnExpressionArgument[_T],
+ *order_by: _ColumnExpressionArgument[Any],
+ ): ...
+
+ def __init__(
+ self,
+ target: _ColumnExpressionArgument[_T],
+ *order_by: _ColumnExpressionArgument[Any],
+ ):
+ self.target: ClauseElement = coercions.expect(
+ roles.ExpressionElementRole, target
+ )
+ self.type = self.target.type
+
+ _lob = len(order_by)
+ self.order_by: ClauseElement
+ if _lob == 0:
+ raise TypeError("at least one ORDER BY element is required")
+ elif _lob == 1:
+ self.order_by = coercions.expect(
+ roles.ExpressionElementRole, order_by[0]
+ )
+ else:
+ self.order_by = ClauseList(
+ *order_by, _literal_as_text_role=roles.ExpressionElementRole
+ )
+
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> ClauseElement:
+ return self
+
+ def get_children(self, **kwargs: Any) -> Iterable[ClauseElement]:
+ return self.target, self.order_by
+
+ def _copy_internals(
+ self, clone: _CloneCallableType = _clone, **kw: Any
+ ) -> None:
+ self.target = clone(self.target, **kw)
+ self.order_by = clone(self.order_by, **kw)
+
+ @property
+ def _from_objects(self) -> List[FromClause]:
+ return self.target._from_objects + self.order_by._from_objects
+
+
class NamedColumn(KeyedColumnElement[_T]):
is_literal = False
table: Optional[FromClause] = None
from ._dml_constructors import delete as delete
from ._dml_constructors import insert as insert
from ._dml_constructors import update as update
+from ._elements_constructors import aggregate_order_by as aggregate_order_by
from ._elements_constructors import all_ as all_
from ._elements_constructors import and_ as and_
from ._elements_constructors import any_ as any_
from .elements import Case as Case
from .elements import Cast as Cast
from .elements import ClauseElement as ClauseElement
+from .elements import AggregateOrderBy as AggregateOrderBy
from .elements import ClauseList as ClauseList
from .elements import CollectionAggregate as CollectionAggregate
from .elements import ColumnClause as ColumnClause
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
