codekingpro/portable-devtools

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