Coverage for snekql/model.py: 87%

1"""Table model declaration and materialization behavior."""

3from __future__ import annotations

5from collections.abc import Mapping

6from types import EllipsisType

7from typing import (

8 Any,

9 ClassVar,

10 Literal,

11 Self,

12 TypeVar,

13 cast,

14 dataclass_transform,

15 get_origin,

16)

18from snekql.errors import (

19 FrozenModelError,

20 ModelDeclarationError,

21 ModelValidationError,

22 SnekqlError,

23)

24from snekql.indexes import NormalizedIndex, require_index_declaration

25from snekql.storage import (

26 MISSING,

27 Attr,

28 Blob,

29 Boolean,

30 CurrentTimestamp,

31 DateTime,

32 Integer,

33 Json,

34 Missing,

35 Real,

36 Text,

37)

39type BackendFamily = Literal["mariadb", "sqlite"]

41StateT = TypeVar("StateT")

42ReadModelT = TypeVar("ReadModelT", bound="Table[Any]")

43T = TypeVar("T")

46class Pending:

47 """Marker state for application-constructed table models.

49 >>> state: type[Pending] = Pending

50 >>> state.__name__

51 'Pending'

52 """

55class Fetched:

56 """Marker state for table models materialized by the Query Runtime.

58 A generated column may be `T | Missing` on `Pending` instances but `T` on

59 `Fetched` instances.

61 >>> state: type[Fetched] = Fetched

62 >>> state.__name__

63 'Fetched'

64 """

67class Table[StateT]:

68 """Base type shared by concrete table models in any lifecycle state.

70 Query builders use this shallow base to constrain model-like generic

71 parameters without requiring runtime construction behavior yet.

72 """

74 @classmethod

75 def __owner_type__(cls) -> type[Self]:

76 return cls

79# Private normal persisted-column alias used to build the public Col alias.

80type _Col[WriteModelT: Table[Any], FetchedModelT, T] = Attr[

81 WriteModelT,

82 FetchedModelT,

83 WriteModelT,

84 T,

85 T,

86]

88# Private generated-column alias used to model pending Missing vs fetched T.

89type _GenCol[WriteModelT: Table[Any], FetchedModelT, T] = Attr[

90 WriteModelT,

91 FetchedModelT,

92 WriteModelT,

93 T | Missing,

94 T,

95]

97# Public normal persisted-column alias for external table model helpers.

98type Col[WriteModelT: Table[Any], FetchedModelT, T] = _Col[

99 WriteModelT,

100 FetchedModelT,

101 T,

102]

103

104# Public generated/server-filled column alias for external model helpers.

105type GenCol[WriteModelT: Table[Any], FetchedModelT, T] = _GenCol[

106 WriteModelT,

107 FetchedModelT,

108 T,

109]

110

111

112@dataclass_transform(

113 field_specifiers=(Integer, Real, Text, Blob, Json, Boolean, DateTime),

114 kw_only_default=True,

115)

116class ModelMeta(type):

117 """Typing/runtime hook for direct public column descriptors.

118

119 Intended runtime behavior:

120 - treat public column descriptors like `email: User.Col[str] = Text(...)`

121 as both constructor fields and query descriptors

122 - use descriptor `__set__` typing for constructor/write values

123 - bind public descriptors directly on the model class

124 - store values in hidden internal storage keyed by public names

125 - keep fetched-state generated values narrowed relative to pending-state values

126 """

127

128 def __new__(

129 mcls,

130 name: str,

131 bases: tuple[type, ...],

132 namespace: dict[str, object],

133 **kwargs: object,

134 ) -> type:

135 is_model_base = name == "Model"

136 if not is_model_base:

137 ModelMeta._validate_model_bases(bases)

138 ModelMeta._validate_model_namespace(namespace)

139 model_class = super().__new__(mcls, name, bases, namespace, **kwargs)

140 annotations = ModelMeta._namespace_annotations(namespace)

141 columns = ModelMeta._bind_columns(model_class, annotations)

142 model_metadata = cast("Any", model_class)

143 if not is_model_base:

144 model_metadata.__tablename__ = ModelMeta._resolve_table_name(

145 name,

146 namespace,

147 )

148 model_metadata.__snekql_backend__ = ModelMeta._resolve_backend_family(

149 bases,

150 namespace,

151 )

152 model_metadata.__snekql_columns__ = columns

153 if is_model_base:

154 model_metadata.__snekql_indexes__ = ()

155 else:

156 model_metadata.__snekql_indexes__ = ModelMeta._bind_indexes(

157 model_class,

158 namespace,

159 columns,

160 )

161 return model_class

162

163 @staticmethod

164 def _validate_model_bases(bases: tuple[type, ...]) -> None:

165 """Reject inheritance forms that would blur v1 table model semantics."""

166

167 for base in bases:

168 if isinstance(base, ModelMeta):

169 if base.__name__ != "Model":

170 msg = f"cannot subclass concrete model: {base.__name__}"

171 raise ModelDeclarationError(msg)

172 continue

173 if base.__name__ == "Generic":

174 continue

175 msg = f"model mixin bases are not supported: {base.__name__}"

176 raise ModelDeclarationError(msg)

177

178 @staticmethod

179 def _validate_model_namespace(namespace: dict[str, object]) -> None:

180 """Validate class-body contents before freezing snekql metadata."""

181

182 annotations = ModelMeta._namespace_annotations(namespace)

183 for annotated_name, annotation in annotations.items():

184 annotated_value = namespace.get(annotated_name)

185 if isinstance(annotated_value, Attr):

186 continue

187 if annotated_name in {"__tablename__", "__indexes__"}:

188 continue

189 if ModelMeta._is_classvar_annotation(annotation):

190 continue

191 msg = f"unsupported model annotation: {annotated_name!r}"

192 raise ModelDeclarationError(msg)

193 for attribute_name, attribute_value in namespace.items():

194 if isinstance(attribute_value, property):

195 msg = f"computed properties are not supported: {attribute_name!r}"

196 raise ModelDeclarationError(msg)

197 if getattr(attribute_value, "__isabstractmethod__", False):

198 msg = f"abstract members are not supported: {attribute_name!r}"

199 raise ModelDeclarationError(msg)

200 if attribute_name == "__indexes__" and not isinstance(

201 attribute_value,

202 list,

203 ):

204 msg = "__indexes__ must be a list"

205 raise ModelDeclarationError(msg)

206

207 @staticmethod

208 def _namespace_annotations(namespace: dict[str, object]) -> dict[str, object]:

209 annotations_object = namespace.get("__annotations__", {})

210 if not isinstance(annotations_object, dict):

211 return {}

212 return cast("dict[str, object]", annotations_object)

213

214 @staticmethod

215 def _bind_columns(

216 model_class: type,

217 annotations: dict[str, object],

218 ) -> dict[str, Attr[Any, Any, Any, Any, Any]]:

219 columns: dict[str, Attr[Any, Any, Any, Any, Any]] = {}

220 for attribute_name, attribute_value in model_class.__dict__.items():

221 if not isinstance(attribute_value, Attr):

222 continue

223 column = cast("Attr[Any, Any, Any, Any, Any]", attribute_value)

224 if not ModelMeta._is_sql_identifier(attribute_name):

225 msg = f"invalid column identifier: {attribute_name!r}"

226 raise ModelDeclarationError(msg)

227 column.is_generated = ModelMeta._is_generated_annotation(

228 annotations.get(attribute_name),

229 )

230 ModelMeta._validate_column_declaration(attribute_name, column)

231 columns[attribute_name] = column

232 return columns

233

234 @staticmethod

235 def _bind_indexes(

236 model_class: type,

237 namespace: dict[str, object],

238 columns: dict[str, Attr[Any, Any, Any, Any, Any]],

239 ) -> tuple[NormalizedIndex, ...]:

240 indexes_object = namespace.get("__indexes__", [])

241 if not isinstance(indexes_object, list):

242 msg = "__indexes__ must be a list"

243 raise ModelDeclarationError(msg)

244 index_declarations = cast("list[object]", indexes_object)

245 column_names_by_id = {id(column): name for name, column in columns.items()}

246 table_name = cast("str", cast("Any", model_class).__tablename__)

247 indexes: list[NormalizedIndex] = [

248 NormalizedIndex(

249 column_names=(column_name,),

250 name=f"ux_{table_name}_{column_name}",

251 unique=True,

252 )

253 for column_name, column in columns.items()

254 if column.unique

255 ]

256 table_indexes: list[NormalizedIndex] = []

257 for index_object in index_declarations:

258 index = require_index_declaration(index_object)

259 column_names: list[str] = []

260 for column in index.columns:

261 if column.owner is not model_class:

262 msg = "index columns must belong to the declaring model"

263 raise ModelDeclarationError(msg)

264 try:

265 column_names.append(column_names_by_id[id(column)])

266 except KeyError as error:

267 msg = "index columns must be declared model columns"

268 raise ModelDeclarationError(msg) from error

269 index_name = index.name

270 if index_name is None:

271 prefix = "ux" if index.unique else "ix"

272 index_name = f"{prefix}_{table_name}_{'_'.join(column_names)}"

273 elif not ModelMeta._is_sql_identifier(index_name):

274 msg = f"invalid index identifier: {index_name!r}"

275 raise ModelDeclarationError(msg)

276 normalized_index = NormalizedIndex(

277 column_names=tuple(column_names),

278 name=index_name,

279 unique=index.unique,

280 )

281 indexes.append(normalized_index)

282 table_indexes.append(normalized_index)

283 ModelMeta._validate_index_set(indexes)

284 return tuple(table_indexes)

285

286 @staticmethod

287 def _validate_index_set(indexes: list[NormalizedIndex]) -> None:

288 names: set[str] = set()

289 column_lists: set[tuple[str, ...]] = set()

290 for index in indexes:

291 if index.name in names:

292 msg = f"duplicate index name: {index.name!r}"

293 raise ModelDeclarationError(msg)

294 names.add(index.name)

295 if index.column_names in column_lists:

296 msg = f"duplicate index column list: {index.column_names!r}"

297 raise ModelDeclarationError(msg)

298 column_lists.add(index.column_names)

299

300 @staticmethod

301 def _resolve_backend_family(

302 bases: tuple[type, ...],

303 namespace: dict[str, object],

304 ) -> BackendFamily:

305 configured_backend = namespace.get("__snekql_backend__")

306 if configured_backend in {"mariadb", "sqlite"}:

307 return cast("BackendFamily", configured_backend)

308 for base in bases:

309 inherited_backend = getattr(base, "__snekql_backend__", None)

310 if inherited_backend in {"mariadb", "sqlite"}:

311 return cast("BackendFamily", inherited_backend)

312 return "sqlite"

313

314 @staticmethod

315 def _resolve_table_name(name: str, namespace: dict[str, object]) -> str:

316 table_name = namespace.get("__tablename__", ModelMeta._infer_table_name(name))

317 if not isinstance(table_name, str) or not ModelMeta._is_sql_identifier(

318 table_name,

319 ):

320 msg = f"invalid table identifier: {table_name!r}"

321 raise ModelDeclarationError(msg)

322 return table_name

323

324 @staticmethod

325 def _is_generated_annotation(annotation: object) -> bool:

326 if annotation is None:

327 return False

328 return "GenCol[" in str(annotation)

329

330 @staticmethod

331 def _validate_column_declaration(

332 name: str,

333 column: Attr[Any, Any, Any, Any, Any],

334 ) -> None:

335 if isinstance(column.default, CurrentTimestamp):

336 msg = f"CurrentTimestamp cannot be a Python default for {name!r}"

337 raise ModelDeclarationError(

338 msg,

339 )

340 if column.unique and column.primary_key:

341 msg = f"primary-key columns cannot be unique: {name!r}"

342 raise ModelDeclarationError(msg)

343 if column.auto_increment and (

344 not column.primary_key or column.storage_type_name != "Integer"

345 ):

346 msg = f"auto-increment requires an integer primary-key column: {name!r}"

347 raise ModelDeclarationError(msg)

348 if column.server_default is None:

349 return

350 if not isinstance(column.server_default, CurrentTimestamp):

351 msg = f"unsupported server default for {name!r}"

352 raise ModelDeclarationError(

353 msg,

354 )

355 if column.storage_type_name != "DateTime":

356 msg = f"CurrentTimestamp requires a DateTime column: {name!r}"

357 raise ModelDeclarationError(

358 msg,

359 )

360 if not column.is_generated:

361 msg = f"CurrentTimestamp requires a generated column: {name!r}"

362 raise ModelDeclarationError(

363 msg,

364 )

365 if column.default is not MISSING:

366 msg = (

367 f"CurrentTimestamp generated columns must default to MISSING: {name!r}"

368 )

369 raise ModelDeclarationError(

370 msg,

371 )

372 if not isinstance(column.default_factory, EllipsisType):

373 msg = f"CurrentTimestamp generated columns cannot use default_factory: {name!r}"

374 raise ModelDeclarationError(

375 msg,

376 )

377

378 @staticmethod

379 def _infer_table_name(class_name: str) -> str:

380 characters: list[str] = []

381 previous_was_lower_or_digit = False

382 for character in class_name:

383 if character.isupper() and previous_was_lower_or_digit:

384 characters.append("_")

385 characters.append(character.lower())

386 previous_was_lower_or_digit = character.islower() or character.isdigit()

387 return "".join(characters)

388

389 @staticmethod

390 def _is_classvar_annotation(annotation: object) -> bool:

391 if isinstance(annotation, str):

392 return annotation.startswith(("ClassVar[", "typing.ClassVar["))

393 return get_origin(annotation) is ClassVar

394

395 @staticmethod

396 def _is_sql_identifier(value: str) -> bool:

397 if value == "":

398 return False

399 first_character = value[0]

400 if not (first_character.isalpha() or first_character == "_"):

401 return False

402 return all(character.isalnum() or character == "_" for character in value)

403

404

405class Model[StateT, ReadModelT: "Table[Any]"](Table[StateT], metaclass=ModelMeta):

406 """Base class for declaring table models.

407

408 >>> class User[S = Pending](Model[S, "User[Fetched]"]):

409 ... email: User.Col[str] = Text(nullable=False)

410 """

411

412 __snekql_backend__: ClassVar[Literal["sqlite"]] = "sqlite"

413 __snekql_columns__: ClassVar[dict[str, Attr[Any, Any, Any, Any, Any]]]

414 __snekql_indexes__: ClassVar[tuple[NormalizedIndex, ...]]

415 __tablename__: ClassVar[str]

416

417 # Normal persisted-column alias scoped to the declaring model class.

418 type Col[T] = Attr[Self, ReadModelT, Self, T, T]

419 # Generated/server-filled column alias scoped to the declaring model class.

420 type GenCol[T] = Attr[Self, ReadModelT, Self, T | Missing, T]

421

422 def __init__(self, **values: object) -> None:

423 remaining_values = dict(values)

424 storage = cast(

425 "dict[str, object]",

426 object.__getattribute__(self, "__dict__"),

427 )

428 storage["_snekql_frozen"] = False

429 storage["_snekql_state"] = "Pending"

430 for name, column in self.__class__.__snekql_columns__.items():

431 if name in remaining_values:

432 value = remaining_values.pop(name)

433 else:

434 try:

435 value = column.build_default()

436 except SnekqlError:

437 raise

438 except Exception as error:

439 msg = f"default factory failed for {name!r}"

440 raise ModelValidationError(

441 msg,

442 ) from error

443 if isinstance(value, EllipsisType):

444 msg = f"missing required value for {name!r}"

445 raise ModelValidationError(msg)

446 setattr(self, name, column.validate_model_value(value))

447 if remaining_values:

448 names = ", ".join(sorted(remaining_values))

449 msg = f"unknown model values: {names}"

450 raise ModelValidationError(msg)

451 storage["_snekql_frozen"] = True

452

453 def __setattr__(self, name: str, value: object) -> None:

454 if getattr(self, "_snekql_frozen", False):

455 msg = "table models are immutable"

456 raise FrozenModelError(msg)

457 super().__setattr__(name, value)

458

459 def __repr__(self) -> str:

460 state = self._snekql_state_name()

461 field_reprs: list[str] = []

462 for name in self.__class__.__snekql_columns__:

463 value = getattr(self, name)

464 if value is MISSING:

465 continue

466 field_reprs.append(f"{name}={value!r}")

467 fields = ", ".join(field_reprs)

468 return f"{self.__class__.__name__}[{state}]({fields})"

469

470 def __eq__(self, other: object) -> bool:

471 if self.__class__ is not other.__class__:

472 return False

473 other_model = cast("Model[Any, Any]", other)

474 for name in self.__class__.__snekql_columns__:

475 if getattr(self, name) != getattr(other_model, name):

476 return False

477 return True

478

479 def __hash__(self) -> int:

480 msg = f"unhashable type: {self.__class__.__name__!r}"

481 raise TypeError(msg)

482

483 def _snekql_state_name(self) -> str:

484 storage = cast(

485 "dict[str, object]",

486 object.__getattribute__(self, "__dict__"),

487 )

488 state = storage.get("_snekql_state", "Pending")

489 return cast("str", state)

490

491 def _snekql_to_row(self) -> dict[str, object]:

492 """Encode this model's present values for SQLite storage."""

493

494 _, row = encode_model_row(self, backend="sqlite")

495 return row

496

497 @classmethod

498 def _snekql_from_row(cls, row: Mapping[str, object]) -> Self:

499 """Materialize a fetched model from SQLite storage values."""

500

501 return cast("Self", decode_model_row(cls, row, backend="sqlite"))

502

503 @classmethod

504 def __read_type__(cls) -> type[ReadModelT]:

505 return cast("type[ReadModelT]", cls)

506

507

508def require_model_columns(

509 model: type[Table[Any]],

510) -> dict[str, Attr[Any, Any, Any, Any, Any]]:

511 """Return frozen snekql column metadata for a table model."""

512

513 columns = getattr(model, "__snekql_columns__", None)

514 if not isinstance(columns, dict):

515 msg = "schema setup requires snekql table models"

516 raise ModelDeclarationError(msg)

517 return cast("dict[str, Attr[Any, Any, Any, Any, Any]]", columns)

518

519

520def require_model_table_name(model: type[Table[Any]]) -> str:

521 """Return the resolved SQLite table name for a table model."""

522

523 table_name = getattr(model, "__tablename__", None)

524 if not isinstance(table_name, str):

525 msg = "schema setup requires snekql table models"

526 raise ModelDeclarationError(msg)

527 return table_name

528

529

530def require_model_backend(model: type[Table[Any]]) -> BackendFamily:

531 """Return the backend family declared by a table model."""

532

533 backend = getattr(model, "__snekql_backend__", None)

534 if backend not in {"mariadb", "sqlite"}:

535 msg = "schema setup requires snekql table models"

536 raise ModelDeclarationError(msg)

537 return cast("BackendFamily", backend)

538

539

540def encode_column_value(

541 column: Attr[Any, Any, Any, Any, Any],

542 value: object,

543 *,

544 backend: BackendFamily = "sqlite",

545) -> object:

546 """Encode one logical model value through a backend-specific column codec."""

547

548 from snekql._model_materialization import ( # noqa: PLC0415

549 encode_column_value as encode_value,

550 )

551

552 return encode_value(column, value, backend=backend)

553

554

555def decode_column_value(

556 column: Attr[Any, Any, Any, Any, Any],

557 value: object,

558 *,

559 backend: BackendFamily = "sqlite",

560) -> object:

561 """Decode one database value through a backend-specific column codec."""

562

563 from snekql._model_materialization import ( # noqa: PLC0415

564 decode_column_value as decode_value,

565 )

566

567 return decode_value(column, value, backend=backend)

568

569

570def decode_model_row(

571 model: type[Table[Any]],

572 row: Mapping[str, object],

573 *,

574 backend: BackendFamily = "sqlite",

575) -> Table[Any]:

576 """Decode backend row values into a fetched table model instance."""

577

578 from snekql._model_materialization import ( # noqa: PLC0415

579 decode_model_row as decode_row,

580 )

581

582 return cast("Table[Any]", decode_row(model, row, backend=backend))

583

584

585def encode_model_row(

586 row: object,

587 *,

588 backend: BackendFamily = "sqlite",

589) -> tuple[type[Table[Any]], dict[str, object]]:

590 """Encode a pending model into table metadata and backend row values."""

591

592 from snekql._model_materialization import ( # noqa: PLC0415

593 encode_model_row as encode_row,

594 )

595

596 return cast(

597 "tuple[type[Table[Any]], dict[str, object]]", encode_row(row, backend=backend)

598 )