Coverage for snekql/storage.py: 77%
356 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-07 21:13 +0300
1"""SQLite storage declarations and value codecs for table models."""
3from __future__ import annotations
5from collections.abc import Callable
6from dataclasses import dataclass
7from datetime import UTC, datetime
8from json import JSONDecodeError, dumps, loads
9from types import EllipsisType
10from typing import Any, Literal, Self, TypeVar, cast, overload
12from snekql.errors import (
13 FrozenModelError,
14 ModelDeclarationError,
15 ModelValidationError,
16 QueryConstructionError,
17 SnekqlError,
18)
19from snekql.expressions import Assignment, OrderBy, Predicate
21type SQLiteStorageClass = Literal["INTEGER", "REAL", "TEXT", "BLOB"]
24@dataclass(frozen=True, kw_only=True)
25class AttrConfig:
26 """Constructor bundle for column descriptors.
28 Keeping descriptor configuration in one value avoids long internal
29 constructors while preserving explicit storage metadata at each call site.
30 """
32 sqlite_storage_class: SQLiteStorageClass
33 storage_type_name: str
34 auto_increment: bool = False
35 default: object = ...
36 default_factory: Callable[[], object] | EllipsisType = ...
37 nullable: bool | None = None
38 primary_key: bool = False
39 server_default: object | None = None
40 unique: bool = False
43WriteOwnerT = TypeVar("WriteOwnerT")
44LoadedOwnerT = TypeVar("LoadedOwnerT")
45OwnerT = TypeVar("OwnerT")
46WriteT = TypeVar("WriteT")
47ReadValueT = TypeVar("ReadValueT")
50class Missing:
51 """Sentinel type for generated values that are not available yet.
53 >>> MISSING is Missing()
54 True
55 >>> repr(MISSING)
56 'MISSING'
57 """
59 _instance: Self | None = None
61 def __new__(cls) -> Self:
62 if cls._instance is None:
63 cls._instance = super().__new__(cls)
64 return cls._instance
66 def __repr__(self) -> str:
67 return "MISSING"
70MISSING = Missing()
73# Startup schema verification behavior: strict raises on drift, warn logs and
74# continues. TypeAliasType currently exposes a read-only generic __doc__.
75type SchemaPolicy = Literal["strict", "warn"]
78def build_attr(config: AttrConfig) -> Any:
79 """Build a public column descriptor from normalized storage metadata.
81 Storage classes differ mostly by SQLite class and logical type name. Keeping
82 descriptor wiring here gives the storage declaration module one place to
83 change how declaration metadata becomes query/model behavior.
84 """
86 return Attr[Any, Any, Any, Any, Any](config)
89class Integer:
90 """SQLite INTEGER column declaration for table model fields.
92 >>> class User[S = Pending](Model[S, "User[Fetched]"]):
93 ... id: User.GenCol[int] = Integer(primary_key=True, default=MISSING)
94 """
96 def __new__( # noqa: PLR0913
97 cls,
98 *,
99 primary_key: bool = False,
100 auto_increment: bool = False,
101 nullable: bool | None = None,
102 unique: bool = False,
103 default: object = ...,
104 default_factory: Callable[[], object] | EllipsisType = ...,
105 ) -> Any:
106 return build_attr(
107 AttrConfig(
108 auto_increment=auto_increment,
109 default=default,
110 default_factory=default_factory,
111 nullable=nullable,
112 primary_key=primary_key,
113 unique=unique,
114 sqlite_storage_class="INTEGER",
115 storage_type_name="Integer",
116 ),
117 )
120class Real:
121 """SQLite REAL column declaration for float-like model values.
123 >>> class Reading[S = Pending](Model[S, "Reading[Fetched]"]):
124 ... value: Reading.Col[float] = Real(nullable=False)
125 """
127 def __new__(
128 cls,
129 *,
130 primary_key: bool = False,
131 nullable: bool | None = None,
132 unique: bool = False,
133 default: object = ...,
134 default_factory: Callable[[], object] | EllipsisType = ...,
135 ) -> Any:
136 return build_attr(
137 AttrConfig(
138 default=default,
139 default_factory=default_factory,
140 nullable=nullable,
141 primary_key=primary_key,
142 unique=unique,
143 sqlite_storage_class="REAL",
144 storage_type_name="Real",
145 ),
146 )
149class Text:
150 """SQLite TEXT column declaration for string model values.
152 >>> class User[S = Pending](Model[S, "User[Fetched]"]):
153 ... email: User.Col[str] = Text(nullable=False)
154 """
156 def __new__(
157 cls,
158 *,
159 primary_key: bool = False,
160 nullable: bool | None = None,
161 unique: bool = False,
162 default: object = ...,
163 default_factory: Callable[[], object] | EllipsisType = ...,
164 ) -> Any:
165 return build_attr(
166 AttrConfig(
167 default=default,
168 default_factory=default_factory,
169 nullable=nullable,
170 primary_key=primary_key,
171 unique=unique,
172 sqlite_storage_class="TEXT",
173 storage_type_name="Text",
174 ),
175 )
178class Blob:
179 """SQLite BLOB column declaration for bytes model values.
181 >>> class File[S = Pending](Model[S, "File[Fetched]"]):
182 ... content: File.Col[bytes] = Blob(nullable=False)
183 """
185 def __new__(
186 cls,
187 *,
188 primary_key: bool = False,
189 nullable: bool | None = None,
190 unique: bool = False,
191 default: object = ...,
192 default_factory: Callable[[], object] | EllipsisType = ...,
193 ) -> Any:
194 return build_attr(
195 AttrConfig(
196 default=default,
197 default_factory=default_factory,
198 nullable=nullable,
199 primary_key=primary_key,
200 unique=unique,
201 sqlite_storage_class="BLOB",
202 storage_type_name="Blob",
203 ),
204 )
207class Json:
208 """SQLite TEXT-backed JSON column declaration.
210 Values are serialized to JSON text before writes and decoded before fetched
211 model validation.
213 >>> class Event[S = Pending](Model[S, "Event[Fetched]"]):
214 ... payload: Event.Col[dict[str, object]] = Json(nullable=False)
215 """
217 def __new__(
218 cls,
219 *,
220 nullable: bool | None = None,
221 unique: bool = False,
222 default: object = ...,
223 default_factory: Callable[[], object] | EllipsisType = ...,
224 ) -> Any:
225 return build_attr(
226 AttrConfig(
227 default=default,
228 default_factory=default_factory,
229 nullable=nullable,
230 unique=unique,
231 sqlite_storage_class="TEXT",
232 storage_type_name="Json",
233 ),
234 )
237class Boolean:
238 """SQLite INTEGER-backed boolean column declaration.
240 >>> class FeatureFlag[S = Pending](Model[S, "FeatureFlag[Fetched]"]):
241 ... enabled: FeatureFlag.Col[bool] = Boolean(default=False)
242 """
244 def __new__(
245 cls,
246 *,
247 nullable: bool | None = None,
248 unique: bool = False,
249 default: object = ...,
250 default_factory: Callable[[], object] | EllipsisType = ...,
251 ) -> Any:
252 return build_attr(
253 AttrConfig(
254 default=default,
255 default_factory=default_factory,
256 nullable=nullable,
257 unique=unique,
258 sqlite_storage_class="INTEGER",
259 storage_type_name="Boolean",
260 ),
261 )
264class DateTime:
265 """SQLite TEXT-backed timezone-aware datetime column declaration.
267 >>> class Event[S = Pending](Model[S, "Event[Fetched]"]):
268 ... created_at: Event.GenCol[datetime] = DateTime(
269 ... server_default=CurrentTimestamp(),
270 ... default=MISSING,
271 ... )
272 """
274 def __new__(
275 cls,
276 *,
277 server_default: object | None = None,
278 nullable: bool | None = None,
279 unique: bool = False,
280 default: object = ...,
281 default_factory: Callable[[], object] | EllipsisType = ...,
282 ) -> Any:
283 return build_attr(
284 AttrConfig(
285 default=default,
286 default_factory=default_factory,
287 nullable=nullable,
288 server_default=server_default,
289 unique=unique,
290 sqlite_storage_class="TEXT",
291 storage_type_name="DateTime",
292 ),
293 )
296class CurrentTimestamp:
297 """Server default marker for database-filled UTC timestamps.
299 >>> DateTime(server_default=CurrentTimestamp(), default=MISSING)
300 """
303class Attr[WriteOwnerT, LoadedOwnerT, OwnerT, WriteT, ReadValueT]:
304 """Typed model column descriptor used for fields and query construction.
306 The descriptor exposes pending-state write values on application-created
307 models, fetched-state read values on runtime materialized models, and query
308 helper methods on the model class.
309 """
311 def __init__(self, config: AttrConfig) -> None:
312 self.auto_increment: bool = config.auto_increment
313 self.default: object = config.default
314 self.default_factory: Callable[[], object] | EllipsisType = (
315 config.default_factory
316 )
317 self.is_generated: bool = False
318 self.name: str | None = None
319 self.owner: type[object] | None = None
320 self.nullable: bool | None = config.nullable
321 self.primary_key: bool = config.primary_key
322 self.server_default: object | None = config.server_default
323 self.sqlite_storage_class: SQLiteStorageClass = config.sqlite_storage_class
324 self.storage_type_name: str = config.storage_type_name
325 self.unique: bool = config.unique
327 def __set_name__(self, owner: type[object], name: str) -> None:
328 self.name = name
329 self.owner = owner
331 @overload
332 def __get__(
333 self, instance: None, owner: type[Any]
334 ) -> Attr[WriteOwnerT, LoadedOwnerT, OwnerT, WriteT, ReadValueT]: ...
335 @overload
336 def __get__(self, instance: WriteOwnerT, owner: type[Any]) -> WriteT: ...
337 @overload
338 def __get__(self, instance: LoadedOwnerT, owner: type[Any]) -> ReadValueT: ...
339 def __get__(self, instance: object | None, owner: type[Any]) -> object:
340 if instance is None:
341 return self
342 storage = cast(
343 "dict[str, object]",
344 object.__getattribute__(instance, "__dict__"),
345 )
346 return storage[self._require_name()]
348 def __set__(self, instance: object, value: WriteT) -> None:
349 if getattr(instance, "_snekql_frozen", False):
350 msg = "table models are immutable"
351 raise FrozenModelError(msg)
352 storage = cast(
353 "dict[str, object]",
354 object.__getattribute__(instance, "__dict__"),
355 )
356 storage[self._require_name()] = value
358 def build_default(self) -> object:
359 if not isinstance(self.default_factory, EllipsisType):
360 return self.default_factory()
361 return self.default
363 def decode_sqlite(self, value: object) -> object:
364 """Decode a SQLite value to its logical Python value."""
366 return self._decode_backend_value(value, backend="SQLite")
368 def decode_mariadb(self, value: object) -> object:
369 """Decode a MariaDB value to its logical Python value."""
371 return self._decode_backend_value(value, backend="MariaDB")
373 def encode_sqlite(self, value: object) -> object:
374 """Encode a logical Python value for SQLite storage."""
376 return self._encode_backend_value(value, backend="SQLite")
378 def encode_mariadb(self, value: object) -> object:
379 """Encode a logical Python value for MariaDB storage."""
381 return self._encode_backend_value(value, backend="MariaDB")
383 def _decode_backend_value(self, value: object, *, backend: str) -> object:
384 """Decode database values using backend-specific storage codecs."""
386 try:
387 if backend == "MariaDB":
388 decoded_value = self._decode_mariadb(value)
389 else:
390 decoded_value = self._decode_sqlite(value)
391 return self._coerce_logical_value(decoded_value, fetched=True)
392 except SnekqlError:
393 raise
394 except Exception as error:
395 msg = f"invalid database value for {self._require_name()!r}"
396 raise ModelValidationError(
397 msg,
398 ) from error
400 def _encode_backend_value(self, value: object, *, backend: str) -> object:
401 """Encode model values using backend-specific storage codecs."""
403 try:
404 logical_value = self._coerce_logical_value(value, fetched=False)
405 if logical_value is MISSING:
406 return MISSING
407 if backend == "MariaDB":
408 return self._encode_mariadb(logical_value)
409 return self._encode_sqlite(logical_value)
410 except SnekqlError:
411 raise
412 except Exception as error:
413 msg = f"invalid model value for {self._require_name()!r}"
414 raise ModelValidationError(
415 msg,
416 ) from error
418 def validate_model_value(self, value: object) -> object:
419 """Validate and normalize a pending model value."""
421 try:
422 return self._coerce_logical_value(value, fetched=False)
423 except SnekqlError:
424 raise
425 except Exception as error:
426 msg = f"invalid model value for {self._require_name()!r}"
427 raise ModelValidationError(
428 msg,
429 ) from error
431 def _require_name(self) -> str:
432 if self.name is None:
433 msg = "column descriptor is not bound"
434 raise ModelDeclarationError(msg)
435 return self.name
437 def _coerce_logical_value(self, value: object, *, fetched: bool) -> object:
438 if value is MISSING:
439 return self._coerce_missing_value(fetched=fetched)
440 if value is None:
441 return self._coerce_null_value()
442 coercers: dict[str, Callable[[object], object]] = {
443 "Blob": self._coerce_blob_value,
444 "Boolean": self._coerce_boolean_value,
445 "DateTime": self._coerce_datetime_value,
446 "Integer": self._coerce_integer_value,
447 "Json": self._coerce_json_value,
448 "Real": self._coerce_real_value,
449 "Text": self._coerce_text_value,
450 }
451 try:
452 coercer = coercers[self.storage_type_name]
453 except KeyError as error:
454 msg = f"unknown storage type {self.storage_type_name!r}"
455 raise ModelDeclarationError(msg) from error
456 return coercer(value)
458 def _coerce_missing_value(self, *, fetched: bool) -> Missing:
459 if self.is_generated and not fetched:
460 return MISSING
461 msg = f"missing generated value for {self._require_name()!r}"
462 raise ModelValidationError(msg)
464 def _coerce_null_value(self) -> None:
465 if self.nullable is False:
466 msg = f"{self._require_name()!r} cannot be null"
467 raise ModelValidationError(msg)
469 def _coerce_blob_value(self, value: object) -> bytes:
470 if not isinstance(value, bytes):
471 msg = f"{self._require_name()!r} must be bytes"
472 raise ModelValidationError(msg)
473 return value
475 def _coerce_boolean_value(self, value: object) -> bool:
476 if type(value) is not bool:
477 msg = f"{self._require_name()!r} must be a bool"
478 raise ModelValidationError(msg)
479 return value
481 def _coerce_datetime_value(self, value: object) -> datetime:
482 if not isinstance(value, datetime):
483 msg = f"{self._require_name()!r} must be a datetime"
484 raise ModelValidationError(msg)
485 return self._normalize_datetime(value)
487 def _coerce_integer_value(self, value: object) -> int:
488 if type(value) is not int:
489 msg = f"{self._require_name()!r} must be an int"
490 raise ModelValidationError(msg)
491 return value
493 def _coerce_json_value(self, value: object) -> object:
494 try:
495 _ = dumps(value, separators=(",", ":"))
496 except (TypeError, ValueError) as error:
497 msg = f"{self._require_name()!r} is not JSON serializable"
498 raise ModelValidationError(msg) from error
499 return value
501 def _coerce_real_value(self, value: object) -> float:
502 if isinstance(value, bool) or not isinstance(value, int | float):
503 msg = f"{self._require_name()!r} must be a number"
504 raise ModelValidationError(msg)
505 return float(value)
507 def _coerce_text_value(self, value: object) -> str:
508 if not isinstance(value, str):
509 msg = f"{self._require_name()!r} must be a str"
510 raise ModelValidationError(msg)
511 return value
513 def _decode_sqlite(self, value: object) -> object:
514 if value is None:
515 return None
516 if self.storage_type_name == "Json":
517 if not isinstance(value, str):
518 msg = f"{self._require_name()!r} database value must be JSON text"
519 raise ModelValidationError(
520 msg,
521 )
522 try:
523 return loads(value)
524 except JSONDecodeError as error:
525 msg = f"{self._require_name()!r} database value is not valid JSON"
526 raise ModelValidationError(
527 msg,
528 ) from error
529 if self.storage_type_name == "Boolean":
530 if value == 0:
531 return False
532 if value == 1:
533 return True
534 msg = f"{self._require_name()!r} database value must be 0 or 1"
535 raise ModelValidationError(
536 msg,
537 )
538 if self.storage_type_name == "DateTime":
539 if not isinstance(value, str):
540 msg = f"{self._require_name()!r} database value must be timestamp text"
541 raise ModelValidationError(
542 msg,
543 )
544 return self._decode_datetime_text(value)
545 return value
547 def _decode_mariadb(self, value: object) -> object:
548 if value is None:
549 return None
550 decoders: dict[str, Callable[[object], object]] = {
551 "Boolean": self._decode_mariadb_boolean,
552 "DateTime": self._decode_mariadb_datetime,
553 "Json": self._decode_mariadb_json,
554 }
555 decoder = decoders.get(self.storage_type_name)
556 if decoder is None:
557 return value
558 return decoder(value)
560 def _decode_mariadb_json(self, value: object) -> object:
561 if not isinstance(value, str | bytes | bytearray):
562 msg = f"{self._require_name()!r} database value must be JSON text"
563 raise ModelValidationError(msg)
564 json_text = value.decode() if isinstance(value, bytes | bytearray) else value
565 try:
566 return loads(json_text)
567 except JSONDecodeError as error:
568 msg = f"{self._require_name()!r} database value is not valid JSON"
569 raise ModelValidationError(msg) from error
571 def _decode_mariadb_boolean(self, value: object) -> bool:
572 if value == 0:
573 return False
574 if value == 1:
575 return True
576 msg = f"{self._require_name()!r} database value must be 0 or 1"
577 raise ModelValidationError(msg)
579 def _decode_mariadb_datetime(self, value: object) -> datetime:
580 if isinstance(value, datetime):
581 if value.tzinfo is None:
582 return value.replace(tzinfo=UTC)
583 return value
584 if isinstance(value, str):
585 return self._decode_mariadb_datetime_text(value)
586 msg = f"{self._require_name()!r} database value must be a datetime"
587 raise ModelValidationError(msg)
589 def _decode_datetime_text(self, value: str) -> datetime:
590 if not value.endswith("Z"):
591 msg = f"{self._require_name()!r} timestamp must end with Z"
592 raise ModelValidationError(
593 msg,
594 )
595 try:
596 parsed = datetime.fromisoformat(f"{value[:-1]}+00:00")
597 except ValueError as error:
598 msg = f"{self._require_name()!r} timestamp is not valid ISO text"
599 raise ModelValidationError(
600 msg,
601 ) from error
602 return self._normalize_datetime(parsed)
604 def _decode_mariadb_datetime_text(self, value: str) -> datetime:
605 try:
606 parsed = datetime.fromisoformat(value.replace(" ", "T"))
607 except ValueError as error:
608 msg = f"{self._require_name()!r} timestamp is not valid MariaDB text"
609 raise ModelValidationError(msg) from error
610 if parsed.tzinfo is None:
611 parsed = parsed.replace(tzinfo=UTC)
612 return self._normalize_datetime(parsed)
614 def _encode_sqlite(self, value: object) -> object:
615 if value is None:
616 return None
617 if self.storage_type_name == "Json":
618 try:
619 return dumps(value, separators=(",", ":"))
620 except (TypeError, ValueError) as error:
621 msg = f"{self._require_name()!r} is not JSON serializable"
622 raise ModelValidationError(
623 msg,
624 ) from error
625 if self.storage_type_name == "Boolean":
626 return 1 if value else 0
627 if self.storage_type_name == "DateTime":
628 timestamp = cast("datetime", value)
629 return (
630 timestamp.strftime("%Y-%m-%dT%H:%M:%S.")
631 + f"{timestamp.microsecond // 1000:03d}Z"
632 )
633 return value
635 def _encode_mariadb(self, value: object) -> object:
636 if value is None:
637 return None
638 if self.storage_type_name == "Json":
639 try:
640 return dumps(value, separators=(",", ":"))
641 except (TypeError, ValueError) as error:
642 msg = f"{self._require_name()!r} is not JSON serializable"
643 raise ModelValidationError(msg) from error
644 if self.storage_type_name == "Boolean":
645 return 1 if value else 0
646 if self.storage_type_name == "DateTime":
647 timestamp = cast("datetime", value)
648 return (
649 timestamp.strftime("%Y-%m-%d %H:%M:%S.")
650 + f"{timestamp.microsecond // 1000:03d}"
651 )
652 return value
654 def _normalize_datetime(self, value: datetime) -> datetime:
655 if value.tzinfo is None or value.utcoffset() is None:
656 msg = f"{self._require_name()!r} must be timezone-aware"
657 raise ModelValidationError(
658 msg,
659 )
660 utc_value = value.astimezone(UTC)
661 milliseconds = utc_value.microsecond // 1000
662 return utc_value.replace(microsecond=milliseconds * 1000)
664 def eq(self, value: ReadValueT) -> Predicate[OwnerT]:
665 if value is None:
666 msg = "eq(None) is invalid; use is_null()"
667 raise QueryConstructionError(msg)
668 return Predicate(kind="eq", column=self, value=value)
670 def ne(self, value: ReadValueT) -> Predicate[OwnerT]:
671 if value is None:
672 msg = "ne(None) is invalid; use is_not_null()"
673 raise QueryConstructionError(msg)
674 return Predicate(kind="ne", column=self, value=value)
676 def is_null(self) -> Predicate[OwnerT]:
677 return Predicate(kind="is_null", column=self)
679 def is_not_null(self) -> Predicate[OwnerT]:
680 return Predicate(kind="is_not_null", column=self)
682 def in_(self, *values: ReadValueT) -> Predicate[OwnerT]:
683 if not values:
684 msg = "in_() requires at least one value"
685 raise QueryConstructionError(msg)
686 if any(candidate is None for candidate in values):
687 msg = "in_() values cannot be None"
688 raise QueryConstructionError(msg)
689 return Predicate(kind="in", column=self, values=values)
691 def not_in(self, *values: ReadValueT) -> Predicate[OwnerT]:
692 if not values:
693 msg = "not_in() requires at least one value"
694 raise QueryConstructionError(msg)
695 if any(candidate is None for candidate in values):
696 msg = "not_in() values cannot be None"
697 raise QueryConstructionError(msg)
698 return Predicate(kind="not_in", column=self, values=values)
700 def like(self, pattern: str) -> Predicate[OwnerT]:
701 if self.storage_type_name != "Text":
702 msg = "like() is only valid for text columns"
703 raise QueryConstructionError(msg)
704 return Predicate(kind="like", column=self, value=pattern)
706 def not_like(self, pattern: str) -> Predicate[OwnerT]:
707 if self.storage_type_name != "Text":
708 msg = "not_like() is only valid for text columns"
709 raise QueryConstructionError(msg)
710 return Predicate(kind="not_like", column=self, value=pattern)
712 def asc(self) -> OrderBy[OwnerT]:
713 return OrderBy(column=self, direction="ASC")
715 def desc(self) -> OrderBy[OwnerT]:
716 return OrderBy(column=self, direction="DESC")
718 def to(self, value: ReadValueT) -> Assignment[OwnerT]:
719 return Assignment(column=self, value=value)