Coverage for .tox/coverage/lib/python3.13/site-packages/wuttjamaican/db/util.py: 100%

1# -*- coding: utf-8; -*- 

2################################################################################ 

3# 

4# WuttJamaican -- Base package for Wutta Framework 

5# Copyright © 2023-2025 Lance Edgar 

6# 

7# This file is part of Wutta Framework. 

8# 

9# Wutta Framework is free software: you can redistribute it and/or modify it 

10# under the terms of the GNU General Public License as published by the Free 

11# Software Foundation, either version 3 of the License, or (at your option) any 

12# later version. 

13# 

14# Wutta Framework is distributed in the hope that it will be useful, but 

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for 

17# more details. 

18# 

19# You should have received a copy of the GNU General Public License along with 

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>. 

21# 

22################################################################################ 

23""" 

24Database Utilities 

25""" 

26 

27import uuid as _uuid 

28from importlib.metadata import version 

29 

30from packaging.version import Version 

31import sqlalchemy as sa 

32from sqlalchemy import orm 

33from sqlalchemy.dialects.postgresql import UUID as PGUUID 

34 

35from wuttjamaican.util import make_true_uuid 

36 

37# nb. this convention comes from upstream docs 

38# https://docs.sqlalchemy.org/en/14/core/constraints.html#constraint-naming-conventions 

39naming_convention = { 

40 "ix": "ix_%(column_0_label)s", 

41 "uq": "uq_%(table_name)s_%(column_0_name)s", 

42 "ck": "ck_%(table_name)s_%(constraint_name)s", 

43 "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", 

44 "pk": "pk_%(table_name)s", 

45} 

46 

47 

48SA2 = True # pylint: disable=invalid-name 

49if Version(version("SQLAlchemy")) < Version("2"): # pragma: no cover 

50 SA2 = False # pylint: disable=invalid-name 

51 

52 

53class ModelBase: # pylint: disable=empty-docstring 

54 """ """ 

55 

56 def __iter__(self): 

57 # nb. we override this to allow for `dict(self)` 

58 # nb. this does *not* include association proxy values; 

59 # see also wuttjamaican.util.get_value() 

60 state = sa.inspect(self) 

61 fields = [attr.key for attr in state.attrs] 

62 return iter([(field, getattr(self, field)) for field in fields]) 

63 

64 def __getitem__(self, key): 

65 # nb. we override this to allow for `x = self['field']` 

66 state = sa.inspect(self) 

67 if hasattr(state.attrs, key): 

68 return getattr(self, key) 

69 raise KeyError( 

70 f"{self.__class__.__name__} instance has no attr with key: {key}" 

71 ) 

72 

73 

74class UUID( 

75 sa.types.TypeDecorator 

76): # pylint: disable=abstract-method,too-many-ancestors 

77 """ 

78 Platform-independent UUID type. 

79 

80 Uses PostgreSQL's UUID type, otherwise uses CHAR(32), storing as 

81 stringified hex values. 

82 

83 This type definition is based on example from the `SQLAlchemy 

84 documentation 

85 <https://docs.sqlalchemy.org/en/14/core/custom_types.html#backend-agnostic-guid-type>`_. 

86 """ 

87 

88 impl = sa.CHAR 

89 cache_ok = True 

90 """ """ # nb. suppress sphinx autodoc for cache_ok 

91 

92 def load_dialect_impl(self, dialect): # pylint: disable=empty-docstring 

93 """ """ 

94 if dialect.name == "postgresql": 

95 return dialect.type_descriptor(PGUUID()) 

96 return dialect.type_descriptor(sa.CHAR(32)) 

97 

98 def process_bind_param(self, value, dialect): # pylint: disable=empty-docstring 

99 """ """ 

100 if value is None: 

101 return value 

102 

103 if dialect.name == "postgresql": 

104 return str(value) 

105 

106 if not isinstance(value, _uuid.UUID): 

107 value = _uuid.UUID(value) 

108 

109 # hexstring 

110 return f"{value.int:032x}" 

111 

112 def process_result_value( 

113 self, value, dialect 

114 ): # pylint: disable=unused-argument,empty-docstring 

115 """ """ 

116 if value is None: 

117 return value 

118 if not isinstance(value, _uuid.UUID): 

119 value = _uuid.UUID(value) 

120 return value 

121 

122 

123def uuid_column(*args, **kwargs): 

124 """ 

125 Returns a UUID column for use as a table's primary key. 

126 """ 

127 if not args: 

128 args = (UUID(),) 

129 kwargs.setdefault("primary_key", True) 

130 kwargs.setdefault("nullable", False) 

131 kwargs.setdefault("default", make_true_uuid) 

132 if kwargs["primary_key"]: 

133 kwargs.setdefault("doc", "UUID primary key for the table.") 

134 return sa.Column(*args, **kwargs) 

135 

136 

137def uuid_fk_column(target_column, *args, **kwargs): 

138 """ 

139 Returns a UUID column for use as a foreign key to another table. 

140 

141 :param target_column: Name of the table column on the remote side, 

142 e.g. ``'user.uuid'``. 

143 """ 

144 if not args: 

145 args = (UUID(), sa.ForeignKey(target_column)) 

146 return sa.Column(*args, **kwargs) 

147 

148 

149def make_topo_sortkey(model): 

150 """ 

151 Returns a function suitable for use as a ``key`` kwarg to a 

152 standard Python sorting call. This key function will expect a 

153 single class mapper and return a sequence number associated with 

154 that model. The sequence is determined by SQLAlchemy's 

155 topological table sorting. 

156 

157 :param model: Usually the :term:`app model`, but can be any module 

158 containing model classes. 

159 """ 

160 metadata = model.Base.metadata 

161 tables = {table.name: i for i, table in enumerate(metadata.sorted_tables, 1)} 

162 

163 def sortkey(name): 

164 cls = getattr(model, name) 

165 mapper = orm.class_mapper(cls) 

166 return tuple(tables[t.name] for t in mapper.tables) 

167 

168 return sortkey