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

1import uuid 

2 

3from sqlalchemy import types, util 

4from sqlalchemy.dialects import mssql, postgresql 

5 

6from ..compat import get_sqlalchemy_version 

7from .scalar_coercible import ScalarCoercible 

8 

9sqlalchemy_version = get_sqlalchemy_version() 

10 

11 

12class UUIDType(ScalarCoercible, types.TypeDecorator): 

13 """ 

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

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

16 

17 :: 

18 

19 from sqlalchemy_utils import UUIDType 

20 import uuid 

21 

22 class User(Base): 

23 __tablename__ = 'user' 

24 

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

26 id = sa.Column( 

27 UUIDType(binary=False), 

28 primary_key=True, 

29 default=uuid.uuid4 

30 ) 

31 """ 

32 

33 impl = types.BINARY(16) 

34 

35 python_type = uuid.UUID 

36 

37 cache_ok = True 

38 

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

40 """ 

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

42 """ 

43 self.binary = binary 

44 self.native = native 

45 

46 def __repr__(self): 

47 return util.generic_repr(self) 

48 

49 def load_dialect_impl(self, dialect): 

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

51 # Use the native UUID type. 

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

53 

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

55 # Use the native UNIQUEIDENTIFIER type. 

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

57 

58 else: 

59 # Fallback to either a BINARY or a CHAR. 

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

61 return dialect.type_descriptor(kind) 

62 

63 @staticmethod 

64 def _coerce(value): 

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

66 try: 

67 value = uuid.UUID(value) 

68 

69 except (TypeError, ValueError): 

70 value = uuid.UUID(bytes=value) 

71 

72 return value 

73 

74 # sqlalchemy >= 1.4.30 quotes UUID's automatically. 

75 # It is only necessary to quote UUID's in sqlalchemy < 1.4.30. 

76 if sqlalchemy_version < (1, 4, 30): 

77 

78 def process_literal_param(self, value, dialect): 

79 return f"'{value}'" if value else value 

80 else: 

81 

82 def process_literal_param(self, value, dialect): 

83 return value 

84 

85 def process_bind_param(self, value, dialect): 

86 if value is None: 

87 return value 

88 

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

90 value = self._coerce(value) 

91 

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

93 return str(value) 

94 

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

96 

97 def process_result_value(self, value, dialect): 

98 if value is None: 

99 return value 

100 

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

102 if isinstance(value, uuid.UUID): 

103 # Some drivers convert PostgreSQL's uuid values to 

104 # Python's uuid.UUID objects by themselves 

105 return value 

106 return uuid.UUID(value) 

107 

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