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

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 Type 

24from typing import TYPE_CHECKING 

25from typing import TypeAlias 

26from typing import TypeVar 

27from typing import Union 

28 

29from . import roles 

30from .. import exc 

31from .. import util 

32from ..inspection import Inspectable 

33from ..util.typing import TupleAny 

34from ..util.typing import TypeVarTuple 

35from ..util.typing import Unpack 

36 

37if TYPE_CHECKING: 

38 from datetime import date 

39 from datetime import datetime 

40 from datetime import time 

41 from datetime import timedelta 

42 from decimal import Decimal 

43 from typing import TypeGuard 

44 from uuid import UUID 

45 

46 from .base import Executable 

47 from .compiler import Compiled 

48 from .compiler import DDLCompiler 

49 from .compiler import SQLCompiler 

50 from .dml import UpdateBase 

51 from .dml import ValuesBase 

52 from .elements import ClauseElement 

53 from .elements import ColumnElement 

54 from .elements import KeyedColumnElement 

55 from .elements import quoted_name 

56 from .elements import SQLCoreOperations 

57 from .elements import TextClause 

58 from .lambdas import LambdaElement 

59 from .roles import FromClauseRole 

60 from .schema import Column 

61 from .selectable import Alias 

62 from .selectable import CompoundSelect 

63 from .selectable import CTE 

64 from .selectable import FromClause 

65 from .selectable import Join 

66 from .selectable import NamedFromClause 

67 from .selectable import ReturnsRows 

68 from .selectable import Select 

69 from .selectable import Selectable 

70 from .selectable import SelectBase 

71 from .selectable import Subquery 

72 from .selectable import TableClause 

73 from .sqltypes import TableValueType 

74 from .sqltypes import TupleType 

75 from .type_api import TypeEngine 

76 from ..engine import Connection 

77 from ..engine import Dialect 

78 from ..engine import Engine 

79 from ..engine.mock import MockConnection 

80 

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

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

83_Ts = TypeVarTuple("_Ts") 

84_Ts2 = TypeVarTuple("_Ts2") 

85 

86 

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

88 

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

90 

91 

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

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

94 

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

96 

97 

98class _CoreAdapterProto(Protocol): 

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

100 

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

102 

103 

104class _HasDialect(Protocol): 

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

106 attribute. 

107 """ 

108 

109 @property 

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

111 

112 

113# match column types that are not ORM entities 

114_NOT_ENTITY = TypeVar( 

115 "_NOT_ENTITY", 

116 int, 

117 str, 

118 bool, 

119 "datetime", 

120 "date", 

121 "time", 

122 "timedelta", 

123 "UUID", 

124 float, 

125 "Decimal", 

126) 

127 

128_StarOrOne = Literal["*", 1] 

129 

130_MAYBE_ENTITY = TypeVar( 

131 "_MAYBE_ENTITY", 

132 roles.ColumnsClauseRole, 

133 _StarOrOne, 

134 Type[Any], 

135 Inspectable[_HasClauseElement[Any]], 

136 _HasClauseElement[Any], 

137) 

138 

139 

140# convention: 

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

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

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

144# XYZElement. 

145 

146_TextCoercedExpressionArgument = Union[ 

147 str, 

148 "TextClause", 

149 "ColumnElement[_T]", 

150 _HasClauseElement[_T], 

151 roles.ExpressionElementRole[_T], 

152] 

153 

154_ColumnsClauseArgument = Union[ 

155 roles.TypedColumnsClauseRole[_T], 

156 roles.ColumnsClauseRole, 

157 "SQLCoreOperations[_T]", 

158 _StarOrOne, 

159 Type[_T], 

160 Inspectable[_HasClauseElement[_T]], 

161 _HasClauseElement[_T], 

162] 

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

164 

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

166 

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

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

169 

170 

171""" 

172 

173_TypedColumnClauseArgument = Union[ 

174 roles.TypedColumnsClauseRole[_T], 

175 "SQLCoreOperations[_T]", 

176 Type[_T], 

177] 

178 

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

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

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

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

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

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

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

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

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

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

189 

190 

191_OnlyColumnArgument = Union[ 

192 "ColumnElement[_T]", 

193 _HasClauseElement[_T], 

194 roles.DMLColumnRole, 

195] 

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

197name) or an ORM element that produces this. 

198 

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

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

201 

202""" 

203 

204_ColumnExpressionArgument = Union[ 

205 "ColumnElement[_T]", 

206 _HasClauseElement[_T], 

207 "SQLCoreOperations[_T]", 

208 roles.ExpressionElementRole[_T], 

209 roles.TypedColumnsClauseRole[_T], 

210 Callable[[], "ColumnElement[_T]"], 

211 "LambdaElement", 

212] 

213"See docs in public alias ColumnExpressionArgument." 

214 

215ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T] 

216"""Narrower "column expression" argument. 

217 

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

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

220way a table or ORM entity does. 

221 

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

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

224overall which brings in the TextClause object also. 

225 

226.. versionadded:: 2.0.13 

227 

228""" 

229 

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

231 

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

233 

234_ByArgument = Union[ 

235 Iterable[_ColumnExpressionOrStrLabelArgument[Any]], 

236 _ColumnExpressionOrStrLabelArgument[Any], 

237] 

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

239 

240 

241_InfoType = Dict[Any, Any] 

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

243 

244_FromClauseArgument = Union[ 

245 roles.FromClauseRole, 

246 roles.TypedColumnsClauseRole[Any], 

247 Type[Any], 

248 Inspectable[_HasClauseElement[Any]], 

249 _HasClauseElement[Any], 

250] 

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

252 

253Also accommodates ORM entities and related constructs. 

254 

255""" 

256 

257_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole] 

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

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

260 

261""" 

262 

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

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

265come from the ORM. 

266 

267""" 

268 

269_SelectStatementForCompoundArgument = Union[ 

270 "Select[Unpack[_Ts]]", 

271 "CompoundSelect[Unpack[_Ts]]", 

272 roles.CompoundElementRole, 

273] 

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

275 

276_DMLColumnArgument = Union[ 

277 str, 

278 _HasClauseElement[Any], 

279 roles.DMLColumnRole, 

280 "SQLCoreOperations[Any]", 

281] 

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

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

284 

285These are usually strings or SQL table columns. 

286 

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

288the DMLColumnRole to be able to accommodate. 

289 

290""" 

291 

292 

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

294_DMLColumnKeyMapping = Mapping[_DMLKey, Any] 

295 

296 

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

298"""DDL column. 

299 

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

301 

302""" 

303 

304_DDLColumnReferenceArgument = _DDLColumnArgument 

305 

306_DMLTableArgument = Union[ 

307 "TableClause", 

308 "Join", 

309 "Alias", 

310 "CTE", 

311 Type[Any], 

312 Inspectable[_HasClauseElement[Any]], 

313 _HasClauseElement[Any], 

314] 

315 

316_PropagateAttrsType = util.immutabledict[str, Any] 

317 

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

319 

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

321 

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

323 

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

325 

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

327 

328if TYPE_CHECKING: 

329 

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

331 

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

333 

334 def is_named_from_clause( 

335 t: FromClauseRole, 

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

337 

338 def is_column_element( 

339 c: ClauseElement, 

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

341 

342 def is_keyed_column_element( 

343 c: ClauseElement, 

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

345 

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

347 

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

349 

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

351 

352 def is_table_value_type( 

353 t: TypeEngine[Any], 

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

355 

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

357 

358 def is_select_base( 

359 t: Union[Executable, ReturnsRows], 

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

361 

362 def is_select_statement( 

363 t: Union[Executable, ReturnsRows], 

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

365 

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

367 

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

369 

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

371 

372else: 

373 is_sql_compiler = operator.attrgetter("is_sql") 

374 is_ddl_compiler = operator.attrgetter("is_ddl") 

375 is_named_from_clause = operator.attrgetter("named_with_column") 

376 is_column_element = operator.attrgetter("_is_column_element") 

377 is_keyed_column_element = operator.attrgetter("_is_keyed_column_element") 

378 is_text_clause = operator.attrgetter("_is_text_clause") 

379 is_from_clause = operator.attrgetter("_is_from_clause") 

380 is_tuple_type = operator.attrgetter("_is_tuple_type") 

381 is_table_value_type = operator.attrgetter("_is_table_value") 

382 is_selectable = operator.attrgetter("is_selectable") 

383 is_select_base = operator.attrgetter("_is_select_base") 

384 is_select_statement = operator.attrgetter("_is_select_statement") 

385 is_table = operator.attrgetter("_is_table") 

386 is_subquery = operator.attrgetter("_is_subquery") 

387 is_dml = operator.attrgetter("is_dml") 

388 

389 

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

391 return hasattr(t, "schema") 

392 

393 

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

395 return hasattr(s, "quote") 

396 

397 

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

399 return hasattr(s, "__clause_element__") 

400 

401 

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

403 return c.is_dml and (c.is_insert or c.is_update) # type: ignore 

404 

405 

406def _no_kw() -> exc.ArgumentError: 

407 return exc.ArgumentError( 

408 "Additional keyword arguments are not accepted by this " 

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

410 ) 

411 

412 

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

414 k = list(kw)[0] 

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

416 

417 

418@overload 

419def Nullable( 

420 val: "SQLCoreOperations[_T]", 

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

422 

423 

424@overload 

425def Nullable( 

426 val: roles.ExpressionElementRole[_T], 

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

428 

429 

430@overload 

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

432 

433 

434def Nullable( 

435 val: _TypedColumnClauseArgument[_T], 

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

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

438 

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

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

441 

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

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

444 

445 At runtime this method returns the input unchanged. 

446 

447 .. versionadded:: 2.0.20 

448 """ 

449 return val 

450 

451 

452@overload 

453def NotNullable( 

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

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

456 

457 

458@overload 

459def NotNullable( 

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

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

462 

463 

464@overload 

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

466 

467 

468@overload 

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

470 

471 

472def NotNullable( 

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

474) -> _TypedColumnClauseArgument[_T]: 

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

476 

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

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

479 nullable column:: 

480 

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

482 

483 At runtime this method returns the input unchanged. 

484 

485 .. versionadded:: 2.0.20 

486 """ 

487 return val # type: ignore