Coverage for snekql/_model_materialization.py: 89%

1"""Shared Table Model row encoding and fetched-row materialization.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Mapping 

6from typing import Any, Literal, cast 

7 

8from snekql.errors import ( 

9 ModelDeclarationError, 

10 ModelValidationError, 

11 QueryConstructionError, 

12) 

13from snekql.storage import MISSING, Attr 

14 

15type StorageBackend = Literal["mariadb", "sqlite"] 

16 

17 

18def _require_model_columns( 

19 model: type[object], 

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

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

22 if not isinstance(columns, dict): 

23 msg = "schema setup requires snekql table models" 

24 raise ModelDeclarationError(msg) 

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

26 

27 

28def _require_insert_model(row: object) -> type[object]: 

29 model_class = row.__class__ 

30 try: 

31 _ = _require_model_columns(model_class) 

32 except ModelDeclarationError as error: 

33 msg = "insert requires a snekql model instance" 

34 raise QueryConstructionError(msg) from error 

35 return model_class 

36 

37 

38def encode_column_value( 

39 column: Attr[Any, Any, Any, Any, Any], 

40 value: object, 

41 *, 

42 backend: StorageBackend, 

43) -> object: 

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

45 

46 if backend == "mariadb": 

47 return column.encode_mariadb(value) 

48 return column.encode_sqlite(value) 

49 

50 

51def decode_column_value( 

52 column: Attr[Any, Any, Any, Any, Any], 

53 value: object, 

54 *, 

55 backend: StorageBackend, 

56) -> object: 

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

58 

59 if backend == "mariadb": 

60 return column.decode_mariadb(value) 

61 return column.decode_sqlite(value) 

62 

63 

64def encode_model_row( 

65 row: object, 

66 *, 

67 backend: StorageBackend, 

68) -> tuple[type[object], dict[str, object]]: 

69 """Encode a Pending Model into table metadata and backend row values.""" 

70 

71 model_class = _require_insert_model(row) 

72 encoded_row: dict[str, object] = {} 

73 for name, column in _require_model_columns(model_class).items(): 

74 value = getattr(row, name) 

75 if value is MISSING: 

76 continue 

77 encoded_row[name] = encode_column_value(column, value, backend=backend) 

78 return model_class, encoded_row 

79 

80 

81def decode_model_row( 

82 model: type[object], 

83 row: Mapping[str, object], 

84 *, 

85 backend: StorageBackend, 

86) -> object: 

87 """Materialize a Fetched Model from backend row values.""" 

88 

89 remaining_values = dict(row) 

90 model_instance = object.__new__(model) 

91 storage = cast( 

92 "dict[str, object]", 

93 object.__getattribute__(model_instance, "__dict__"), 

94 ) 

95 storage["_snekql_frozen"] = False 

96 storage["_snekql_state"] = "Fetched" 

97 for name, column in _require_model_columns(model).items(): 

98 if name not in remaining_values: 

99 msg = f"missing database value for {name!r}" 

100 raise ModelValidationError(msg) 

101 value = decode_column_value( 

102 column, 

103 remaining_values.pop(name), 

104 backend=backend, 

105 ) 

106 setattr(model_instance, name, value) 

107 if remaining_values: 

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

109 msg = f"unknown database values: {names}" 

110 raise ModelValidationError(msg) 

111 storage["_snekql_frozen"] = True 

112 return model_instance