summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/sql/_typing.py
blob: a5da8780275233a4855c12426bb607d32bb71077 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
from __future__ import annotations

from typing import Any
from typing import Iterable
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union

from . import roles
from .. import util
from ..inspection import Inspectable
from ..util.typing import Literal
from ..util.typing import Protocol

if TYPE_CHECKING:
    from .elements import ClauseElement
    from .elements import ColumnClause
    from .elements import ColumnElement
    from .elements import quoted_name
    from .elements import SQLCoreOperations
    from .elements import TextClause
    from .roles import ColumnsClauseRole
    from .roles import FromClauseRole
    from .schema import DefaultGenerator
    from .schema import Sequence
    from .selectable import FromClause
    from .selectable import NamedFromClause
    from .selectable import TableClause
    from .sqltypes import TupleType
    from .type_api import TypeEngine
    from ..util.typing import TypeGuard

_T = TypeVar("_T", bound=Any)


class _HasClauseElement(Protocol):
    """indicates a class that has a __clause_element__() method"""

    def __clause_element__(self) -> ColumnsClauseRole:
        ...


# convention:
# XYZArgument - something that the end user is passing to a public API method
# XYZElement - the internal representation that we use for the thing.
# the coercions system is responsible for converting from XYZArgument to
# XYZElement.

_ColumnsClauseArgument = Union[
    Literal["*", 1],
    roles.ColumnsClauseRole,
    Type[Any],
    Inspectable[_HasClauseElement],
    _HasClauseElement,
]

_SelectIterable = Iterable[Union["ColumnElement[Any]", "TextClause"]]

_FromClauseArgument = Union[
    roles.FromClauseRole,
    Type[Any],
    Inspectable[_HasClauseElement],
    _HasClauseElement,
]

_ColumnExpressionArgument = Union[
    "ColumnElement[_T]", _HasClauseElement, roles.ExpressionElementRole[_T]
]

_DMLColumnArgument = Union[str, "ColumnClause[Any]", _HasClauseElement]

_PropagateAttrsType = util.immutabledict[str, Any]

_TypeEngineArgument = Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"]


def is_named_from_clause(t: FromClauseRole) -> TypeGuard[NamedFromClause]:
    return t.named_with_column


def is_column_element(c: ClauseElement) -> TypeGuard[ColumnElement[Any]]:
    return c._is_column_element


def is_text_clause(c: ClauseElement) -> TypeGuard[TextClause]:
    return c._is_text_clause


def has_schema_attr(t: FromClauseRole) -> TypeGuard[TableClause]:
    return hasattr(t, "schema")


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__")