Coverage for snekql/indexes.py: 90%

1"""Public index declarations and normalized model index metadata.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import Any, TypeVar, cast, overload 

7 

8from snekql.errors import ModelDeclarationError 

9from snekql.storage import Attr 

10 

11OwnerT = TypeVar("OwnerT") 

12 

13 

14@dataclass(frozen=True) 

15class Index[OwnerT]: 

16 """Public table-level index declaration. 

17 

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

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

20 ... __indexes__ = [Index(email)] 

21 """ 

22 

23 columns: tuple[Attr[Any, Any, OwnerT, Any, Any], ...] 

24 unique: bool 

25 name: str | None 

26 

27 @overload 

28 def __init__( 

29 self, 

30 *columns: Attr[Any, Any, OwnerT, Any, Any], 

31 unique: bool = False, 

32 name: str | None = None, 

33 ) -> None: ... 

34 

35 @overload 

36 def __init__( 

37 self, 

38 *columns: object, 

39 unique: bool = False, 

40 name: str | None = None, 

41 ) -> None: ... 

42 

43 def __init__( 

44 self, 

45 *columns: object, 

46 unique: bool = False, 

47 name: str | None = None, 

48 ) -> None: 

49 if not columns: 

50 msg = "Index() requires at least one column" 

51 raise ModelDeclarationError(msg) 

52 seen_column_ids: set[int] = set() 

53 for column in columns: 

54 if not isinstance(column, Attr): 

55 msg = "Index() arguments must be snekql column descriptors" 

56 raise ModelDeclarationError(msg) 

57 attr_column = cast("Attr[Any, Any, OwnerT, Any, Any]", column) 

58 column_id = id(attr_column) 

59 if column_id in seen_column_ids: 

60 msg = "Index() cannot repeat a column" 

61 raise ModelDeclarationError(msg) 

62 seen_column_ids.add(column_id) 

63 object.__setattr__( 

64 self, 

65 "columns", 

66 cast("tuple[Attr[Any, Any, OwnerT, Any, Any], ...]", columns), 

67 ) 

68 object.__setattr__(self, "unique", unique) 

69 object.__setattr__(self, "name", name) 

70 

71 

72@dataclass(frozen=True, kw_only=True) 

73class NormalizedIndex: 

74 """Immutable model-owned index metadata used by schema startup.""" 

75 

76 column_names: tuple[str, ...] 

77 name: str 

78 unique: bool 

79 

80 

81def require_index_declaration(value: object) -> Index[Any]: 

82 """Validate a public __indexes__ entry before normalization.""" 

83 

84 if not isinstance(value, Index): 

85 msg = "__indexes__ entries must be Index declarations" 

86 raise ModelDeclarationError(msg) 

87 return cast("Index[Any]", value)