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

1# sql/_typing.py 

2# Copyright (C) 2022-2024 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 Unpack 

35 

36if TYPE_CHECKING: 

37 from datetime import date 

38 from datetime import datetime 

39 from datetime import time 

40 from datetime import timedelta 

41 from decimal import Decimal 

42 from uuid import UUID 

43 

44 from .base import Executable 

45 from .compiler import Compiled 

46 from .compiler import DDLCompiler 

47 from .compiler import SQLCompiler 

48 from .dml import UpdateBase 

49 from .dml import ValuesBase 

50 from .elements import ClauseElement 

51 from .elements import ColumnElement 

52 from .elements import KeyedColumnElement 

53 from .elements import quoted_name 

54 from .elements import SQLCoreOperations 

55 from .elements import TextClause 

56 from .lambdas import LambdaElement 

57 from .roles import FromClauseRole 

58 from .schema import Column 

59 from .selectable import Alias 

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 Dialect 

74 from ..util.typing import TypeGuard 

75 

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

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

78 

79 

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

81 

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

83 

84 

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

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

87 

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

89 

90 

91class _CoreAdapterProto(Protocol): 

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

93 

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

95 

96 

97class _HasDialect(Protocol): 

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

99 attribute. 

100 """ 

101 

102 @property 

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

104 

105 

106# match column types that are not ORM entities 

107_NOT_ENTITY = TypeVar( 

108 "_NOT_ENTITY", 

109 int, 

110 str, 

111 bool, 

112 "datetime", 

113 "date", 

114 "time", 

115 "timedelta", 

116 "UUID", 

117 float, 

118 "Decimal", 

119) 

120 

121_StarOrOne = Literal["*", 1] 

122 

123_MAYBE_ENTITY = TypeVar( 

124 "_MAYBE_ENTITY", 

125 roles.ColumnsClauseRole, 

126 _StarOrOne, 

127 Type[Any], 

128 Inspectable[_HasClauseElement[Any]], 

129 _HasClauseElement[Any], 

130) 

131 

132 

133# convention: 

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

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

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

137# XYZElement. 

138 

139_TextCoercedExpressionArgument = Union[ 

140 str, 

141 "TextClause", 

142 "ColumnElement[_T]", 

143 _HasClauseElement[_T], 

144 roles.ExpressionElementRole[_T], 

145] 

146 

147_ColumnsClauseArgument = Union[ 

148 roles.TypedColumnsClauseRole[_T], 

149 roles.ColumnsClauseRole, 

150 "SQLCoreOperations[_T]", 

151 _StarOrOne, 

152 Type[_T], 

153 Inspectable[_HasClauseElement[_T]], 

154 _HasClauseElement[_T], 

155] 

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

157 

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

159 

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

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

162 

163 

164""" 

165 

166_TypedColumnClauseArgument = Union[ 

167 roles.TypedColumnsClauseRole[_T], 

168 "SQLCoreOperations[_T]", 

169 Type[_T], 

170] 

171 

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

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

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

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

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

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

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

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

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

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

182 

183 

184_ColumnExpressionArgument = Union[ 

185 "ColumnElement[_T]", 

186 _HasClauseElement[_T], 

187 "SQLCoreOperations[_T]", 

188 roles.ExpressionElementRole[_T], 

189 roles.TypedColumnsClauseRole[_T], 

190 Callable[[], "ColumnElement[_T]"], 

191 "LambdaElement", 

192] 

193"See docs in public alias ColumnExpressionArgument." 

194 

195ColumnExpressionArgument: TypeAlias = _ColumnExpressionArgument[_T] 

196"""Narrower "column expression" argument. 

197 

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

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

200way a table or ORM entity does. 

201 

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

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

204overall which brings in the TextClause object also. 

205 

206.. versionadded:: 2.0.13 

207 

208""" 

209 

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

211 

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

213 

214_ByArgument = Union[ 

215 Iterable[_ColumnExpressionOrStrLabelArgument[Any]], 

216 _ColumnExpressionOrStrLabelArgument[Any], 

217] 

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

219 

220 

221_InfoType = Dict[Any, Any] 

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

223 

224_FromClauseArgument = Union[ 

225 roles.FromClauseRole, 

226 Type[Any], 

227 Inspectable[_HasClauseElement[Any]], 

228 _HasClauseElement[Any], 

229] 

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

231 

232Also accommodates ORM entities and related constructs. 

233 

234""" 

235 

236_JoinTargetArgument = Union[_FromClauseArgument, roles.JoinTargetRole] 

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

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

239 

240""" 

241 

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

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

244come from the ORM. 

245 

246""" 

247 

248_SelectStatementForCompoundArgument = Union[ 

249 "SelectBase", roles.CompoundElementRole 

250] 

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

252 

253_DMLColumnArgument = Union[ 

254 str, 

255 _HasClauseElement[Any], 

256 roles.DMLColumnRole, 

257 "SQLCoreOperations[Any]", 

258] 

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

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

261 

262These are usually strings or SQL table columns. 

263 

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

265the DMLColumnRole to be able to accommodate. 

266 

267""" 

268 

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

270_DMLColumnKeyMapping = Mapping[_DMLKey, Any] 

271 

272 

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

274"""DDL column. 

275 

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

277 

278""" 

279 

280_DDLColumnReferenceArgument = _DDLColumnArgument 

281 

282_DMLTableArgument = Union[ 

283 "TableClause", 

284 "Join", 

285 "Alias", 

286 "CTE", 

287 Type[Any], 

288 Inspectable[_HasClauseElement[Any]], 

289 _HasClauseElement[Any], 

290] 

291 

292_PropagateAttrsType = util.immutabledict[str, Any] 

293 

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

295 

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

297 

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

299 

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

301 

302if TYPE_CHECKING: 

303 

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

305 

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

307 

308 def is_named_from_clause( 

309 t: FromClauseRole, 

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

311 

312 def is_column_element( 

313 c: ClauseElement, 

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

315 

316 def is_keyed_column_element( 

317 c: ClauseElement, 

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

319 

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

321 

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

323 

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

325 

326 def is_table_value_type( 

327 t: TypeEngine[Any], 

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

329 

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

331 

332 def is_select_base( 

333 t: Union[Executable, ReturnsRows] 

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

335 

336 def is_select_statement( 

337 t: Union[Executable, ReturnsRows] 

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

339 

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

341 

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

343 

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

345 

346else: 

347 is_sql_compiler = operator.attrgetter("is_sql") 

348 is_ddl_compiler = operator.attrgetter("is_ddl") 

349 is_named_from_clause = operator.attrgetter("named_with_column") 

350 is_column_element = operator.attrgetter("_is_column_element") 

351 is_keyed_column_element = operator.attrgetter("_is_keyed_column_element") 

352 is_text_clause = operator.attrgetter("_is_text_clause") 

353 is_from_clause = operator.attrgetter("_is_from_clause") 

354 is_tuple_type = operator.attrgetter("_is_tuple_type") 

355 is_table_value_type = operator.attrgetter("_is_table_value") 

356 is_selectable = operator.attrgetter("is_selectable") 

357 is_select_base = operator.attrgetter("_is_select_base") 

358 is_select_statement = operator.attrgetter("_is_select_statement") 

359 is_table = operator.attrgetter("_is_table") 

360 is_subquery = operator.attrgetter("_is_subquery") 

361 is_dml = operator.attrgetter("is_dml") 

362 

363 

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

365 return hasattr(t, "schema") 

366 

367 

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

369 return hasattr(s, "quote") 

370 

371 

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

373 return hasattr(s, "__clause_element__") 

374 

375 

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

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

378 

379 

380def _no_kw() -> exc.ArgumentError: 

381 return exc.ArgumentError( 

382 "Additional keyword arguments are not accepted by this " 

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

384 ) 

385 

386 

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

388 k = list(kw)[0] 

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

390 

391 

392@overload 

393def Nullable( 

394 val: "SQLCoreOperations[_T]", 

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

396 

397 

398@overload 

399def Nullable( 

400 val: roles.ExpressionElementRole[_T], 

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

402 

403 

404@overload 

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

406 

407 

408def Nullable( 

409 val: _TypedColumnClauseArgument[_T], 

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

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

412 

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

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

415 

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

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

418 

419 At runtime this method returns the input unchanged. 

420 

421 .. versionadded:: 2.0.20 

422 """ 

423 return val 

424 

425 

426@overload 

427def NotNullable( 

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

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

430 

431 

432@overload 

433def NotNullable( 

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

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

436 

437 

438@overload 

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

440 

441 

442@overload 

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

444 

445 

446def NotNullable( 

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

448) -> _TypedColumnClauseArgument[_T]: 

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

450 

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

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

453 nullable column:: 

454 

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

456 

457 At runtime this method returns the input unchanged. 

458 

459 .. versionadded:: 2.0.20 

460 """ 

461 return val # type: ignore