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

1# sql/_annotated_cols.py 

2# Copyright (C) 2005-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 

7from __future__ import annotations 

8 

9from typing import Any 

10from typing import Generic 

11from typing import Literal 

12from typing import NoReturn 

13from typing import overload 

14from typing import Protocol 

15from typing import TYPE_CHECKING 

16 

17from . import sqltypes 

18from ._typing import _T 

19from ._typing import _Ts 

20from .base import _NoArg 

21from .base import ReadOnlyColumnCollection 

22from .. import util 

23from ..exc import ArgumentError 

24from ..exc import InvalidRequestError 

25from ..util import typing as sa_typing 

26from ..util.langhelpers import dunders_re 

27from ..util.typing import Never 

28from ..util.typing import Self 

29from ..util.typing import TypeVar 

30from ..util.typing import Unpack 

31 

32if TYPE_CHECKING: 

33 from .elements import ColumnClause # noqa (for zimports) 

34 from .elements import KeyedColumnElement # noqa (for zimports) 

35 from .schema import Column 

36 from .type_api import TypeEngine 

37 from ..util.typing import _AnnotationScanType 

38 

39 

40class Named(Generic[_T]): 

41 """A named descriptor that is interpreted by SQLAlchemy in various ways. 

42 

43 .. seealso:: 

44 

45 :class:`_schema.TypedColumns` Define table columns using this 

46 descriptor. 

47 

48 .. versionadded:: 2.1.0b2 

49 """ 

50 

51 __slots__ = () 

52 

53 key: str 

54 if TYPE_CHECKING: 

55 

56 # NOTE: this overload prevents users from using the a TypedColumns 

57 # class like if it were an orm mapped class 

58 @overload 

59 def __get__(self, instance: None, owner: Any) -> Never: ... 

60 

61 @overload 

62 def __get__( 

63 self, instance: TypedColumns, owner: Any 

64 ) -> Column[_T]: ... 

65 @overload 

66 def __get__(self, instance: Any, owner: Any) -> Self: ... 

67 

68 def __get__(self, instance: object | None, owner: Any) -> Any: ... 

69 

70 

71# NOTE: TypedColumns subclasses are ignored by the ORM mapping process 

72class TypedColumns(ReadOnlyColumnCollection[str, "Column[Any]"]): 

73 """Class that generally represent the typed columns of a :class:`.Table`, 

74 but can be used with most :class:`_sql.FromClause` subclasses with the 

75 :meth:`_sql.FromClause.with_cols` method. 

76 

77 This is a "typing only" class that is never instantiated at runtime: the 

78 type checker will think that this class is exposed as the ``table.c`` 

79 attribute, but in reality a normal :class:`_schema.ColumnCollection` is 

80 used at runtime. 

81 

82 Subclasses should just list the columns as class attributes, without 

83 specifying method or other non column members. 

84 

85 To resolve the columns, a simplified version of the ORM logic is used, 

86 in particular, columns can be declared by: 

87 

88 * directly instantiating them, to declare constraint, custom SQL types and 

89 additional column options; 

90 * using only a :class:`.Named` or :class:`_schema.Column` type annotation, 

91 where nullability and SQL type will be inferred by the python type 

92 provided. 

93 Type inference is available for a common subset of python types. 

94 * a mix of both, where the instance can be used to declare 

95 constraints and other column options while the annotation will be used 

96 to set the SQL type and nullability if not provided by the instance. 

97 

98 In all cases the name is inferred from the attribute name, unless 

99 explicitly provided. 

100 

101 .. note:: 

102 

103 The generated table will create a copy of any column instance assigned 

104 as attributes of this class, so columns should be accessed only via 

105 the ``table.c`` collection, not using this class directly. 

106 

107 Example of the inference behavior:: 

108 

109 from sqlalchemy import Column, Integer, Named, String, TypedColumns 

110 

111 

112 class tbl_cols(TypedColumns): 

113 # the name will be set to ``id``, type is inferred as Column[int] 

114 id = Column(Integer, primary_key=True) 

115 

116 # not null String column is generated 

117 name: Named[str] 

118 

119 # nullable Double column is generated 

120 weight: Named[float | None] 

121 

122 # nullable Integer column, with sql name 'user_age' 

123 age: Named[int | None] = Column("user_age") 

124 

125 # not null column with type String(42) 

126 middle_name: Named[str] = Column(String(42)) 

127 

128 Mixins and subclasses are also supported:: 

129 

130 class with_id(TypedColumns): 

131 id = Column(Integer, primary_key=True) 

132 

133 

134 class named_cols(TypedColumns): 

135 name: Named[str] 

136 description: Named[str | None] 

137 

138 

139 class product_cols(named_cols, with_id): 

140 ean: Named[str] = Column(unique=True) 

141 

142 

143 product = Table("product", metadata, product_cols) 

144 

145 

146 class office_cols(named_cols, with_id): 

147 address: Named[str] 

148 

149 

150 office = Table("office", metadata, office_cols) 

151 

152 The positional types returned when selecting the table can 

153 be optionally declared by specifying a :attr:`.HasRowPos.__row_pos__` 

154 annotation:: 

155 

156 from sqlalchemy import select 

157 

158 

159 class some_cols(TypedColumns): 

160 id = Column(Integer, primary_key=True) 

161 name: Named[str] 

162 weight: Named[float | None] 

163 

164 __row_pos__: tuple[int, str, float | None] 

165 

166 

167 some_table = Table("st", metadata, some_cols) 

168 

169 # both will be typed as Select[int, str, float | None] 

170 stmt1 = some_table.select() 

171 stmt2 = select(some_table) 

172 

173 .. seealso:: 

174 

175 :class:`.Table` for usage details on how to use this class to 

176 create a table instance. 

177 

178 :meth:`_sql.FromClause.with_cols` to apply a :class:`.TypedColumns` 

179 to a from clause. 

180 

181 .. versionadded:: 2.1.0b2 

182 """ # noqa 

183 

184 __slots__ = () 

185 

186 if not TYPE_CHECKING: 

187 

188 def __new__(cls, *args: Any, **kwargs: Any) -> NoReturn: 

189 raise InvalidRequestError( 

190 "Cannot instantiate a TypedColumns object." 

191 ) 

192 

193 def __init_subclass__(cls) -> None: 

194 methods = { 

195 name 

196 for name, value in cls.__dict__.items() 

197 if not dunders_re.match(name) and callable(value) 

198 } 

199 if methods: 

200 raise InvalidRequestError( 

201 "TypedColumns subclasses may not define methods. " 

202 f"Found {sorted(methods)}" 

203 ) 

204 

205 

206_KeyColCC_co = TypeVar( 

207 "_KeyColCC_co", 

208 bound=ReadOnlyColumnCollection[str, "KeyedColumnElement[Any]"], 

209 covariant=True, 

210 default=ReadOnlyColumnCollection[str, "KeyedColumnElement[Any]"], 

211) 

212_ColClauseCC_co = TypeVar( 

213 "_ColClauseCC_co", 

214 bound=ReadOnlyColumnCollection[str, "ColumnClause[Any]"], 

215 covariant=True, 

216 default=ReadOnlyColumnCollection[str, "ColumnClause[Any]"], 

217) 

218_ColCC_co = TypeVar( 

219 "_ColCC_co", 

220 bound=ReadOnlyColumnCollection[str, "Column[Any]"], 

221 covariant=True, 

222 default=ReadOnlyColumnCollection[str, "Column[Any]"], 

223) 

224 

225_TC = TypeVar("_TC", bound=TypedColumns) 

226_TC_co = TypeVar("_TC_co", bound=TypedColumns, covariant=True) 

227 

228 

229class HasRowPos(Protocol[Unpack[_Ts]]): 

230 """Protocol for a :class:`_schema.TypedColumns` used to indicate the 

231 positional types will be returned when selecting the table. 

232 

233 .. versionadded:: 2.1.0b2 

234 """ 

235 

236 __row_pos__: tuple[Unpack[_Ts]] 

237 """A tuple that represents the types that will be returned when 

238 selecting from the table. 

239 """ 

240 

241 

242@util.preload_module("sqlalchemy.sql.schema") 

243def _extract_columns_from_class( 

244 table_columns_cls: type[TypedColumns], 

245) -> list[Column[Any]]: 

246 columns: dict[str, Column[Any]] = {} 

247 

248 Column = util.preloaded.sql_schema.Column 

249 NULL_UNSPECIFIED = util.preloaded.sql_schema.NULL_UNSPECIFIED 

250 

251 for base in table_columns_cls.__mro__[::-1]: 

252 if base in TypedColumns.__mro__: 

253 continue 

254 

255 # _ClassScanAbstractConfig._cls_attr_resolver 

256 cls_annotations = util.get_annotations(base) 

257 cls_vars = vars(base) 

258 items = [ 

259 (n, cls_vars.get(n), cls_annotations.get(n)) 

260 for n in util.merge_lists_w_ordering( 

261 list(cls_vars), list(cls_annotations) 

262 ) 

263 if not dunders_re.match(n) 

264 ] 

265 # -- 

266 for name, obj, annotation in items: 

267 if obj is None: 

268 assert annotation is not None 

269 # no attribute, just annotation 

270 extracted_type = _collect_annotation( 

271 table_columns_cls, name, base.__module__, annotation 

272 ) 

273 if extracted_type is _NoArg.NO_ARG: 

274 raise ArgumentError( 

275 "No type information could be extracted from " 

276 f"annotation {annotation} for attribute " 

277 f"'{base.__name__}.{name}'" 

278 ) 

279 sqltype = _get_sqltype(extracted_type) 

280 if sqltype is None: 

281 raise ArgumentError( 

282 f"Could not find a SQL type for type {extracted_type} " 

283 f"obtained from annotation {annotation} in " 

284 f"attribute '{base.__name__}.{name}'" 

285 ) 

286 columns[name] = Column( 

287 name, 

288 sqltype, 

289 nullable=sa_typing.includes_none(extracted_type), 

290 ) 

291 elif isinstance(obj, Column): 

292 # has attribute attribute 

293 # _DeclarativeMapperConfig._produce_column_copies 

294 # as with orm this case is not supported 

295 for fk in obj.foreign_keys: 

296 if ( 

297 fk._table_column is not None 

298 and fk._table_column.table is None 

299 ): 

300 raise InvalidRequestError( 

301 f"Column '{base.__name__}.{name}' with foreign " 

302 "key to non-table-bound columns is not supported " 

303 "when using a TypedColumns. If possible use the " 

304 "qualified string name the column" 

305 ) 

306 

307 col = obj._copy() 

308 # MapptedColumn.declarative_scan 

309 if col.key == col.name and col.key != name: 

310 col.key = name 

311 if col.name is None: 

312 col.name = name 

313 

314 sqltype = col.type 

315 anno_sqltype = None 

316 nullable: Literal[_NoArg.NO_ARG] | bool = _NoArg.NO_ARG 

317 if annotation is not None: 

318 # there is an annotation, extract the type 

319 extracted_type = _collect_annotation( 

320 table_columns_cls, name, base.__module__, annotation 

321 ) 

322 if extracted_type is not _NoArg.NO_ARG: 

323 anno_sqltype = _get_sqltype(extracted_type) 

324 nullable = sa_typing.includes_none(extracted_type) 

325 

326 if sqltype._isnull: 

327 if anno_sqltype is None and not col.foreign_keys: 

328 raise ArgumentError( 

329 "Python typing annotation is required for " 

330 f"attribute '{base.__name__}.{name}' when " 

331 "primary argument(s) for Column construct are " 

332 "None or not present" 

333 ) 

334 elif anno_sqltype is not None: 

335 col._set_type(anno_sqltype) 

336 

337 if ( 

338 nullable is not _NoArg.NO_ARG 

339 and col._user_defined_nullable is NULL_UNSPECIFIED 

340 and not col.primary_key 

341 ): 

342 col.nullable = nullable 

343 columns[name] = col 

344 else: 

345 raise ArgumentError( 

346 f"Unexpected value for attribute '{base.__name__}.{name}'" 

347 f". Expected a Column, not: {type(obj)}" 

348 ) 

349 

350 # Return columns as a list 

351 return list(columns.values()) 

352 

353 

354@util.preload_module("sqlalchemy.sql.schema") 

355def _collect_annotation( 

356 cls: type[Any], name: str, module: str, raw_annotation: _AnnotationScanType 

357) -> _AnnotationScanType | Literal[_NoArg.NO_ARG]: 

358 Column = util.preloaded.sql_schema.Column 

359 

360 _locals = {"Column": Column, "Named": Named} 

361 # _ClassScanAbstractConfig._collect_annotation & _extract_mapped_subtype 

362 try: 

363 annotation = sa_typing.de_stringify_annotation( 

364 cls, raw_annotation, module, _locals 

365 ) 

366 except Exception as e: 

367 raise ArgumentError( 

368 f"Could not interpret annotation {raw_annotation} for " 

369 f"attribute '{cls.__name__}.{name}'" 

370 ) from e 

371 

372 if ( 

373 not sa_typing.is_generic(annotation) 

374 and isinstance(annotation, type) 

375 and issubclass(annotation, (Column, Named)) 

376 ): 

377 # no generic information, ignore 

378 return _NoArg.NO_ARG 

379 elif not sa_typing.is_origin_of_cls(annotation, (Column, Named)): 

380 raise ArgumentError( 

381 f"Annotation {raw_annotation} for attribute " 

382 f"'{cls.__name__}.{name}' is not of type Named/Column[...]" 

383 ) 

384 else: 

385 assert len(annotation.__args__) == 1 # Column[int, int] raises 

386 return annotation.__args__[0] # type: ignore[no-any-return] 

387 

388 

389def _get_sqltype(annotation: _AnnotationScanType) -> TypeEngine[Any] | None: 

390 our_type = sa_typing.de_optionalize_union_types(annotation) 

391 # simplified version of registry._resolve_type given no customizable 

392 # type map 

393 sql_type = sqltypes._type_map_get(our_type) # type: ignore[arg-type] 

394 if sql_type is not None and not sql_type._isnull: 

395 return sqltypes.to_instance(sql_type) 

396 else: 

397 return None