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-2025 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 Protocol 

21from typing import Set 

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 TupleAny 

33from ..util.typing import TypeAlias 

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 uuid import UUID 

44 

45 from .base import Executable 

46 from .compiler import Compiled 

47 from .compiler import DDLCompiler 

48 from .compiler import SQLCompiler 

49 from .dml import UpdateBase 

50 from .dml import ValuesBase 

51 from .elements import ClauseElement 

52 from .elements import ColumnElement 

53 from .elements import KeyedColumnElement 

54 from .elements import quoted_name 

55 from .elements import SQLCoreOperations 

56 from .elements import TextClause 

57 from .lambdas import LambdaElement 

58 from .roles import FromClauseRole 

59 from .schema import Column 

60 from .selectable import Alias 

61 from .selectable import CompoundSelect 

62 from .selectable import CTE 

63 from .selectable import FromClause 

64 from .selectable import Join 

65 from .selectable import NamedFromClause 

66 from .selectable import ReturnsRows 

67 from .selectable import Select 

68 from .selectable import Selectable 

69 from .selectable import SelectBase 

70 from .selectable import Subquery 

71 from .selectable import TableClause 

72 from .sqltypes import TableValueType 

73 from .sqltypes import TupleType 

74 from .type_api import TypeEngine 

75 from ..engine import Connection 

76 from ..engine import Dialect 

77 from ..engine import Engine 

78 from ..engine.mock import MockConnection 

79 from ..util.typing import TypeGuard 

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_ColumnExpressionArgument = Union[ 

191 "ColumnElement[_T]", 

192 _HasClauseElement[_T], 

193 "SQLCoreOperations[_T]", 

194 roles.ExpressionElementRole[_T], 

195 roles.TypedColumnsClauseRole[_T], 

196 Callable[[], "ColumnElement[_T]"], 

197 "LambdaElement", 

198] 

199"See docs in public alias ColumnExpressionArgument." 

200 

201ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T] 

202"""Narrower "column expression" argument. 

203 

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

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

206way a table or ORM entity does. 

207 

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

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

210overall which brings in the TextClause object also. 

211 

212.. versionadded:: 2.0.13 

213 

214""" 

215 

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

217 

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

219 

220_ByArgument = Union[ 

221 Iterable[_ColumnExpressionOrStrLabelArgument[Any]], 

222 _ColumnExpressionOrStrLabelArgument[Any], 

223] 

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

225 

226 

227_InfoType = Dict[Any, Any] 

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

229 

230_FromClauseArgument = Union[ 

231 roles.FromClauseRole, 

232 Type[Any], 

233 Inspectable[_HasClauseElement[Any]], 

234 _HasClauseElement[Any], 

235] 

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

237 

238Also accommodates ORM entities and related constructs. 

239 

240""" 

241 

242_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole] 

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

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

245 

246""" 

247 

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

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

250come from the ORM. 

251 

252""" 

253 

254_SelectStatementForCompoundArgument = Union[ 

255 "Select[Unpack[_Ts]]", 

256 "CompoundSelect[Unpack[_Ts]]", 

257 roles.CompoundElementRole, 

258] 

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

260 

261_DMLColumnArgument = Union[ 

262 str, 

263 _HasClauseElement[Any], 

264 roles.DMLColumnRole, 

265 "SQLCoreOperations[Any]", 

266] 

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

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

269 

270These are usually strings or SQL table columns. 

271 

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

273the DMLColumnRole to be able to accommodate. 

274 

275""" 

276 

277_DMLOnlyColumnArgument = Union[ 

278 _HasClauseElement[_T], 

279 roles.DMLColumnRole, 

280 "SQLCoreOperations[_T]", 

281] 

282 

283 

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

285_DMLColumnKeyMapping = Mapping[_DMLKey, Any] 

286 

287 

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

289"""DDL column. 

290 

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

292 

293""" 

294 

295_DDLColumnReferenceArgument = _DDLColumnArgument 

296 

297_DMLTableArgument = Union[ 

298 "TableClause", 

299 "Join", 

300 "Alias", 

301 "CTE", 

302 Type[Any], 

303 Inspectable[_HasClauseElement[Any]], 

304 _HasClauseElement[Any], 

305] 

306 

307_PropagateAttrsType = util.immutabledict[str, Any] 

308 

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

310 

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

312 

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

314 

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

316 

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

318 

319if TYPE_CHECKING: 

320 

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

322 

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

324 

325 def is_named_from_clause( 

326 t: FromClauseRole, 

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

328 

329 def is_column_element( 

330 c: ClauseElement, 

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

332 

333 def is_keyed_column_element( 

334 c: ClauseElement, 

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

336 

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

338 

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

340 

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

342 

343 def is_table_value_type( 

344 t: TypeEngine[Any], 

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

346 

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

348 

349 def is_select_base( 

350 t: Union[Executable, ReturnsRows], 

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

352 

353 def is_select_statement( 

354 t: Union[Executable, ReturnsRows], 

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

356 

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

358 

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

360 

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

362 

363else: 

364 is_sql_compiler = operator.attrgetter("is_sql") 

365 is_ddl_compiler = operator.attrgetter("is_ddl") 

366 is_named_from_clause = operator.attrgetter("named_with_column") 

367 is_column_element = operator.attrgetter("_is_column_element") 

368 is_keyed_column_element = operator.attrgetter("_is_keyed_column_element") 

369 is_text_clause = operator.attrgetter("_is_text_clause") 

370 is_from_clause = operator.attrgetter("_is_from_clause") 

371 is_tuple_type = operator.attrgetter("_is_tuple_type") 

372 is_table_value_type = operator.attrgetter("_is_table_value") 

373 is_selectable = operator.attrgetter("is_selectable") 

374 is_select_base = operator.attrgetter("_is_select_base") 

375 is_select_statement = operator.attrgetter("_is_select_statement") 

376 is_table = operator.attrgetter("_is_table") 

377 is_subquery = operator.attrgetter("_is_subquery") 

378 is_dml = operator.attrgetter("is_dml") 

379 

380 

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

382 return hasattr(t, "schema") 

383 

384 

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

386 return hasattr(s, "quote") 

387 

388 

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

390 return hasattr(s, "__clause_element__") 

391 

392 

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

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

395 

396 

397def _no_kw() -> exc.ArgumentError: 

398 return exc.ArgumentError( 

399 "Additional keyword arguments are not accepted by this " 

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

401 ) 

402 

403 

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

405 k = list(kw)[0] 

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

407 

408 

409@overload 

410def Nullable( 

411 val: "SQLCoreOperations[_T]", 

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

413 

414 

415@overload 

416def Nullable( 

417 val: roles.ExpressionElementRole[_T], 

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

419 

420 

421@overload 

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

423 

424 

425def Nullable( 

426 val: _TypedColumnClauseArgument[_T], 

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

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

429 

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

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

432 

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

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

435 

436 At runtime this method returns the input unchanged. 

437 

438 .. versionadded:: 2.0.20 

439 """ 

440 return val 

441 

442 

443@overload 

444def NotNullable( 

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

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

447 

448 

449@overload 

450def NotNullable( 

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

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

453 

454 

455@overload 

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

457 

458 

459@overload 

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

461 

462 

463def NotNullable( 

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

465) -> _TypedColumnClauseArgument[_T]: 

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

467 

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

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

470 nullable column:: 

471 

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

473 

474 At runtime this method returns the input unchanged. 

475 

476 .. versionadded:: 2.0.20 

477 """ 

478 return val # type: ignore