Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/_typing.py: 70%

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 Literal 

17from typing import Mapping 

18from typing import NoReturn 

19from typing import Optional 

20from typing import overload 

21from typing import Protocol 

22from typing import Set 

23from typing import Tuple 

24from typing import Type 

25from typing import TYPE_CHECKING 

26from typing import TypeAlias 

27from typing import TypeVar 

28from typing import Union 

29 

30from . import roles 

31from .. import exc 

32from .. import util 

33from ..inspection import Inspectable 

34from ..util.typing import TupleAny 

35from ..util.typing import TypeVarTuple 

36from ..util.typing import Unpack 

37 

38if TYPE_CHECKING: 

39 from datetime import date 

40 from datetime import datetime 

41 from datetime import time 

42 from datetime import timedelta 

43 from decimal import Decimal 

44 from typing import TypeGuard 

45 from uuid import UUID 

46 

47 from .base import Executable 

48 from .compiler import Compiled 

49 from .compiler import DDLCompiler 

50 from .compiler import SQLCompiler 

51 from .dml import UpdateBase 

52 from .dml import ValuesBase 

53 from .elements import ClauseElement 

54 from .elements import ColumnElement 

55 from .elements import KeyedColumnElement 

56 from .elements import quoted_name 

57 from .elements import SQLCoreOperations 

58 from .elements import TextClause 

59 from .lambdas import LambdaElement 

60 from .roles import FromClauseRole 

61 from .schema import Column 

62 from .selectable import Alias 

63 from .selectable import CompoundSelect 

64 from .selectable import CTE 

65 from .selectable import FromClause 

66 from .selectable import Join 

67 from .selectable import NamedFromClause 

68 from .selectable import ReturnsRows 

69 from .selectable import Select 

70 from .selectable import Selectable 

71 from .selectable import SelectBase 

72 from .selectable import Subquery 

73 from .selectable import TableClause 

74 from .sqltypes import TableValueType 

75 from .sqltypes import TupleType 

76 from .type_api import TypeEngine 

77 from ..engine import Connection 

78 from ..engine import Dialect 

79 from ..engine import Engine 

80 from ..engine.mock import MockConnection 

81 

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

83_T_co = TypeVar("_T_co", bound=Any, covariant=True) 

84_Ts = TypeVarTuple("_Ts") 

85_Ts2 = TypeVarTuple("_Ts2") 

86 

87 

88_CE = TypeVar("_CE", bound="ColumnElement[Any]") 

89 

90_CLE = TypeVar("_CLE", bound="ClauseElement") 

91 

92 

93class _HasClauseElement(Protocol, Generic[_T_co]): 

94 """indicates a class that has a __clause_element__() method""" 

95 

96 def __clause_element__(self) -> roles.ExpressionElementRole[_T_co]: ... 

97 

98 

99class _CoreAdapterProto(Protocol): 

100 """protocol for the ClauseAdapter/ColumnAdapter.traverse() method.""" 

101 

102 def __call__(self, obj: _CE) -> _CE: ... 

103 

104 

105class _HasDialect(Protocol): 

106 """protocol for Engine/Connection-like objects that have dialect 

107 attribute. 

108 """ 

109 

110 @property 

111 def dialect(self) -> Dialect: ... 

112 

113 

114# match column types that are not ORM entities 

115_NOT_ENTITY = TypeVar( 

116 "_NOT_ENTITY", 

117 int, 

118 str, 

119 bool, 

120 "datetime", 

121 "date", 

122 "time", 

123 "timedelta", 

124 "UUID", 

125 float, 

126 "Decimal", 

127) 

128 

129_StarOrOne = Literal["*", 1] 

130 

131_MAYBE_ENTITY = TypeVar( 

132 "_MAYBE_ENTITY", 

133 roles.ColumnsClauseRole, 

134 _StarOrOne, 

135 Type[Any], 

136 Inspectable[_HasClauseElement[Any]], 

137 _HasClauseElement[Any], 

138) 

139 

140 

141# convention: 

142# XYZArgument - something that the end user is passing to a public API method 

143# XYZElement - the internal representation that we use for the thing. 

144# the coercions system is responsible for converting from XYZArgument to 

145# XYZElement. 

146 

147_TextCoercedExpressionArgument = Union[ 

148 str, 

149 "TextClause", 

150 "ColumnElement[_T]", 

151 _HasClauseElement[_T], 

152 roles.ExpressionElementRole[_T], 

153] 

154 

155_ColumnsClauseArgument = Union[ 

156 roles.TypedColumnsClauseRole[_T], 

157 roles.ColumnsClauseRole, 

158 "SQLCoreOperations[_T]", 

159 _StarOrOne, 

160 Type[_T], 

161 Inspectable[_HasClauseElement[_T]], 

162 _HasClauseElement[_T], 

163] 

164"""open-ended SELECT columns clause argument. 

165 

166Includes column expressions, tables, ORM mapped entities, a few literal values. 

167 

168This type is used for lists of columns / entities to be returned in result 

169sets; select(...), insert().returning(...), etc. 

170 

171 

172""" 

173 

174_TypedColumnClauseArgument = Union[ 

175 roles.TypedColumnsClauseRole[_T], 

176 "SQLCoreOperations[_T]", 

177 Type[_T], 

178] 

179 

180_T0 = TypeVar("_T0", bound=Any) 

181_T1 = TypeVar("_T1", bound=Any) 

182_T2 = TypeVar("_T2", bound=Any) 

183_T3 = TypeVar("_T3", bound=Any) 

184_T4 = TypeVar("_T4", bound=Any) 

185_T5 = TypeVar("_T5", bound=Any) 

186_T6 = TypeVar("_T6", bound=Any) 

187_T7 = TypeVar("_T7", bound=Any) 

188_T8 = TypeVar("_T8", bound=Any) 

189_T9 = TypeVar("_T9", bound=Any) 

190 

191 

192_OnlyColumnArgument = Union[ 

193 "ColumnElement[_T]", 

194 _HasClauseElement[_T], 

195 roles.DMLColumnRole, 

196] 

197"""A narrow type that is looking for a ColumnClause (e.g. table column with a 

198name) or an ORM element that produces this. 

199 

200This is used for constructs that need a named column to represent a 

201position in a selectable, like TextClause().columns() or values(...). 

202 

203""" 

204 

205_ColumnExpressionArgument = Union[ 

206 "ColumnElement[_T]", 

207 _HasClauseElement[_T], 

208 "SQLCoreOperations[_T]", 

209 roles.ExpressionElementRole[_T], 

210 roles.TypedColumnsClauseRole[_T], 

211 Callable[[], "ColumnElement[_T]"], 

212 "LambdaElement", 

213] 

214"See docs in public alias ColumnExpressionArgument." 

215 

216ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T] 

217"""Narrower "column expression" argument. 

218 

219This type is used for all the other "column" kinds of expressions that 

220typically represent a single SQL column expression, not a set of columns the 

221way a table or ORM entity does. 

222 

223This includes ColumnElement, or ORM-mapped attributes that will have a 

224``__clause_element__()`` method, it also has the ExpressionElementRole 

225overall which brings in the TextClause object also. 

226 

227.. versionadded:: 2.0.13 

228 

229""" 

230 

231_ColumnExpressionOrLiteralArgument = Union[Any, _ColumnExpressionArgument[_T]] 

232 

233_ColumnExpressionOrStrLabelArgument = Union[str, _ColumnExpressionArgument[_T]] 

234 

235_ByArgument = Union[ 

236 Iterable[_ColumnExpressionOrStrLabelArgument[Any]], 

237 _ColumnExpressionOrStrLabelArgument[Any], 

238] 

239"""Used for keyword-based ``order_by`` and ``partition_by`` parameters.""" 

240 

241 

242_InfoType = Dict[Any, Any] 

243"""the .info dictionary accepted and used throughout Core /ORM""" 

244 

245_FromClauseArgument = Union[ 

246 roles.FromClauseRole, 

247 roles.TypedColumnsClauseRole[Any], 

248 Type[Any], 

249 Inspectable[_HasClauseElement[Any]], 

250 _HasClauseElement[Any], 

251] 

252"""A FROM clause, like we would send to select().select_from(). 

253 

254Also accommodates ORM entities and related constructs. 

255 

256""" 

257 

258_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole] 

259"""target for join() builds on _FromClauseArgument to include additional 

260join target roles such as those which come from the ORM. 

261 

262""" 

263 

264_OnClauseArgument = Union[_ColumnExpressionArgument[Any], roles.OnClauseRole] 

265"""target for an ON clause, includes additional roles such as those which 

266come from the ORM. 

267 

268""" 

269 

270_SelectStatementForCompoundArgument = Union[ 

271 "Select[Unpack[_Ts]]", 

272 "CompoundSelect[Unpack[_Ts]]", 

273 roles.CompoundElementRole, 

274] 

275"""SELECT statement acceptable by ``union()`` and other SQL set operations""" 

276 

277_DMLColumnArgument = Union[ 

278 str, 

279 _HasClauseElement[Any], 

280 roles.DMLColumnRole, 

281 "SQLCoreOperations[Any]", 

282] 

283"""A DML column expression. This is a "key" inside of insert().values(), 

284update().values(), and related. 

285 

286These are usually strings or SQL table columns. 

287 

288There's also edge cases like JSON expression assignment, which we would want 

289the DMLColumnRole to be able to accommodate. 

290 

291""" 

292 

293 

294_DMLKey = TypeVar("_DMLKey", bound=_DMLColumnArgument) 

295_DMLColumnKeyMapping = Mapping[_DMLKey, Any] 

296 

297 

298_DDLColumnArgument = Union[str, "Column[Any]", roles.DDLConstraintColumnRole] 

299"""DDL column. 

300 

301used for :class:`.PrimaryKeyConstraint`, :class:`.UniqueConstraint`, etc. 

302 

303""" 

304 

305_DDLColumnReferenceArgument = Union[ 

306 _DDLColumnArgument, 

307 Tuple[Optional[str], str, Optional[str]], 

308 Tuple[str, Optional[str]], 

309] 

310"""DDL column reference, as used by :class:`.ForeignKey`. 

311 

312In addition to the forms accepted by ``_DDLColumnArgument``, the target may 

313be given as a ``(schema, table_name, column_name)`` or ``(table_name, 

314column_name)`` tuple, which is the only form that can express a name that 

315itself contains a dot. 

316 

317""" 

318 

319_DMLTableArgument = Union[ 

320 "TableClause", 

321 "Join", 

322 "Alias", 

323 "CTE", 

324 Type[Any], 

325 Inspectable[_HasClauseElement[Any]], 

326 _HasClauseElement[Any], 

327] 

328 

329_PropagateAttrsType = util.immutabledict[str, Any] 

330 

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

332 

333 

334class _HasPythonType(Protocol[_T_co]): 

335 """covariant view of the python type of a TypeEngine[_T]. 

336 

337 TypeEngine is invariant in _T. When a Column is assigned to a 

338 declared type such as ``Named[int | None]``, mypy infers the Column's 

339 _T from the declared type before looking at the arguments, so an 

340 invariant ``TypeEngine[int]`` argument would be rejected. Matching 

341 against this covariant protocol instead accepts ``TypeEngine[int]`` 

342 where ``int | None`` is expected. 

343 

344 """ 

345 

346 @property 

347 def _sa_typing_python_type(self) -> _T_co: ... 

348 

349 

350_CovariantTypeEngineArgument = Union[ 

351 Type[_HasPythonType[_T]], _HasPythonType[_T] 

352] 

353 

354_EquivalentColumnMap = Dict["ColumnElement[Any]", Set["ColumnElement[Any]"]] 

355 

356_LimitOffsetType = Union[int, _ColumnExpressionArgument[int], None] 

357 

358_AutoIncrementType = Union[bool, Literal["auto", "ignore_fk"]] 

359 

360_CreateDropBind = Union["Engine", "Connection", "MockConnection"] 

361 

362if TYPE_CHECKING: 

363 

364 def is_sql_compiler(c: Compiled) -> TypeGuard[SQLCompiler]: ... 

365 

366 def is_ddl_compiler(c: Compiled) -> TypeGuard[DDLCompiler]: ... 

367 

368 def is_named_from_clause( 

369 t: FromClauseRole, 

370 ) -> TypeGuard[NamedFromClause]: ... 

371 

372 def is_column_element( 

373 c: ClauseElement, 

374 ) -> TypeGuard[ColumnElement[Any]]: ... 

375 

376 def is_keyed_column_element( 

377 c: ClauseElement, 

378 ) -> TypeGuard[KeyedColumnElement[Any]]: ... 

379 

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

381 

382 def is_from_clause(c: ClauseElement) -> TypeGuard[FromClause]: ... 

383 

384 def is_tuple_type(t: TypeEngine[Any]) -> TypeGuard[TupleType]: ... 

385 

386 def is_table_value_type( 

387 t: TypeEngine[Any], 

388 ) -> TypeGuard[TableValueType]: ... 

389 

390 def is_selectable(t: Any) -> TypeGuard[Selectable]: ... 

391 

392 def is_select_base( 

393 t: Union[Executable, ReturnsRows], 

394 ) -> TypeGuard[SelectBase]: ... 

395 

396 def is_select_statement( 

397 t: Union[Executable, ReturnsRows], 

398 ) -> TypeGuard[Select[Unpack[TupleAny]]]: ... 

399 

400 def is_table(t: FromClause) -> TypeGuard[TableClause]: ... 

401 

402 def is_subquery(t: FromClause) -> TypeGuard[Subquery]: ... 

403 

404 def is_dml(c: ClauseElement) -> TypeGuard[UpdateBase]: ... 

405 

406else: 

407 is_sql_compiler = operator.attrgetter("is_sql") 

408 is_ddl_compiler = operator.attrgetter("is_ddl") 

409 is_named_from_clause = operator.attrgetter("named_with_column") 

410 is_column_element = operator.attrgetter("_is_column_element") 

411 is_keyed_column_element = operator.attrgetter("_is_keyed_column_element") 

412 is_text_clause = operator.attrgetter("_is_text_clause") 

413 is_from_clause = operator.attrgetter("_is_from_clause") 

414 is_tuple_type = operator.attrgetter("_is_tuple_type") 

415 is_table_value_type = operator.attrgetter("_is_table_value") 

416 is_selectable = operator.attrgetter("is_selectable") 

417 is_select_base = operator.attrgetter("_is_select_base") 

418 is_select_statement = operator.attrgetter("_is_select_statement") 

419 is_table = operator.attrgetter("_is_table") 

420 is_subquery = operator.attrgetter("_is_subquery") 

421 is_dml = operator.attrgetter("is_dml") 

422 

423 

424def has_schema_attr(t: FromClauseRole) -> TypeGuard[TableClause]: 

425 return hasattr(t, "schema") 

426 

427 

428def is_quoted_name(s: str) -> TypeGuard[quoted_name]: 

429 return hasattr(s, "quote") 

430 

431 

432def is_has_clause_element(s: object) -> TypeGuard[_HasClauseElement[Any]]: 

433 return hasattr(s, "__clause_element__") 

434 

435 

436def is_insert_update(c: ClauseElement) -> TypeGuard[ValuesBase]: 

437 return c.is_dml and (c.is_insert or c.is_update) # type: ignore[attr-defined] # noqa: E501 

438 

439 

440def _no_kw() -> exc.ArgumentError: 

441 return exc.ArgumentError( 

442 "Additional keyword arguments are not accepted by this " 

443 "function/method. The presence of **kw is for pep-484 typing purposes" 

444 ) 

445 

446 

447def _unexpected_kw(methname: str, kw: Dict[str, Any]) -> NoReturn: 

448 k = list(kw)[0] 

449 raise TypeError(f"{methname} got an unexpected keyword argument '{k}'") 

450 

451 

452@overload 

453def Nullable( 

454 val: "SQLCoreOperations[_T]", 

455) -> "SQLCoreOperations[Optional[_T]]": ... 

456 

457 

458@overload 

459def Nullable( 

460 val: roles.ExpressionElementRole[_T], 

461) -> roles.ExpressionElementRole[Optional[_T]]: ... 

462 

463 

464@overload 

465def Nullable(val: Type[_T]) -> Type[Optional[_T]]: ... 

466 

467 

468def Nullable( 

469 val: _TypedColumnClauseArgument[_T], 

470) -> _TypedColumnClauseArgument[Optional[_T]]: 

471 """Types a column or ORM class as nullable. 

472 

473 This can be used in select and other contexts to express that the value of 

474 a column can be null, for example due to an outer join:: 

475 

476 stmt1 = select(A, Nullable(B)).outerjoin(A.bs) 

477 stmt2 = select(A.data, Nullable(B.data)).outerjoin(A.bs) 

478 

479 At runtime this method returns the input unchanged. 

480 

481 .. versionadded:: 2.0.20 

482 """ 

483 return val 

484 

485 

486@overload 

487def NotNullable( 

488 val: "SQLCoreOperations[Optional[_T]]", 

489) -> "SQLCoreOperations[_T]": ... 

490 

491 

492@overload 

493def NotNullable( 

494 val: roles.ExpressionElementRole[Optional[_T]], 

495) -> roles.ExpressionElementRole[_T]: ... 

496 

497 

498@overload 

499def NotNullable(val: Type[Optional[_T]]) -> Type[_T]: ... 

500 

501 

502@overload 

503def NotNullable(val: Optional[Type[_T]]) -> Type[_T]: ... 

504 

505 

506def NotNullable( 

507 val: Union[_TypedColumnClauseArgument[Optional[_T]], Optional[Type[_T]]], 

508) -> _TypedColumnClauseArgument[_T]: 

509 """Types a column or ORM class as not nullable. 

510 

511 This can be used in select and other contexts to express that the value of 

512 a column cannot be null, for example due to a where condition on a 

513 nullable column:: 

514 

515 stmt = select(NotNullable(A.value)).where(A.value.is_not(None)) 

516 

517 At runtime this method returns the input unchanged. 

518 

519 .. versionadded:: 2.0.20 

520 """ 

521 return val # type: ignore[return-value]