Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.9/dist-packages/sqlalchemy/sql/_typing.py: 14%

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 Set 

21from typing import Tuple 

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 Protocol 

33from ..util.typing import TypeAlias 

34 

35if TYPE_CHECKING: 

36 from datetime import date 

37 from datetime import datetime 

38 from datetime import time 

39 from datetime import timedelta 

40 from decimal import Decimal 

41 from uuid import UUID 

42 

43 from .base import Executable 

44 from .compiler import Compiled 

45 from .compiler import DDLCompiler 

46 from .compiler import SQLCompiler 

47 from .dml import UpdateBase 

48 from .dml import ValuesBase 

49 from .elements import ClauseElement 

50 from .elements import ColumnElement 

51 from .elements import KeyedColumnElement 

52 from .elements import quoted_name 

53 from .elements import SQLCoreOperations 

54 from .elements import TextClause 

55 from .lambdas import LambdaElement 

56 from .roles import FromClauseRole 

57 from .schema import Column 

58 from .selectable import Alias 

59 from .selectable import CTE 

60 from .selectable import FromClause 

61 from .selectable import Join 

62 from .selectable import NamedFromClause 

63 from .selectable import ReturnsRows 

64 from .selectable import Select 

65 from .selectable import Selectable 

66 from .selectable import SelectBase 

67 from .selectable import Subquery 

68 from .selectable import TableClause 

69 from .sqltypes import TableValueType 

70 from .sqltypes import TupleType 

71 from .type_api import TypeEngine 

72 from ..engine import Dialect 

73 from ..util.typing import TypeGuard 

74 

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

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

77 

78 

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

80 

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

82 

83 

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

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

86 

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

88 

89 

90class _CoreAdapterProto(Protocol): 

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

92 

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

94 

95 

96class _HasDialect(Protocol): 

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

98 attribute. 

99 """ 

100 

101 @property 

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

103 

104 

105# match column types that are not ORM entities 

106_NOT_ENTITY = TypeVar( 

107 "_NOT_ENTITY", 

108 int, 

109 str, 

110 bool, 

111 "datetime", 

112 "date", 

113 "time", 

114 "timedelta", 

115 "UUID", 

116 float, 

117 "Decimal", 

118) 

119 

120_StarOrOne = Literal["*", 1] 

121 

122_MAYBE_ENTITY = TypeVar( 

123 "_MAYBE_ENTITY", 

124 roles.ColumnsClauseRole, 

125 _StarOrOne, 

126 Type[Any], 

127 Inspectable[_HasClauseElement[Any]], 

128 _HasClauseElement[Any], 

129) 

130 

131 

132# convention: 

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

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

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

136# XYZElement. 

137 

138_TextCoercedExpressionArgument = Union[ 

139 str, 

140 "TextClause", 

141 "ColumnElement[_T]", 

142 _HasClauseElement[_T], 

143 roles.ExpressionElementRole[_T], 

144] 

145 

146_ColumnsClauseArgument = Union[ 

147 roles.TypedColumnsClauseRole[_T], 

148 roles.ColumnsClauseRole, 

149 "SQLCoreOperations[_T]", 

150 _StarOrOne, 

151 Type[_T], 

152 Inspectable[_HasClauseElement[_T]], 

153 _HasClauseElement[_T], 

154] 

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

156 

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

158 

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

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

161 

162 

163""" 

164 

165_TypedColumnClauseArgument = Union[ 

166 roles.TypedColumnsClauseRole[_T], 

167 "SQLCoreOperations[_T]", 

168 Type[_T], 

169] 

170 

171_TP = TypeVar("_TP", bound=Tuple[Any, ...]) 

172 

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

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

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

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

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

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

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

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

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

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

183 

184 

185_ColumnExpressionArgument = Union[ 

186 "ColumnElement[_T]", 

187 _HasClauseElement[_T], 

188 "SQLCoreOperations[_T]", 

189 roles.ExpressionElementRole[_T], 

190 roles.TypedColumnsClauseRole[_T], 

191 Callable[[], "ColumnElement[_T]"], 

192 "LambdaElement", 

193] 

194"See docs in public alias ColumnExpressionArgument." 

195 

196ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T] 

197"""Narrower "column expression" argument. 

198 

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

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

201way a table or ORM entity does. 

202 

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

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

205overall which brings in the TextClause object also. 

206 

207.. versionadded:: 2.0.13 

208 

209""" 

210 

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

212 

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

214 

215_ByArgument = Union[ 

216 Iterable[_ColumnExpressionOrStrLabelArgument[Any]], 

217 _ColumnExpressionOrStrLabelArgument[Any], 

218] 

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

220 

221 

222_InfoType = Dict[Any, Any] 

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

224 

225_FromClauseArgument = Union[ 

226 roles.FromClauseRole, 

227 Type[Any], 

228 Inspectable[_HasClauseElement[Any]], 

229 _HasClauseElement[Any], 

230] 

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

232 

233Also accommodates ORM entities and related constructs. 

234 

235""" 

236 

237_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole] 

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

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

240 

241""" 

242 

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

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

245come from the ORM. 

246 

247""" 

248 

249_SelectStatementForCompoundArgument = Union[ 

250 "SelectBase", roles.CompoundElementRole 

251] 

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

253 

254_DMLColumnArgument = Union[ 

255 str, 

256 _HasClauseElement[Any], 

257 roles.DMLColumnRole, 

258 "SQLCoreOperations[Any]", 

259] 

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

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

262 

263These are usually strings or SQL table columns. 

264 

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

266the DMLColumnRole to be able to accommodate. 

267 

268""" 

269 

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

271_DMLColumnKeyMapping = Mapping[_DMLKey, Any] 

272 

273 

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

275"""DDL column. 

276 

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

278 

279""" 

280 

281_DMLTableArgument = Union[ 

282 "TableClause", 

283 "Join", 

284 "Alias", 

285 "CTE", 

286 Type[Any], 

287 Inspectable[_HasClauseElement[Any]], 

288 _HasClauseElement[Any], 

289] 

290 

291_PropagateAttrsType = util.immutabledict[str, Any] 

292 

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

294 

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

296 

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

298 

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

300 

301if TYPE_CHECKING: 

302 

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

304 

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

306 

307 def is_named_from_clause( 

308 t: FromClauseRole, 

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

310 

311 def is_column_element( 

312 c: ClauseElement, 

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

314 

315 def is_keyed_column_element( 

316 c: ClauseElement, 

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

318 

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

320 

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

322 

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

324 

325 def is_table_value_type( 

326 t: TypeEngine[Any], 

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

328 

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

330 

331 def is_select_base( 

332 t: Union[Executable, ReturnsRows] 

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

334 

335 def is_select_statement( 

336 t: Union[Executable, ReturnsRows] 

337 ) -> TypeGuard[Select[Any]]: ... 

338 

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

340 

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

342 

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

344 

345else: 

346 is_sql_compiler = operator.attrgetter("is_sql") 

347 is_ddl_compiler = operator.attrgetter("is_ddl") 

348 is_named_from_clause = operator.attrgetter("named_with_column") 

349 is_column_element = operator.attrgetter("_is_column_element") 

350 is_keyed_column_element = operator.attrgetter("_is_keyed_column_element") 

351 is_text_clause = operator.attrgetter("_is_text_clause") 

352 is_from_clause = operator.attrgetter("_is_from_clause") 

353 is_tuple_type = operator.attrgetter("_is_tuple_type") 

354 is_table_value_type = operator.attrgetter("_is_table_value") 

355 is_selectable = operator.attrgetter("is_selectable") 

356 is_select_base = operator.attrgetter("_is_select_base") 

357 is_select_statement = operator.attrgetter("_is_select_statement") 

358 is_table = operator.attrgetter("_is_table") 

359 is_subquery = operator.attrgetter("_is_subquery") 

360 is_dml = operator.attrgetter("is_dml") 

361 

362 

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

364 return hasattr(t, "schema") 

365 

366 

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

368 return hasattr(s, "quote") 

369 

370 

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

372 return hasattr(s, "__clause_element__") 

373 

374 

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

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

377 

378 

379def _no_kw() -> exc.ArgumentError: 

380 return exc.ArgumentError( 

381 "Additional keyword arguments are not accepted by this " 

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

383 ) 

384 

385 

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

387 k = list(kw)[0] 

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

389 

390 

391@overload 

392def Nullable( 

393 val: "SQLCoreOperations[_T]", 

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

395 

396 

397@overload 

398def Nullable( 

399 val: roles.ExpressionElementRole[_T], 

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

401 

402 

403@overload 

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

405 

406 

407def Nullable( 

408 val: _TypedColumnClauseArgument[_T], 

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

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

411 

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

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

414 

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

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

417 

418 At runtime this method returns the input unchanged. 

419 

420 .. versionadded:: 2.0.20 

421 """ 

422 return val 

423 

424 

425@overload 

426def NotNullable( 

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

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

429 

430 

431@overload 

432def NotNullable( 

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

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

435 

436 

437@overload 

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

439 

440 

441@overload 

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

443 

444 

445def NotNullable( 

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

447) -> _TypedColumnClauseArgument[_T]: 

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

449 

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

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

452 nullable column:: 

453 

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

455 

456 At runtime this method returns the input unchanged. 

457 

458 .. versionadded:: 2.0.20 

459 """ 

460 return val # type: ignore