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 

85 

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

87 

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

89 

90 

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

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

93 

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

95 

96 

97class _CoreAdapterProto(Protocol): 

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

99 

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

101 

102 

103class _HasDialect(Protocol): 

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

105 attribute. 

106 """ 

107 

108 @property 

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

110 

111 

112# match column types that are not ORM entities 

113_NOT_ENTITY = TypeVar( 

114 "_NOT_ENTITY", 

115 int, 

116 str, 

117 bool, 

118 "datetime", 

119 "date", 

120 "time", 

121 "timedelta", 

122 "UUID", 

123 float, 

124 "Decimal", 

125) 

126 

127_StarOrOne = Literal["*", 1] 

128 

129_MAYBE_ENTITY = TypeVar( 

130 "_MAYBE_ENTITY", 

131 roles.ColumnsClauseRole, 

132 _StarOrOne, 

133 Type[Any], 

134 Inspectable[_HasClauseElement[Any]], 

135 _HasClauseElement[Any], 

136) 

137 

138 

139# convention: 

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

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

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

143# XYZElement. 

144 

145_TextCoercedExpressionArgument = Union[ 

146 str, 

147 "TextClause", 

148 "ColumnElement[_T]", 

149 _HasClauseElement[_T], 

150 roles.ExpressionElementRole[_T], 

151] 

152 

153_ColumnsClauseArgument = Union[ 

154 roles.TypedColumnsClauseRole[_T], 

155 roles.ColumnsClauseRole, 

156 "SQLCoreOperations[_T]", 

157 _StarOrOne, 

158 Type[_T], 

159 Inspectable[_HasClauseElement[_T]], 

160 _HasClauseElement[_T], 

161] 

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

163 

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

165 

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

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

168 

169 

170""" 

171 

172_TypedColumnClauseArgument = Union[ 

173 roles.TypedColumnsClauseRole[_T], 

174 "SQLCoreOperations[_T]", 

175 Type[_T], 

176] 

177 

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

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

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

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

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

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

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

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

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

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

188 

189 

190_OnlyColumnArgument = Union[ 

191 "ColumnElement[_T]", 

192 _HasClauseElement[_T], 

193 roles.DMLColumnRole, 

194] 

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

196name) or an ORM element that produces this. 

197 

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

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

200 

201""" 

202 

203_ColumnExpressionArgument = Union[ 

204 "ColumnElement[_T]", 

205 _HasClauseElement[_T], 

206 "SQLCoreOperations[_T]", 

207 roles.ExpressionElementRole[_T], 

208 roles.TypedColumnsClauseRole[_T], 

209 Callable[[], "ColumnElement[_T]"], 

210 "LambdaElement", 

211] 

212"See docs in public alias ColumnExpressionArgument." 

213 

214ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T] 

215"""Narrower "column expression" argument. 

216 

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

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

219way a table or ORM entity does. 

220 

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

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

223overall which brings in the TextClause object also. 

224 

225.. versionadded:: 2.0.13 

226 

227""" 

228 

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

230 

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

232 

233_ByArgument = Union[ 

234 Iterable[_ColumnExpressionOrStrLabelArgument[Any]], 

235 _ColumnExpressionOrStrLabelArgument[Any], 

236] 

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

238 

239 

240_InfoType = Dict[Any, Any] 

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

242 

243_FromClauseArgument = Union[ 

244 roles.FromClauseRole, 

245 roles.TypedColumnsClauseRole[Any], 

246 Type[Any], 

247 Inspectable[_HasClauseElement[Any]], 

248 _HasClauseElement[Any], 

249] 

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

251 

252Also accommodates ORM entities and related constructs. 

253 

254""" 

255 

256_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole] 

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

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

259 

260""" 

261 

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

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

264come from the ORM. 

265 

266""" 

267 

268_SelectStatementForCompoundArgument = Union[ 

269 "Select[Unpack[_Ts]]", 

270 "CompoundSelect[Unpack[_Ts]]", 

271 roles.CompoundElementRole, 

272] 

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

274 

275_DMLColumnArgument = Union[ 

276 str, 

277 _HasClauseElement[Any], 

278 roles.DMLColumnRole, 

279 "SQLCoreOperations[Any]", 

280] 

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

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

283 

284These are usually strings or SQL table columns. 

285 

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

287the DMLColumnRole to be able to accommodate. 

288 

289""" 

290 

291 

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

293_DMLColumnKeyMapping = Mapping[_DMLKey, Any] 

294 

295 

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

297"""DDL column. 

298 

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

300 

301""" 

302 

303_DDLColumnReferenceArgument = _DDLColumnArgument 

304 

305_DMLTableArgument = Union[ 

306 "TableClause", 

307 "Join", 

308 "Alias", 

309 "CTE", 

310 Type[Any], 

311 Inspectable[_HasClauseElement[Any]], 

312 _HasClauseElement[Any], 

313] 

314 

315_PropagateAttrsType = util.immutabledict[str, Any] 

316 

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

318 

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

320 

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

322 

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

324 

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

326 

327if TYPE_CHECKING: 

328 

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

330 

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

332 

333 def is_named_from_clause( 

334 t: FromClauseRole, 

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

336 

337 def is_column_element( 

338 c: ClauseElement, 

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

340 

341 def is_keyed_column_element( 

342 c: ClauseElement, 

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

344 

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

346 

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

348 

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

350 

351 def is_table_value_type( 

352 t: TypeEngine[Any], 

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

354 

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

356 

357 def is_select_base( 

358 t: Union[Executable, ReturnsRows], 

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

360 

361 def is_select_statement( 

362 t: Union[Executable, ReturnsRows], 

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

364 

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

366 

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

368 

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

370 

371else: 

372 is_sql_compiler = operator.attrgetter("is_sql") 

373 is_ddl_compiler = operator.attrgetter("is_ddl") 

374 is_named_from_clause = operator.attrgetter("named_with_column") 

375 is_column_element = operator.attrgetter("_is_column_element") 

376 is_keyed_column_element = operator.attrgetter("_is_keyed_column_element") 

377 is_text_clause = operator.attrgetter("_is_text_clause") 

378 is_from_clause = operator.attrgetter("_is_from_clause") 

379 is_tuple_type = operator.attrgetter("_is_tuple_type") 

380 is_table_value_type = operator.attrgetter("_is_table_value") 

381 is_selectable = operator.attrgetter("is_selectable") 

382 is_select_base = operator.attrgetter("_is_select_base") 

383 is_select_statement = operator.attrgetter("_is_select_statement") 

384 is_table = operator.attrgetter("_is_table") 

385 is_subquery = operator.attrgetter("_is_subquery") 

386 is_dml = operator.attrgetter("is_dml") 

387 

388 

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

390 return hasattr(t, "schema") 

391 

392 

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

394 return hasattr(s, "quote") 

395 

396 

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

398 return hasattr(s, "__clause_element__") 

399 

400 

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

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

403 

404 

405def _no_kw() -> exc.ArgumentError: 

406 return exc.ArgumentError( 

407 "Additional keyword arguments are not accepted by this " 

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

409 ) 

410 

411 

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

413 k = list(kw)[0] 

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

415 

416 

417@overload 

418def Nullable( 

419 val: "SQLCoreOperations[_T]", 

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

421 

422 

423@overload 

424def Nullable( 

425 val: roles.ExpressionElementRole[_T], 

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

427 

428 

429@overload 

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

431 

432 

433def Nullable( 

434 val: _TypedColumnClauseArgument[_T], 

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

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

437 

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

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

440 

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

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

443 

444 At runtime this method returns the input unchanged. 

445 

446 .. versionadded:: 2.0.20 

447 """ 

448 return val 

449 

450 

451@overload 

452def NotNullable( 

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

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

455 

456 

457@overload 

458def NotNullable( 

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

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

461 

462 

463@overload 

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

465 

466 

467@overload 

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

469 

470 

471def NotNullable( 

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

473) -> _TypedColumnClauseArgument[_T]: 

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

475 

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

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

478 nullable column:: 

479 

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

481 

482 At runtime this method returns the input unchanged. 

483 

484 .. versionadded:: 2.0.20 

485 """ 

486 return val # type: ignore