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 :class:`_schema.Column`, to declare constraint, 

89 custom SQL types and additional column options. The annotation is 

90 usually inferred by type checkers from the column instance; 

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

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

93 provided. 

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

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

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

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

98 The information provided in the instance will take precedence over 

99 the annotation when they are conflicting, for example if setting 

100 nullable=True with an annotation tha does not include ``None``. 

101 

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

103 explicitly provided. 

104 

105 .. note:: 

106 

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

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

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

110 

111 Example of the inference behavior:: 

112 

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

114 

115 

116 class tbl_cols(TypedColumns): 

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

118 # from the Integer SQL type. 

119 id = Column(Integer, primary_key=True) 

120 

121 # not null String column is generated 

122 name: Named[str] 

123 

124 # nullable Double column is generated 

125 weight: Named[float | None] 

126 

127 # nullable Integer column, with SQL name 'user_age' 

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

129 

130 # not null column with type String(42) 

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

132 

133 Mixins and subclasses are also supported:: 

134 

135 class with_id(TypedColumns): 

136 id = Column(Integer, primary_key=True) 

137 

138 

139 class named_cols(TypedColumns): 

140 name: Named[str] 

141 description: Named[str | None] 

142 

143 

144 class product_cols(named_cols, with_id): 

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

146 

147 

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

149 

150 

151 class office_cols(named_cols, with_id): 

152 address: Named[str] 

153 

154 

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

156 

157 The positional types returned when selecting the table can 

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

159 annotation:: 

160 

161 from sqlalchemy import select 

162 

163 

164 class some_cols(TypedColumns): 

165 id = Column(Integer, primary_key=True) 

166 name: Named[str] 

167 weight: Named[float | None] 

168 

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

170 

171 

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

173 

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

175 stmt1 = some_table.select() 

176 stmt2 = select(some_table) 

177 

178 .. seealso:: 

179 

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

181 create a table instance. 

182 

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

184 to a from clause. 

185 

186 .. versionadded:: 2.1.0b2 

187 """ # noqa 

188 

189 __slots__ = () 

190 

191 if not TYPE_CHECKING: 

192 

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

194 raise InvalidRequestError( 

195 "Cannot instantiate a TypedColumns object." 

196 ) 

197 

198 def __init_subclass__(cls) -> None: 

199 methods = { 

200 name 

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

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

203 } 

204 if methods: 

205 raise InvalidRequestError( 

206 "TypedColumns subclasses may not define methods. " 

207 f"Found {sorted(methods)}" 

208 ) 

209 

210 

211_KeyColCC_co = TypeVar( 

212 "_KeyColCC_co", 

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

214 covariant=True, 

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

216) 

217_ColClauseCC_co = TypeVar( 

218 "_ColClauseCC_co", 

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

220 covariant=True, 

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

222) 

223_ColCC_co = TypeVar( 

224 "_ColCC_co", 

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

226 covariant=True, 

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

228) 

229 

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

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

232 

233 

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

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

236 positional types will be returned when selecting the table. 

237 

238 .. versionadded:: 2.1.0b2 

239 """ 

240 

241 __row_pos__: tuple[Unpack[_Ts]] 

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

243 selecting from the table. 

244 """ 

245 

246 

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

248def _extract_columns_from_class( 

249 table_columns_cls: type[TypedColumns], 

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

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

252 

253 Column = util.preloaded.sql_schema.Column 

254 NULL_UNSPECIFIED = util.preloaded.sql_schema.NULL_UNSPECIFIED 

255 

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

257 if base in TypedColumns.__mro__: 

258 continue 

259 

260 # _ClassScanAbstractConfig._cls_attr_resolver 

261 cls_annotations = util.get_annotations(base) 

262 cls_vars = vars(base) 

263 items = [ 

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

265 for n in util.merge_lists_w_ordering( 

266 list(cls_vars), list(cls_annotations) 

267 ) 

268 if not dunders_re.match(n) 

269 ] 

270 # -- 

271 for name, obj, annotation in items: 

272 if obj is None: 

273 assert annotation is not None 

274 # no attribute, just annotation 

275 extracted_type = _collect_annotation( 

276 table_columns_cls, name, base.__module__, annotation 

277 ) 

278 if extracted_type is _NoArg.NO_ARG: 

279 raise ArgumentError( 

280 "No type information could be extracted from " 

281 f"annotation {annotation} for attribute " 

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

283 ) 

284 sqltype = _get_sqltype(extracted_type) 

285 if sqltype is None: 

286 raise ArgumentError( 

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

288 f"obtained from annotation {annotation} in " 

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

290 ) 

291 columns[name] = Column( 

292 name, 

293 sqltype, 

294 nullable=sa_typing.includes_none(extracted_type), 

295 ) 

296 elif isinstance(obj, Column): 

297 # has attribute attribute 

298 # _DeclarativeMapperConfig._produce_column_copies 

299 # as with orm this case is not supported 

300 for fk in obj.foreign_keys: 

301 if ( 

302 fk._table_column is not None 

303 and fk._table_column.table is None 

304 ): 

305 raise InvalidRequestError( 

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

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

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

309 "qualified string name the column" 

310 ) 

311 

312 col = obj._copy() 

313 # MapptedColumn.declarative_scan 

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

315 col.key = name 

316 if col.name is None: 

317 col.name = name 

318 

319 sqltype = col.type 

320 anno_sqltype = None 

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

322 if annotation is not None: 

323 # there is an annotation, extract the type 

324 extracted_type = _collect_annotation( 

325 table_columns_cls, name, base.__module__, annotation 

326 ) 

327 if extracted_type is not _NoArg.NO_ARG: 

328 anno_sqltype = _get_sqltype(extracted_type) 

329 nullable = sa_typing.includes_none(extracted_type) 

330 

331 if sqltype._isnull: 

332 if anno_sqltype is None and not col.foreign_keys: 

333 raise ArgumentError( 

334 "Python typing annotation is required for " 

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

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

337 "None or not present" 

338 ) 

339 elif anno_sqltype is not None: 

340 col._set_type(anno_sqltype) 

341 

342 if ( 

343 nullable is not _NoArg.NO_ARG 

344 and col._user_defined_nullable is NULL_UNSPECIFIED 

345 and not col.primary_key 

346 ): 

347 col.nullable = nullable 

348 columns[name] = col 

349 else: 

350 raise ArgumentError( 

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

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

353 ) 

354 

355 # Return columns as a list 

356 return list(columns.values()) 

357 

358 

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

360def _collect_annotation( 

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

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

363 Column = util.preloaded.sql_schema.Column 

364 

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

366 # _ClassScanAbstractConfig._collect_annotation & _extract_mapped_subtype 

367 try: 

368 annotation = sa_typing.de_stringify_annotation( 

369 cls, raw_annotation, module, _locals 

370 ) 

371 except Exception as e: 

372 raise ArgumentError( 

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

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

375 ) from e 

376 

377 if ( 

378 not sa_typing.is_generic(annotation) 

379 and isinstance(annotation, type) 

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

381 ): 

382 # no generic information, ignore 

383 return _NoArg.NO_ARG 

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

385 raise ArgumentError( 

386 f"Annotation {raw_annotation} for attribute " 

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

388 ) 

389 else: 

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

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

392 

393 

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

395 our_type = sa_typing.de_optionalize_union_types(annotation) 

396 # simplified version of registry._resolve_type given no customizable 

397 # type map 

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

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

400 return sqltypes.to_instance(sql_type) 

401 else: 

402 return None