summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/sql/_typing.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2022-04-19 21:06:41 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2022-04-27 14:46:36 -0400
commitad11c482e2233f44e8747d4d5a2b17a995fff1fa (patch)
tree57f8ddd30928951519fd6ac0f418e9cbf8e65610 /lib/sqlalchemy/sql/_typing.py
parent033d1a16e7a220555d7611a5b8cacb1bd83822ae (diff)
downloadsqlalchemy-ad11c482e2233f44e8747d4d5a2b17a995fff1fa.tar.gz
pep484 ORM / SQL result support
after some experimentation it seems mypy is more amenable to the generic types being fully integrated rather than having separate spin-off types. so key structures like Result, Row, Select become generic. For DML Insert, Update, Delete, these are spun into type-specific subclasses ReturningInsert, ReturningUpdate, ReturningDelete, which is fine since the "row-ness" of these constructs doesn't happen until returning() is called in any case. a Tuple based model is then integrated so that these objects can carry along information about their return types. Overloads at the .execute() level carry through the Tuple from the invoked object to the result. To suit the issue of AliasedClass generating attributes that are dynamic, experimented with a custom subclass AsAliased, but then just settled on having aliased() lie to the type checker and return `Type[_O]`, essentially. will need some type-related accessors for with_polymorphic() also. Additionally, identified an issue in Update when used "mysql style" against a join(), it basically doesn't work if asked to UPDATE two tables on the same column name. added an error message to the specific condition where it happens with a very non-specific error message that we hit a thing we can't do right now, suggest multi-table update as a possible cause. Change-Id: I5eff7eefe1d6166ee74160b2785c5e6a81fa8b95
Diffstat (limited to 'lib/sqlalchemy/sql/_typing.py')
-rw-r--r--lib/sqlalchemy/sql/_typing.py78
1 files changed, 71 insertions, 7 deletions
diff --git a/lib/sqlalchemy/sql/_typing.py b/lib/sqlalchemy/sql/_typing.py
index 53d29b628..1df530dbd 100644
--- a/lib/sqlalchemy/sql/_typing.py
+++ b/lib/sqlalchemy/sql/_typing.py
@@ -5,18 +5,27 @@ from typing import Any
from typing import Callable
from typing import Dict
from typing import Set
+from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
from . import roles
+from .. import exc
from .. import util
from ..inspection import Inspectable
from ..util.typing import Literal
from ..util.typing import Protocol
if TYPE_CHECKING:
+ from datetime import date
+ from datetime import datetime
+ from datetime import time
+ from datetime import timedelta
+ from decimal import Decimal
+ from uuid import UUID
+
from .base import Executable
from .compiler import Compiled
from .compiler import DDLCompiler
@@ -26,17 +35,15 @@ if TYPE_CHECKING:
from .elements import ClauseElement
from .elements import ColumnClause
from .elements import ColumnElement
+ from .elements import KeyedColumnElement
from .elements import quoted_name
- from .elements import SQLCoreOperations
from .elements import TextClause
from .lambdas import LambdaElement
from .roles import ColumnsClauseRole
from .roles import FromClauseRole
from .schema import Column
- from .schema import DefaultGenerator
- from .schema import Sequence
- from .schema import Table
from .selectable import Alias
+ from .selectable import CTE
from .selectable import FromClause
from .selectable import Join
from .selectable import NamedFromClause
@@ -61,6 +68,30 @@ class _HasClauseElement(Protocol):
...
+# match column types that are not ORM entities
+_NOT_ENTITY = TypeVar(
+ "_NOT_ENTITY",
+ int,
+ str,
+ "datetime",
+ "date",
+ "time",
+ "timedelta",
+ "UUID",
+ float,
+ "Decimal",
+)
+
+_MAYBE_ENTITY = TypeVar(
+ "_MAYBE_ENTITY",
+ roles.ColumnsClauseRole,
+ Literal["*", 1],
+ Type[Any],
+ Inspectable[_HasClauseElement],
+ _HasClauseElement,
+)
+
+
# convention:
# XYZArgument - something that the end user is passing to a public API method
# XYZElement - the internal representation that we use for the thing.
@@ -76,9 +107,10 @@ _TextCoercedExpressionArgument = Union[
]
_ColumnsClauseArgument = Union[
- Literal["*", 1],
+ roles.TypedColumnsClauseRole[_T],
roles.ColumnsClauseRole,
- Type[Any],
+ Literal["*", 1],
+ Type[_T],
Inspectable[_HasClauseElement],
_HasClauseElement,
]
@@ -92,6 +124,24 @@ sets; select(...), insert().returning(...), etc.
"""
+_TypedColumnClauseArgument = Union[
+ roles.TypedColumnsClauseRole[_T], roles.ExpressionElementRole[_T], Type[_T]
+]
+
+_TP = TypeVar("_TP", bound=Tuple[Any, ...])
+
+_T0 = TypeVar("_T0", bound=Any)
+_T1 = TypeVar("_T1", bound=Any)
+_T2 = TypeVar("_T2", bound=Any)
+_T3 = TypeVar("_T3", bound=Any)
+_T4 = TypeVar("_T4", bound=Any)
+_T5 = TypeVar("_T5", bound=Any)
+_T6 = TypeVar("_T6", bound=Any)
+_T7 = TypeVar("_T7", bound=Any)
+_T8 = TypeVar("_T8", bound=Any)
+_T9 = TypeVar("_T9", bound=Any)
+
+
_ColumnExpressionArgument = Union[
"ColumnElement[_T]",
_HasClauseElement,
@@ -169,6 +219,7 @@ _DMLTableArgument = Union[
"TableClause",
"Join",
"Alias",
+ "CTE",
Type[Any],
Inspectable[_HasClauseElement],
_HasClauseElement,
@@ -194,6 +245,11 @@ if TYPE_CHECKING:
def is_column_element(c: ClauseElement) -> TypeGuard[ColumnElement[Any]]:
...
+ def is_keyed_column_element(
+ c: ClauseElement,
+ ) -> TypeGuard[KeyedColumnElement[Any]]:
+ ...
+
def is_text_clause(c: ClauseElement) -> TypeGuard[TextClause]:
...
@@ -216,7 +272,7 @@ if TYPE_CHECKING:
def is_select_statement(
t: Union[Executable, ReturnsRows]
- ) -> TypeGuard[Select]:
+ ) -> TypeGuard[Select[Any]]:
...
def is_table(t: FromClause) -> TypeGuard[TableClause]:
@@ -234,6 +290,7 @@ else:
is_ddl_compiler = operator.attrgetter("is_ddl")
is_named_from_clause = operator.attrgetter("named_with_column")
is_column_element = operator.attrgetter("_is_column_element")
+ is_keyed_column_element = operator.attrgetter("_is_keyed_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")
@@ -260,3 +317,10 @@ def is_has_clause_element(s: object) -> TypeGuard[_HasClauseElement]:
def is_insert_update(c: ClauseElement) -> TypeGuard[ValuesBase]:
return c.is_dml and (c.is_insert or c.is_update) # type: ignore
+
+
+def _no_kw() -> exc.ArgumentError:
+ return exc.ArgumentError(
+ "Additional keyword arguments are not accepted by this "
+ "function/method. The presence of **kw is for pep-484 typing purposes"
+ )