Coverage for src / lexigram / contracts / tenancy / migration.py: 0%

1"""Tenant tier migration contracts — copy strategy, value types, and constants.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from typing import Any, Protocol, runtime_checkable 

7 

8TIER_ISOLATION_MAP: dict[str, str] = { 

9 "m1": "row_level", 

10 "m2": "row_level", 

11 "m3": "schema", 

12 "m4": "schema", 

13 "m5": "schema", 

14 "m6": "database", 

15 "m7": "database", 

16 "m8": "database", 

17} 

18 

19ISOLATION_TIER_ORDER: list[str] = ["row_level", "schema", "database"] 

20 

21 

22@dataclass(frozen=True) 

23class SnapshotResult: 

24 """Record counts before and after a copy operation.""" 

25 

26 before_count: int 

27 after_count: int 

28 

29 

30@dataclass(frozen=True) 

31class CopyResult: 

32 """Result of a copy operation carried by the migration copy strategy.""" 

33 

34 records_copied: int 

35 records_failed: int 

36 errors: list[str] = field(default_factory=list) 

37 source_snapshot: SnapshotResult | None = None 

38 target_snapshot: SnapshotResult | None = None 

39 

40 

41@dataclass(frozen=True) 

42class MigrationContext: 

43 """Carries source and target isolation context for a copy strategy.""" 

44 

45 source_tier: str 

46 target_tier: str 

47 source_strategy_name: str 

48 target_strategy_name: str 

49 source_config: dict[str, Any] = field(default_factory=dict) 

50 target_config: dict[str, Any] = field(default_factory=dict) 

51 

52 

53@runtime_checkable 

54class MigrationCopyStrategy(Protocol): 

55 """Copies tenant data from source isolation to target isolation. 

56 

57 Implementations raise ``RuntimeError`` or ``ValueError`` on failure so 

58 that the parent saga triggers compensation. 

59 """ 

60 

61 async def validate(self, tenant_id: str, context: MigrationContext) -> None: 

62 """Verify source and target are compatible. 

63 

64 Raises: 

65 ValueError: If the migration path is invalid. 

66 """ 

67 

68 async def copy(self, tenant_id: str, context: MigrationContext) -> CopyResult: 

69 """Execute the data copy. 

70 

71 Returns: 

72 A ``CopyResult`` summarising the operation. 

73 

74 Raises: 

75 RuntimeError: If the copy fails. 

76 """ 

77 

78 async def rollback(self, tenant_id: str, result: CopyResult) -> None: 

79 """Undo a completed copy operation. 

80 

81 Raises: 

82 RuntimeError: If the rollback fails. 

83 """ 

84 

85 

86__all__ = [ 

87 "ISOLATION_TIER_ORDER", 

88 "TIER_ISOLATION_MAP", 

89 "CopyResult", 

90 "MigrationContext", 

91 "MigrationCopyStrategy", 

92 "SnapshotResult", 

93]