codekingpro/portable-devtools

1# sql/_typing.py
2# Copyright (C) 2022-2024 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7
8from __future__ import annotations
9
10import operator
11from typing import Any
12from typing import Callable
13from typing import Dict
14from typing import Generic
15from typing import Iterable
16from typing import Mapping
17from typing import NoReturn
18from typing import Optional
19from typing import overload
20from typing import Set
21from typing import Tuple
22from typing import Type
23from typing import TYPE_CHECKING
24from typing import TypeVar
25from typing import Union
26
27from . import roles
28from .. import exc
29from .. import util
30from ..inspection import Inspectable
31from ..util.typing import Literal
32from ..util.typing import Protocol
33from ..util.typing import TypeAlias
34
35if TYPE_CHECKING:
36    from datetime import date
37    from datetime import datetime
38    from datetime import time
39    from datetime import timedelta
40    from decimal import Decimal
41    from uuid import UUID
42
43    from .base import Executable
44    from .compiler import Compiled
45    from .compiler import DDLCompiler
46    from .compiler import SQLCompiler
47    from .dml import UpdateBase
48    from .dml import ValuesBase
49    from .elements import ClauseElement
50    from .elements import ColumnElement
51    from .elements import KeyedColumnElement
52    from .elements import quoted_name
53    from .elements import SQLCoreOperations
54    from .elements import TextClause
55    from .lambdas import LambdaElement
56    from .roles import FromClauseRole
57    from .schema import Column
58    from .selectable import Alias
59    from .selectable import CTE
60    from .selectable import FromClause
61    from .selectable import Join
62    from .selectable import NamedFromClause
63    from .selectable import ReturnsRows
64    from .selectable import Select
65    from .selectable import Selectable
66    from .selectable import SelectBase
67    from .selectable import Subquery
68    from .selectable import TableClause
69    from .sqltypes import TableValueType
70    from .sqltypes import TupleType
71    from .type_api import TypeEngine
72    from ..engine import Dialect
73    from ..util.typing import TypeGuard
74
75_T = TypeVar("_T", bound=Any)
76_T_co = TypeVar("_T_co", bound=Any, covariant=True)
77
78
79_CE = TypeVar("_CE", bound="ColumnElement[Any]")
80
81_CLE = TypeVar("_CLE", bound="ClauseElement")
82
83
84class _HasClauseElement(Protocol, Generic[_T_co]):
85    """indicates a class that has a __clause_element__() method"""
86
87    def __clause_element__(self) -> roles.ExpressionElementRole[_T_co]: ...
88
89
90class _CoreAdapterProto(Protocol):
91    """protocol for the ClauseAdapter/ColumnAdapter.traverse() method."""
92
93    def __call__(self, obj: _CE) -> _CE: ...
94
95
96class _HasDialect(Protocol):
97    """protocol for Engine/Connection-like objects that have dialect
98    attribute.
99    """
100
101    @property
102    def dialect(self) -> Dialect: ...
103
104
105# match column types that are not ORM entities
106_NOT_ENTITY = TypeVar(
107    "_NOT_ENTITY",
108    int,
109    str,
110    bool,
111    "datetime",
112    "date",
113    "time",
114    "timedelta",
115    "UUID",
116    float,
117    "Decimal",
118)
119
120_StarOrOne = Literal["*", 1]
121
122_MAYBE_ENTITY = TypeVar(
123    "_MAYBE_ENTITY",
124    roles.ColumnsClauseRole,
125    _StarOrOne,
126    Type[Any],
127    Inspectable[_HasClauseElement[Any]],
128    _HasClauseElement[Any],
129)
130
131
132# convention:
133# XYZArgument - something that the end user is passing to a public API method
134# XYZElement - the internal representation that we use for the thing.
135# the coercions system is responsible for converting from XYZArgument to
136# XYZElement.
137
138_TextCoercedExpressionArgument = Union[
139    str,
140    "TextClause",
141    "ColumnElement[_T]",
142    _HasClauseElement[_T],
143    roles.ExpressionElementRole[_T],
144]
145
146_ColumnsClauseArgument = Union[
147    roles.TypedColumnsClauseRole[_T],
148    roles.ColumnsClauseRole,
149    "SQLCoreOperations[_T]",
150    _StarOrOne,
151    Type[_T],
152    Inspectable[_HasClauseElement[_T]],
153    _HasClauseElement[_T],
154]
155"""open-ended SELECT columns clause argument.
156
157Includes column expressions, tables, ORM mapped entities, a few literal values.
158
159This type is used for lists of columns  / entities to be returned in result
160sets; select(...), insert().returning(...), etc.
161
162
163"""
164
165_TypedColumnClauseArgument = Union[
166    roles.TypedColumnsClauseRole[_T],
167    "SQLCoreOperations[_T]",
168    Type[_T],
169]
170
171_TP = TypeVar("_TP", bound=Tuple[Any, ...])
172
173_T0 = TypeVar("_T0", bound=Any)
174_T1 = TypeVar("_T1", bound=Any)
175_T2 = TypeVar("_T2", bound=Any)
176_T3 = TypeVar("_T3", bound=Any)
177_T4 = TypeVar("_T4", bound=Any)
178_T5 = TypeVar("_T5", bound=Any)
179_T6 = TypeVar("_T6", bound=Any)
180_T7 = TypeVar("_T7", bound=Any)
181_T8 = TypeVar("_T8", bound=Any)
182_T9 = TypeVar("_T9", bound=Any)
183
184
185_ColumnExpressionArgument = Union[
186    "ColumnElement[_T]",
187    _HasClauseElement[_T],
188    "SQLCoreOperations[_T]",
189    roles.ExpressionElementRole[_T],
190    roles.TypedColumnsClauseRole[_T],
191    Callable[[], "ColumnElement[_T]"],
192    "LambdaElement",
193]
194"See docs in public alias ColumnExpressionArgument."
195
196ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T]
197"""Narrower "column expression" argument.
198
199This type is used for all the other "column" kinds of expressions that
200typically represent a single SQL column expression, not a set of columns the
201way a table or ORM entity does.
202
203This includes ColumnElement, or ORM-mapped attributes that will have a
204``__clause_element__()`` method, it also has the ExpressionElementRole
205overall which brings in the TextClause object also.
206
207.. versionadded:: 2.0.13
208
209"""
210
211_ColumnExpressionOrLiteralArgument = Union[Any, _ColumnExpressionArgument[_T]]
212
213_ColumnExpressionOrStrLabelArgument = Union[str, _ColumnExpressionArgument[_T]]
214
215_ByArgument = Union[
216    Iterable[_ColumnExpressionOrStrLabelArgument[Any]],
217    _ColumnExpressionOrStrLabelArgument[Any],
218]
219"""Used for keyword-based ``order_by`` and ``partition_by`` parameters."""
220
221
222_InfoType = Dict[Any, Any]
223"""the .info dictionary accepted and used throughout Core /ORM"""
224
225_FromClauseArgument = Union[
226    roles.FromClauseRole,
227    Type[Any],
228    Inspectable[_HasClauseElement[Any]],
229    _HasClauseElement[Any],
230]
231"""A FROM clause, like we would send to select().select_from().
232
233Also accommodates ORM entities and related constructs.
234
235"""
236
237_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole]
238"""target for join() builds on _FromClauseArgument to include additional
239join target roles such as those which come from the ORM.
240
241"""
242
243_OnClauseArgument = Union[_ColumnExpressionArgument[Any], roles.OnClauseRole]
244"""target for an ON clause, includes additional roles such as those which
245come from the ORM.
246
247"""
248
249_SelectStatementForCompoundArgument = Union[
250    "SelectBase", roles.CompoundElementRole
251]
252"""SELECT statement acceptable by ``union()`` and other SQL set operations"""
253
254_DMLColumnArgument = Union[
255    str,
256    _HasClauseElement[Any],
257    roles.DMLColumnRole,
258    "SQLCoreOperations[Any]",
259]
260"""A DML column expression.  This is a "key" inside of insert().values(),
261update().values(), and related.
262
263These are usually strings or SQL table columns.
264
265There's also edge cases like JSON expression assignment, which we would want
266the DMLColumnRole to be able to accommodate.
267
268"""
269
270_DMLKey = TypeVar("_DMLKey", bound=_DMLColumnArgument)
271_DMLColumnKeyMapping = Mapping[_DMLKey, Any]
272
273
274_DDLColumnArgument = Union[str, "Column[Any]", roles.DDLConstraintColumnRole]
275"""DDL column.
276
277used for :class:`.PrimaryKeyConstraint`, :class:`.UniqueConstraint`, etc.
278
279"""
280
281_DMLTableArgument = Union[
282    "TableClause",
283    "Join",
284    "Alias",
285    "CTE",
286    Type[Any],
287    Inspectable[_HasClauseElement[Any]],
288    _HasClauseElement[Any],
289]
290
291_PropagateAttrsType = util.immutabledict[str, Any]
292
293_TypeEngineArgument = Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"]
294
295_EquivalentColumnMap = Dict["ColumnElement[Any]", Set["ColumnElement[Any]"]]
296
297_LimitOffsetType = Union[int, _ColumnExpressionArgument[int], None]
298
299_AutoIncrementType = Union[bool, Literal["auto", "ignore_fk"]]
300
301if TYPE_CHECKING:
302
303    def is_sql_compiler(c: Compiled) -> TypeGuard[SQLCompiler]: ...
304
305    def is_ddl_compiler(c: Compiled) -> TypeGuard[DDLCompiler]: ...
306
307    def is_named_from_clause(
308        t: FromClauseRole,
309    ) -> TypeGuard[NamedFromClause]: ...
310
311    def is_column_element(
312        c: ClauseElement,
313    ) -> TypeGuard[ColumnElement[Any]]: ...
314
315    def is_keyed_column_element(
316        c: ClauseElement,
317    ) -> TypeGuard[KeyedColumnElement[Any]]: ...
318
319    def is_text_clause(c: ClauseElement) -> TypeGuard[TextClause]: ...
320
321    def is_from_clause(c: ClauseElement) -> TypeGuard[FromClause]: ...
322
323    def is_tuple_type(t: TypeEngine[Any]) -> TypeGuard[TupleType]: ...
324
325    def is_table_value_type(
326        t: TypeEngine[Any],
327    ) -> TypeGuard[TableValueType]: ...
328
329    def is_selectable(t: Any) -> TypeGuard[Selectable]: ...
330
331    def is_select_base(
332        t: Union[Executable, ReturnsRows]
333    ) -> TypeGuard[SelectBase]: ...
334
335    def is_select_statement(
336        t: Union[Executable, ReturnsRows]
337    ) -> TypeGuard[Select[Any]]: ...
338
339    def is_table(t: FromClause) -> TypeGuard[TableClause]: ...
340
341    def is_subquery(t: FromClause) -> TypeGuard[Subquery]: ...
342
343    def is_dml(c: ClauseElement) -> TypeGuard[UpdateBase]: ...
344
345else:
346    is_sql_compiler = operator.attrgetter("is_sql")
347    is_ddl_compiler = operator.attrgetter("is_ddl")
348    is_named_from_clause = operator.attrgetter("named_with_column")
349    is_column_element = operator.attrgetter("_is_column_element")
350    is_keyed_column_element = operator.attrgetter("_is_keyed_column_element")
351    is_text_clause = operator.attrgetter("_is_text_clause")
352    is_from_clause = operator.attrgetter("_is_from_clause")
353    is_tuple_type = operator.attrgetter("_is_tuple_type")
354    is_table_value_type = operator.attrgetter("_is_table_value")
355    is_selectable = operator.attrgetter("is_selectable")
356    is_select_base = operator.attrgetter("_is_select_base")
357    is_select_statement = operator.attrgetter("_is_select_statement")
358    is_table = operator.attrgetter("_is_table")
359    is_subquery = operator.attrgetter("_is_subquery")
360    is_dml = operator.attrgetter("is_dml")
361
362
363def has_schema_attr(t: FromClauseRole) -> TypeGuard[TableClause]:
364    return hasattr(t, "schema")
365
366
367def is_quoted_name(s: str) -> TypeGuard[quoted_name]:
368    return hasattr(s, "quote")
369
370
371def is_has_clause_element(s: object) -> TypeGuard[_HasClauseElement[Any]]:
372    return hasattr(s, "__clause_element__")
373
374
375def is_insert_update(c: ClauseElement) -> TypeGuard[ValuesBase]:
376    return c.is_dml and (c.is_insert or c.is_update)  # type: ignore
377
378
379def _no_kw() -> exc.ArgumentError:
380    return exc.ArgumentError(
381        "Additional keyword arguments are not accepted by this "
382        "function/method.  The presence of **kw is for pep-484 typing purposes"
383    )
384
385
386def _unexpected_kw(methname: str, kw: Dict[str, Any]) -> NoReturn:
387    k = list(kw)[0]
388    raise TypeError(f"{methname} got an unexpected keyword argument '{k}'")
389
390
391@overload
392def Nullable(
393    val: "SQLCoreOperations[_T]",
394) -> "SQLCoreOperations[Optional[_T]]": ...
395
396
397@overload
398def Nullable(
399    val: roles.ExpressionElementRole[_T],
400) -> roles.ExpressionElementRole[Optional[_T]]: ...
401
402
403@overload
404def Nullable(val: Type[_T]) -> Type[Optional[_T]]: ...
405
406
407def Nullable(
408    val: _TypedColumnClauseArgument[_T],
409) -> _TypedColumnClauseArgument[Optional[_T]]:
410    """Types a column or ORM class as nullable.
411
412    This can be used in select and other contexts to express that the value of
413    a column can be null, for example due to an outer join::
414
415        stmt1 = select(A, Nullable(B)).outerjoin(A.bs)
416        stmt2 = select(A.data, Nullable(B.data)).outerjoin(A.bs)
417
418    At runtime this method returns the input unchanged.
419
420    .. versionadded:: 2.0.20
421    """
422    return val
423
424
425@overload
426def NotNullable(
427    val: "SQLCoreOperations[Optional[_T]]",
428) -> "SQLCoreOperations[_T]": ...
429
430
431@overload
432def NotNullable(
433    val: roles.ExpressionElementRole[Optional[_T]],
434) -> roles.ExpressionElementRole[_T]: ...
435
436
437@overload
438def NotNullable(val: Type[Optional[_T]]) -> Type[_T]: ...
439
440
441@overload
442def NotNullable(val: Optional[Type[_T]]) -> Type[_T]: ...
443
444
445def NotNullable(
446    val: Union[_TypedColumnClauseArgument[Optional[_T]], Optional[Type[_T]]],
447) -> _TypedColumnClauseArgument[_T]:
448    """Types a column or ORM class as not nullable.
449
450    This can be used in select and other contexts to express that the value of
451    a column cannot be null, for example due to a where condition on a
452    nullable column::
453
454        stmt = select(NotNullable(A.value)).where(A.value.is_not(None))
455
456    At runtime this method returns the input unchanged.
457
458    .. versionadded:: 2.0.20
459    """
460    return val  # type: ignore
461