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

51 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2023-06-07 06:35 +0000

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 impl = types.BINARY(16) 

33 

34 python_type = uuid.UUID 

35 

36 cache_ok = True 

37 

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

39 """ 

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

41 """ 

42 self.binary = binary 

43 self.native = native 

44 

45 def __repr__(self): 

46 return util.generic_repr(self) 

47 

48 def load_dialect_impl(self, dialect): 

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

50 # Use the native UUID type. 

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

52 

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

54 # Use the native UNIQUEIDENTIFIER type. 

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

56 

57 else: 

58 # Fallback to either a BINARY or a CHAR. 

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

60 return dialect.type_descriptor(kind) 

61 

62 @staticmethod 

63 def _coerce(value): 

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

65 try: 

66 value = uuid.UUID(value) 

67 

68 except (TypeError, ValueError): 

69 value = uuid.UUID(bytes=value) 

70 

71 return value 

72 

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

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

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

76 def process_literal_param(self, value, dialect): 

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

78 else: 

79 def process_literal_param(self, value, dialect): 

80 return value 

81 

82 def process_bind_param(self, value, dialect): 

83 if value is None: 

84 return value 

85 

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

87 value = self._coerce(value) 

88 

89 if self.native and dialect.name in ( 

90 'postgresql', 

91 'mssql', 

92 'cockroachdb' 

93 ): 

94 return str(value) 

95 

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

97 

98 def process_result_value(self, value, dialect): 

99 if value is None: 

100 return value 

101 

102 if self.native and dialect.name in ( 

103 'postgresql', 

104 'mssql', 

105 'cockroachdb' 

106 ): 

107 if isinstance(value, uuid.UUID): 

108 # Some drivers convert PostgreSQL's uuid values to 

109 # Python's uuid.UUID objects by themselves 

110 return value 

111 return uuid.UUID(value) 

112 

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