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

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 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 CompoundSelect 

60 from .selectable import CTE 

61 from .selectable import FromClause 

62 from .selectable import Join 

63 from .selectable import NamedFromClause 

64 from .selectable import ReturnsRows 

65 from .selectable import Select 

66 from .selectable import Selectable 

67 from .selectable import SelectBase 

68 from .selectable import Subquery 

69 from .selectable import TableClause 

70 from .sqltypes import TableValueType 

71 from .sqltypes import TupleType 

72 from .type_api import TypeEngine 

73 from ..engine import Connection 

74 from ..engine import Dialect 

75 from ..engine import Engine 

76 from ..engine.mock import MockConnection 

77 from ..util.typing import TypeGuard 

78 

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

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

81 

82 

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

84 

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

86 

87 

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

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

90 

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

92 

93 

94class _CoreAdapterProto(Protocol): 

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

96 

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

98 

99 

100class _HasDialect(Protocol): 

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

102 attribute. 

103 """ 

104 

105 @property 

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

107 

108 

109# match column types that are not ORM entities 

110_NOT_ENTITY = TypeVar( 

111 "_NOT_ENTITY", 

112 int, 

113 str, 

114 bool, 

115 "datetime", 

116 "date", 

117 "time", 

118 "timedelta", 

119 "UUID", 

120 float, 

121 "Decimal", 

122) 

123 

124_StarOrOne = Literal["*", 1] 

125 

126_MAYBE_ENTITY = TypeVar( 

127 "_MAYBE_ENTITY", 

128 roles.ColumnsClauseRole, 

129 _StarOrOne, 

130 Type[Any], 

131 Inspectable[_HasClauseElement[Any]], 

132 _HasClauseElement[Any], 

133) 

134 

135 

136# convention: 

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

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

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

140# XYZElement. 

141 

142_TextCoercedExpressionArgument = Union[ 

143 str, 

144 "TextClause", 

145 "ColumnElement[_T]", 

146 _HasClauseElement[_T], 

147 roles.ExpressionElementRole[_T], 

148] 

149 

150_ColumnsClauseArgument = Union[ 

151 roles.TypedColumnsClauseRole[_T], 

152 roles.ColumnsClauseRole, 

153 "SQLCoreOperations[_T]", 

154 _StarOrOne, 

155 Type[_T], 

156 Inspectable[_HasClauseElement[_T]], 

157 _HasClauseElement[_T], 

158] 

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

160 

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

162 

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

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

165 

166 

167""" 

168 

169_TypedColumnClauseArgument = Union[ 

170 roles.TypedColumnsClauseRole[_T], 

171 "SQLCoreOperations[_T]", 

172 Type[_T], 

173] 

174 

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

176 

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

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

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

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

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

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

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

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

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

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

187 

188 

189_OnlyColumnArgument = Union[ 

190 "ColumnElement[_T]", 

191 _HasClauseElement[_T], 

192 roles.DMLColumnRole, 

193] 

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

195name) or an ORM element that produces this. 

196 

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

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

199 

200""" 

201 

202_ColumnExpressionArgument = Union[ 

203 "ColumnElement[_T]", 

204 _HasClauseElement[_T], 

205 "SQLCoreOperations[_T]", 

206 roles.ExpressionElementRole[_T], 

207 roles.TypedColumnsClauseRole[_T], 

208 Callable[[], "ColumnElement[_T]"], 

209 "LambdaElement", 

210] 

211"See docs in public alias ColumnExpressionArgument." 

212 

213ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T] 

214"""Narrower "column expression" argument. 

215 

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

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

218way a table or ORM entity does. 

219 

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

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

222overall which brings in the TextClause object also. 

223 

224.. versionadded:: 2.0.13 

225 

226""" 

227 

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

229 

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

231 

232_ByArgument = Union[ 

233 Iterable[_ColumnExpressionOrStrLabelArgument[Any]], 

234 _ColumnExpressionOrStrLabelArgument[Any], 

235] 

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

237 

238 

239_InfoType = Dict[Any, Any] 

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

241 

242_FromClauseArgument = Union[ 

243 roles.FromClauseRole, 

244 roles.TypedColumnsClauseRole[Any], 

245 Type[Any], 

246 Inspectable[_HasClauseElement[Any]], 

247 _HasClauseElement[Any], 

248] 

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

250 

251Also accommodates ORM entities and related constructs. 

252 

253""" 

254 

255_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole] 

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

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

258 

259""" 

260 

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

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

263come from the ORM. 

264 

265""" 

266 

267_SelectStatementForCompoundArgument = Union[ 

268 "Select[_TP]", 

269 "CompoundSelect[_TP]", 

270 roles.CompoundElementRole, 

271] 

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

273 

274_DMLColumnArgument = Union[ 

275 str, 

276 _HasClauseElement[Any], 

277 roles.DMLColumnRole, 

278 "SQLCoreOperations[Any]", 

279] 

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

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

282 

283These are usually strings or SQL table columns. 

284 

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

286the DMLColumnRole to be able to accommodate. 

287 

288""" 

289 

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

291_DMLColumnKeyMapping = Mapping[_DMLKey, Any] 

292 

293 

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

295"""DDL column. 

296 

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

298 

299""" 

300 

301_DMLTableArgument = Union[ 

302 "TableClause", 

303 "Join", 

304 "Alias", 

305 "CTE", 

306 Type[Any], 

307 Inspectable[_HasClauseElement[Any]], 

308 _HasClauseElement[Any], 

309] 

310 

311_PropagateAttrsType = util.immutabledict[str, Any] 

312 

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

314 

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

316 

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

318 

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

320 

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

322 

323if TYPE_CHECKING: 

324 

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

326 

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

328 

329 def is_named_from_clause( 

330 t: FromClauseRole, 

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

332 

333 def is_column_element( 

334 c: ClauseElement, 

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

336 

337 def is_keyed_column_element( 

338 c: ClauseElement, 

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

340 

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

342 

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

344 

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

346 

347 def is_table_value_type( 

348 t: TypeEngine[Any], 

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

350 

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

352 

353 def is_select_base( 

354 t: Union[Executable, ReturnsRows], 

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

356 

357 def is_select_statement( 

358 t: Union[Executable, ReturnsRows], 

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

360 

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

362 

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

364 

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

366 

367else: 

368 is_sql_compiler = operator.attrgetter("is_sql") 

369 is_ddl_compiler = operator.attrgetter("is_ddl") 

370 is_named_from_clause = operator.attrgetter("named_with_column") 

371 is_column_element = operator.attrgetter("_is_column_element") 

372 is_keyed_column_element = operator.attrgetter("_is_keyed_column_element") 

373 is_text_clause = operator.attrgetter("_is_text_clause") 

374 is_from_clause = operator.attrgetter("_is_from_clause") 

375 is_tuple_type = operator.attrgetter("_is_tuple_type") 

376 is_table_value_type = operator.attrgetter("_is_table_value") 

377 is_selectable = operator.attrgetter("is_selectable") 

378 is_select_base = operator.attrgetter("_is_select_base") 

379 is_select_statement = operator.attrgetter("_is_select_statement") 

380 is_table = operator.attrgetter("_is_table") 

381 is_subquery = operator.attrgetter("_is_subquery") 

382 is_dml = operator.attrgetter("is_dml") 

383 

384 

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

386 return hasattr(t, "schema") 

387 

388 

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

390 return hasattr(s, "quote") 

391 

392 

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

394 return hasattr(s, "__clause_element__") 

395 

396 

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

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

399 

400 

401def _no_kw() -> exc.ArgumentError: 

402 return exc.ArgumentError( 

403 "Additional keyword arguments are not accepted by this " 

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

405 ) 

406 

407 

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

409 k = list(kw)[0] 

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

411 

412 

413@overload 

414def Nullable( 

415 val: "SQLCoreOperations[_T]", 

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

417 

418 

419@overload 

420def Nullable( 

421 val: roles.ExpressionElementRole[_T], 

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

423 

424 

425@overload 

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

427 

428 

429def Nullable( 

430 val: _TypedColumnClauseArgument[_T], 

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

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

433 

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

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

436 

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

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

439 

440 At runtime this method returns the input unchanged. 

441 

442 .. versionadded:: 2.0.20 

443 """ 

444 return val 

445 

446 

447@overload 

448def NotNullable( 

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

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

451 

452 

453@overload 

454def NotNullable( 

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

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

457 

458 

459@overload 

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

461 

462 

463@overload 

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

465 

466 

467def NotNullable( 

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

469) -> _TypedColumnClauseArgument[_T]: 

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

471 

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

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

474 nullable column:: 

475 

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

477 

478 At runtime this method returns the input unchanged. 

479 

480 .. versionadded:: 2.0.20 

481 """ 

482 return val # type: ignore