Coverage for src / puuid / sqlalchemy.py: 100%

1from typing import final, override 

2 

3from sqlalchemy.engine.interfaces import Dialect 

4from sqlalchemy.types import String, TypeDecorator 

5 

6from puuid.base import PUUID 

7 

8_SEPARATOR_LENGTH = 1 

9_UUID_LENGTH = 36 

10 

11 

12@final 

13class SqlPUUID(TypeDecorator[PUUID[str]]): 

14 """ 

15 SQLAlchemy type for storing Prefixed UUIDs. 

16 

17 Maps a `PUUID` instance to a `VARCHAR` column in the database and 

18 reconstructs the specific `PUUID` subclass on retrieval. 

19 """ 

20 

21 impl = String 

22 cache_ok = True 

23 

24 puuid_cls: type[PUUID[str]] 

25 

26 def __init__(self, puuid_cls: type[PUUID[str]], prefix_length: int = 4) -> None: 

27 """ 

28 Initialize the SqlPUUID type. 

29 

30 Parameters 

31 ---------- 

32 puuid_cls : type[PUUID[str]] 

33 The pUUID class (e.g., `UserUUID`) to associate with this column. 

34 prefix_length : int, default 4 

35 The length of the prefix string to calculate the column width. 

36 """ 

37 self.puuid_cls = puuid_cls 

38 varchar_length = prefix_length + _SEPARATOR_LENGTH + _UUID_LENGTH 

39 super().__init__(length=varchar_length) 

40 

41 @override 

42 def process_bind_param( 

43 self, value: PUUID[str] | None, dialect: Dialect 

44 ) -> str | None: 

45 if value is None: 

46 return None 

47 return value.to_string() 

48 

49 @override 

50 def process_result_value( 

51 self, value: str | None, dialect: Dialect 

52 ) -> PUUID[str] | None: 

53 if value is None: 

54 return None 

55 return self.puuid_cls.from_string(value)