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 PUUIDBase 

7 

8_SEPARATOR_LENGTH = 1 

9_UUID_LENGTH = 36 

10 

11 

12@final 

13class SqlPUUID[TPrefix: str](TypeDecorator[PUUIDBase[TPrefix]]): 

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[PUUIDBase[TPrefix]] 

25 

26 def __init__(self, puuid_cls: type[PUUIDBase[TPrefix]]) -> None: 

27 """ 

28 Initialize the SqlPUUID type. 

29 

30 Parameters 

31 ---------- 

32 puuid_cls : type[PUUIDBase[TPrefix]] 

33 The pUUID class (e.g., `PUUIDv4[Literal["user"]]`) to associate with this 

34 column. 

35 """ 

36 self.puuid_cls = puuid_cls 

37 varchar_length = len(puuid_cls.prefix()) + _SEPARATOR_LENGTH + _UUID_LENGTH 

38 super().__init__(length=varchar_length) 

39 

40 @override 

41 def process_bind_param( 

42 self, value: PUUIDBase[TPrefix] | None, dialect: Dialect 

43 ) -> str | None: 

44 if value is None: 

45 return None 

46 return value.to_string() 

47 

48 @override 

49 def process_result_value( 

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

51 ) -> PUUIDBase[TPrefix] | None: 

52 if value is None: 

53 return None 

54 return self.puuid_cls.from_string(value)