Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy_utils/types/uuid.py: 70%

1import uuid 

2 

3from sqlalchemy import types, util 

4from sqlalchemy.dialects import mssql, postgresql 

5 

6from .scalar_coercible import ScalarCoercible 

7 

8 

9class UUIDType(ScalarCoercible, types.TypeDecorator): 

10 """ 

11 Stores a UUID in the database natively when it can and falls back to 

12 a BINARY(16) or a CHAR(32) when it can't. 

13 

14 :: 

15 

16 from sqlalchemy_utils import UUIDType 

17 import uuid 

18 

19 class User(Base): 

20 __tablename__ = 'user' 

21 

22 # Pass `binary=False` to fallback to CHAR instead of BINARY 

23 id = sa.Column( 

24 UUIDType(binary=False), 

25 primary_key=True, 

26 default=uuid.uuid4 

27 ) 

28 """ 

29 

30 impl = types.BINARY(16) 

31 

32 python_type = uuid.UUID 

33 

34 cache_ok = True 

35 

36 def __init__(self, binary=True, native=True): 

37 """ 

38 :param binary: Whether to use a BINARY(16) or CHAR(32) fallback. 

39 """ 

40 self.binary = binary 

41 self.native = native 

42 

43 def __repr__(self): 

44 return util.generic_repr(self) 

45 

46 def load_dialect_impl(self, dialect): 

47 if self.native and dialect.name in ('postgresql', 'cockroachdb'): 

48 # Use the native UUID type. 

49 return dialect.type_descriptor(postgresql.UUID()) 

50 

51 if dialect.name == 'mssql' and self.native: 

52 # Use the native UNIQUEIDENTIFIER type. 

53 return dialect.type_descriptor(mssql.UNIQUEIDENTIFIER()) 

54 

55 else: 

56 # Fallback to either a BINARY or a CHAR. 

57 kind = self.impl if self.binary else types.CHAR(32) 

58 return dialect.type_descriptor(kind) 

59 

60 @staticmethod 

61 def _coerce(value): 

62 if value and not isinstance(value, uuid.UUID): 

63 try: 

64 value = uuid.UUID(value) 

65 

66 except (TypeError, ValueError): 

67 value = uuid.UUID(bytes=value) 

68 

69 return value 

70 

71 def process_literal_param(self, value, dialect): 

72 return value 

73 

74 def process_bind_param(self, value, dialect): 

75 if value is None: 

76 return value 

77 

78 if not isinstance(value, uuid.UUID): 

79 value = self._coerce(value) 

80 

81 if self.native and dialect.name in ('postgresql', 'mssql', 'cockroachdb'): 

82 return str(value) 

83 

84 return value.bytes if self.binary else value.hex 

85 

86 def process_result_value(self, value, dialect): 

87 if value is None: 

88 return value 

89 

90 if self.native and dialect.name in ('postgresql', 'mssql', 'cockroachdb'): 

91 if isinstance(value, uuid.UUID): 

92 # Some drivers convert PostgreSQL's uuid values to 

93 # Python's uuid.UUID objects by themselves 

94 return value 

95 return uuid.UUID(value) 

96 

97 return uuid.UUID(bytes=value) if self.binary else uuid.UUID(value)