summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/sql/_typing.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2022-03-30 18:01:58 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2022-04-04 09:26:43 -0400
commit3b4d62f4f72e8dfad7f38db192a6a90a8551608c (patch)
treed0334c4bb52f803bd7dad661f2e6a12e25f5880c /lib/sqlalchemy/sql/_typing.py
parent4e603e23755f31278f27a45449120a8dea470a45 (diff)
downloadsqlalchemy-3b4d62f4f72e8dfad7f38db192a6a90a8551608c.tar.gz
pep484 - sql.selectable
the pep484 task becomes more intense as there is mounting pressure to come up with a consistency in how data moves from end-user to instance variable. current thinking is coming into: 1. there are _typing._XYZArgument objects that represent "what the user sent" 2. there's the roles, which represent a kind of "filter" for different kinds of objects. These are mostly important as the argument we pass to coerce(). 3. there's the thing that coerce() returns, which should be what the construct uses as its internal representation of the thing. This is _typing._XYZElement. but there's some controversy over whether or not we should pass actual ClauseElements around by their role or not. I think we shouldn't at the moment, but this makes the "role-ness" of something a little less portable. Like, we have to set DMLTableRole for TableClause, Join, and Alias, but then also we have to repeat those three types in order to set up _DMLTableElement. Other change introduced here, there was a deannotate=True for the left/right of a sql.join(). All tests pass without that. I'd rather not have that there as if we have a join(A, B) where A, B are mapped classes, we want them inside of the _annotations. The rationale seems to be performance, but this performance can be illustrated to be on the compile side which we hope is cached in the normal case. CTEs now accommodate for text selects including recursive. Get typing to accommodate "util.preloaded" cleanly; add "preloaded" as a real module. This seemed like we would have needed pep562 `__getattr__()` but we don't, just set names in globals() as we import them. References: #6810 Change-Id: I34d17f617de2fe2c086fc556bd55748dc782faf0
Diffstat (limited to 'lib/sqlalchemy/sql/_typing.py')
-rw-r--r--lib/sqlalchemy/sql/_typing.py135
1 files changed, 120 insertions, 15 deletions
diff --git a/lib/sqlalchemy/sql/_typing.py b/lib/sqlalchemy/sql/_typing.py
index a5da87802..0a72a93c5 100644
--- a/lib/sqlalchemy/sql/_typing.py
+++ b/lib/sqlalchemy/sql/_typing.py
@@ -1,7 +1,7 @@
from __future__ import annotations
+import operator
from typing import Any
-from typing import Iterable
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
@@ -24,9 +24,16 @@ if TYPE_CHECKING:
from .roles import FromClauseRole
from .schema import DefaultGenerator
from .schema import Sequence
+ from .selectable import Alias
from .selectable import FromClause
+ from .selectable import Join
from .selectable import NamedFromClause
+ from .selectable import ReturnsRows
+ from .selectable import Select
+ from .selectable import SelectBase
+ from .selectable import Subquery
from .selectable import TableClause
+ from .sqltypes import TableValueType
from .sqltypes import TupleType
from .type_api import TypeEngine
from ..util.typing import TypeGuard
@@ -47,6 +54,14 @@ class _HasClauseElement(Protocol):
# the coercions system is responsible for converting from XYZArgument to
# XYZElement.
+_TextCoercedExpressionArgument = Union[
+ str,
+ "TextClause",
+ "ColumnElement[_T]",
+ _HasClauseElement,
+ roles.ExpressionElementRole[_T],
+]
+
_ColumnsClauseArgument = Union[
Literal["*", 1],
roles.ColumnsClauseRole,
@@ -54,8 +69,31 @@ _ColumnsClauseArgument = Union[
Inspectable[_HasClauseElement],
_HasClauseElement,
]
+"""open-ended SELECT columns clause argument.
+
+Includes column expressions, tables, ORM mapped entities, a few literal values.
+
+This type is used for lists of columns / entities to be returned in result
+sets; select(...), insert().returning(...), etc.
+
+
+"""
+
+_ColumnExpressionArgument = Union[
+ "ColumnElement[_T]", _HasClauseElement, roles.ExpressionElementRole[_T]
+]
+"""narrower "column expression" argument.
+
+This type is used for all the other "column" kinds of expressions that
+typically represent a single SQL column expression, not a set of columns the
+way a table or ORM entity does.
+
+This includes ColumnElement, or ORM-mapped attributes that will have a
+`__clause_element__()` method, it also has the ExpressionElementRole
+overall which brings in the TextClause object also.
+
+"""
-_SelectIterable = Iterable[Union["ColumnElement[Any]", "TextClause"]]
_FromClauseArgument = Union[
roles.FromClauseRole,
@@ -63,28 +101,99 @@ _FromClauseArgument = Union[
Inspectable[_HasClauseElement],
_HasClauseElement,
]
+"""A FROM clause, like we would send to select().select_from().
-_ColumnExpressionArgument = Union[
- "ColumnElement[_T]", _HasClauseElement, roles.ExpressionElementRole[_T]
+Also accommodates ORM entities and related constructs.
+
+"""
+
+_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole]
+"""target for join() builds on _FromClauseArgument to include additional
+join target roles such as those which come from the ORM.
+
+"""
+
+_OnClauseArgument = Union[_ColumnExpressionArgument[Any], roles.OnClauseRole]
+"""target for an ON clause, includes additional roles such as those which
+come from the ORM.
+
+"""
+
+_SelectStatementForCompoundArgument = Union[
+ "SelectBase", roles.CompoundElementRole
+]
+"""SELECT statement acceptable by ``union()`` and other SQL set operations"""
+
+_DMLColumnArgument = Union[
+ str, "ColumnClause[Any]", _HasClauseElement, roles.DMLColumnRole
]
+"""A DML column expression. This is a "key" inside of insert().values(),
+update().values(), and related.
+
+These are usually strings or SQL table columns.
+
+There's also edge cases like JSON expression assignment, which we would want
+the DMLColumnRole to be able to accommodate.
-_DMLColumnArgument = Union[str, "ColumnClause[Any]", _HasClauseElement]
+"""
+
+
+_DMLTableArgument = Union[
+ "TableClause",
+ "Join",
+ "Alias",
+ Type[Any],
+ Inspectable[_HasClauseElement],
+ _HasClauseElement,
+]
_PropagateAttrsType = util.immutabledict[str, Any]
_TypeEngineArgument = Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"]
+if TYPE_CHECKING:
-def is_named_from_clause(t: FromClauseRole) -> TypeGuard[NamedFromClause]:
- return t.named_with_column
+ def is_named_from_clause(t: FromClauseRole) -> TypeGuard[NamedFromClause]:
+ ...
+ def is_column_element(c: ClauseElement) -> TypeGuard[ColumnElement[Any]]:
+ ...
-def is_column_element(c: ClauseElement) -> TypeGuard[ColumnElement[Any]]:
- return c._is_column_element
+ def is_text_clause(c: ClauseElement) -> TypeGuard[TextClause]:
+ ...
+ def is_from_clause(c: ClauseElement) -> TypeGuard[FromClause]:
+ ...
-def is_text_clause(c: ClauseElement) -> TypeGuard[TextClause]:
- return c._is_text_clause
+ def is_tuple_type(t: TypeEngine[Any]) -> TypeGuard[TupleType]:
+ ...
+
+ def is_table_value_type(t: TypeEngine[Any]) -> TypeGuard[TableValueType]:
+ ...
+
+ def is_select_base(t: ReturnsRows) -> TypeGuard[SelectBase]:
+ ...
+
+ def is_select_statement(t: ReturnsRows) -> TypeGuard[Select]:
+ ...
+
+ def is_table(t: FromClause) -> TypeGuard[TableClause]:
+ ...
+
+ def is_subquery(t: FromClause) -> TypeGuard[Subquery]:
+ ...
+
+else:
+ is_named_from_clause = operator.attrgetter("named_with_column")
+ is_column_element = operator.attrgetter("_is_column_element")
+ is_text_clause = operator.attrgetter("_is_text_clause")
+ is_from_clause = operator.attrgetter("_is_from_clause")
+ is_tuple_type = operator.attrgetter("_is_tuple_type")
+ is_table_value_type = operator.attrgetter("_is_table_value")
+ is_select_base = operator.attrgetter("_is_select_base")
+ is_select_statement = operator.attrgetter("_is_select_statement")
+ is_table = operator.attrgetter("_is_table")
+ is_subquery = operator.attrgetter("_is_subquery")
def has_schema_attr(t: FromClauseRole) -> TypeGuard[TableClause]:
@@ -95,9 +204,5 @@ def is_quoted_name(s: str) -> TypeGuard[quoted_name]:
return hasattr(s, "quote")
-def is_tuple_type(t: TypeEngine[Any]) -> TypeGuard[TupleType]:
- return t._is_tuple_type
-
-
def is_has_clause_element(s: object) -> TypeGuard[_HasClauseElement]:
return hasattr(s, "__clause_element__")