Coverage for src/nendo/library/model.py: 98%

1# -*- encoding: utf-8 -*- 

2# ruff: noqa: A003 

3"""Module containing all SQLAlchemy ORM models for the nendo core schema. 

4 

5Used by the SQLAlchemy implementation of the nendo library. 

6""" 

7 

8from __future__ import annotations 

9 

10import json 

11import uuid 

12from datetime import date, datetime 

13 

14import numpy as np 

15from sqlalchemy import Column, DateTime, ForeignKey, Integer, MetaData, String 

16from sqlalchemy.dialects.postgresql import ENUM, JSON, UUID 

17from sqlalchemy.orm import declarative_base, relationship 

18from sqlalchemy.sql import func 

19from sqlalchemy.sql.sqltypes import Text 

20from sqlalchemy.types import TypeDecorator 

21from sqlalchemy_json import NestedMutableDict, NestedMutableList, mutable_json_type 

22 

23from nendo import schema 

24 

25Base = declarative_base(metadata=MetaData()) 

26 

27 

28def convert(obj): 

29 if isinstance(obj, uuid.UUID): 

30 return str(obj) 

31 if isinstance(obj, (datetime, date)): 

32 return obj.isoformat() 

33 if isinstance(obj, set): 

34 return convert(list(obj)) 

35 if isinstance(obj, np.float32): 

36 return float(obj) 

37 if isinstance(obj, NestedMutableList) or isinstance(obj, list): 

38 return [convert(x) for x in obj] 

39 if isinstance(obj, NestedMutableDict) or isinstance(obj, dict): 

40 return {k: convert(v) for k, v in obj.items()} 

41 return obj 

42 

43 

44class JSONEncodedDict(TypeDecorator): 

45 impl = JSON 

46 

47 def process_bind_param(self, value, dialect): 

48 return json.dumps(convert(value)) 

49 

50 def process_result_value(self, value, dialect): 

51 if value is not None: 

52 return json.loads(value) 

53 

54 

55class TrackTrackRelationshipDB(Base): 

56 __tablename__ = "track_track_relationships" 

57 id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) 

58 source_id = Column(UUID(as_uuid=True), ForeignKey("tracks.id")) 

59 target_id = Column(UUID(as_uuid=True), ForeignKey("tracks.id")) 

60 created_at = Column(DateTime(timezone=True), server_default=func.now()) 

61 updated_at = Column( 

62 DateTime(timezone=True), server_default=func.now(), onupdate=func.now() 

63 ) 

64 relationship_type = Column(String) 

65 meta = Column(mutable_json_type(dbtype=JSONEncodedDict, nested=True)) 

66 

67 # relationships 

68 source = relationship( 

69 "NendoTrackDB", 

70 foreign_keys=[source_id], 

71 back_populates="related_tracks", 

72 ) 

73 target = relationship("NendoTrackDB", foreign_keys=[target_id]) 

74 

75 

76class TrackCollectionRelationshipDB(Base): 

77 __tablename__ = "track_collection_relationships" 

78 id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) 

79 source_id = Column(UUID(as_uuid=True), ForeignKey("tracks.id")) 

80 target_id = Column(UUID(as_uuid=True), ForeignKey("collections.id")) 

81 created_at = Column(DateTime(timezone=True), server_default=func.now()) 

82 updated_at = Column( 

83 DateTime(timezone=True), 

84 server_default=func.now(), 

85 onupdate=func.now(), 

86 ) 

87 relationship_type = Column(String) 

88 meta = Column(mutable_json_type(dbtype=JSONEncodedDict, nested=True)) 

89 relationship_position = Column(Integer, nullable=False) 

90 

91 # relationships 

92 source = relationship("NendoTrackDB", foreign_keys=[source_id]) 

93 target = relationship("NendoCollectionDB", foreign_keys=[target_id]) 

94 

95 

96class CollectionCollectionRelationshipDB(Base): 

97 __tablename__ = "collection_collection_relationships" 

98 id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) 

99 source_id = Column(UUID(as_uuid=True), ForeignKey("collections.id")) 

100 target_id = Column(UUID(as_uuid=True), ForeignKey("collections.id")) 

101 created_at = Column(DateTime(timezone=True), server_default=func.now()) 

102 updated_at = Column( 

103 DateTime(timezone=True), 

104 server_default=func.now(), 

105 onupdate=func.now(), 

106 ) 

107 relationship_type = Column(String) 

108 meta = Column(mutable_json_type(dbtype=JSONEncodedDict, nested=True)) 

109 

110 # relationships 

111 source = relationship("NendoCollectionDB", foreign_keys=[source_id]) 

112 target = relationship("NendoCollectionDB", foreign_keys=[target_id]) 

113 

114 

115class NendoPluginDataDB(Base): 

116 __tablename__ = "plugin_data" 

117 

118 id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) 

119 track_id = Column(UUID(as_uuid=True), ForeignKey("tracks.id")) 

120 user_id = Column(UUID(as_uuid=True)) # , ForeignKey("users.id")) 

121 plugin_name = Column(String, nullable=False) 

122 plugin_version = Column(String, nullable=False) 

123 key = Column(String, nullable=False) 

124 value = Column(String, nullable=True) 

125 created_at = Column(DateTime(timezone=True), server_default=func.now()) 

126 updated_at = Column( 

127 DateTime(timezone=True), 

128 server_default=func.now(), 

129 onupdate=func.now(), 

130 ) 

131 

132 # relationships 

133 # track = relationship("NendoTrackDB", backref="plugin_data") 

134 

135 

136class NendoTrackDB(Base): 

137 __tablename__ = "tracks" 

138 

139 id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) 

140 user_id = Column(UUID(as_uuid=True)) 

141 track_type = Column(String, default="track") 

142 visibility = Column(ENUM(schema.Visibility), default="private") 

143 updated_at = Column( 

144 DateTime(timezone=True), default=func.now(), onupdate=func.now() 

145 ) 

146 created_at = Column(DateTime(timezone=True), default=func.now()) 

147 images = Column(mutable_json_type(dbtype=JSONEncodedDict, nested=True)) 

148 resource = Column(mutable_json_type(dbtype=JSONEncodedDict, nested=True)) 

149 meta = Column(mutable_json_type(dbtype=JSONEncodedDict, nested=True)) 

150 

151 # Relationships 

152 related_tracks = relationship( 

153 "TrackTrackRelationshipDB", 

154 primaryjoin="NendoTrackDB.id==TrackTrackRelationshipDB.source_id", 

155 foreign_keys="[TrackTrackRelationshipDB.source_id]", 

156 back_populates="source", 

157 # cascade="all, delete-orphan", 

158 ) 

159 related_collections = relationship( 

160 "TrackCollectionRelationshipDB", 

161 primaryjoin="NendoTrackDB.id==TrackCollectionRelationshipDB.source_id", 

162 foreign_keys="[TrackCollectionRelationshipDB.source_id]", 

163 back_populates="source", 

164 # cascade="all, delete-orphan", 

165 ) 

166 plugin_data = relationship( 

167 "NendoPluginDataDB", 

168 primaryjoin="NendoTrackDB.id==NendoPluginDataDB.track_id", 

169 foreign_keys="[NendoPluginDataDB.track_id]", 

170 # cascade="all, delete-orphan", 

171 ) 

172 

173 

174class NendoBlobDB(Base): 

175 __tablename__ = "blobs" 

176 

177 id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) 

178 user_id = Column(UUID(as_uuid=True)) 

179 visibility = Column(ENUM(schema.Visibility), default="private") 

180 updated_at = Column( 

181 DateTime(timezone=True), 

182 default=func.now(), 

183 onupdate=func.now(), 

184 ) 

185 created_at = Column(DateTime(timezone=True), default=func.now()) 

186 resource = Column(mutable_json_type(dbtype=JSONEncodedDict, nested=True)) 

187 

188 

189class NendoCollectionDB(Base): 

190 __tablename__ = "collections" 

191 

192 id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) 

193 name = Column(String) 

194 user_id = Column(UUID(as_uuid=True)) 

195 description = Column(Text, default="") 

196 collection_type = Column(String, default="generic") 

197 visibility = Column(ENUM(schema.Visibility), default="private") 

198 updated_at = Column( 

199 DateTime(timezone=True), 

200 default=func.now(), 

201 onupdate=func.now(), 

202 ) 

203 created_at = Column(DateTime(timezone=True), default=func.now()) 

204 meta = Column(mutable_json_type(dbtype=JSONEncodedDict, nested=True)) 

205 

206 # relationships 

207 related_tracks = relationship( 

208 "TrackCollectionRelationshipDB", 

209 primaryjoin="NendoCollectionDB.id==TrackCollectionRelationshipDB.target_id", 

210 foreign_keys="[TrackCollectionRelationshipDB.target_id]", 

211 back_populates="target", 

212 # cascade="all, delete-orphan", 

213 ) 

214 

215 related_collections = relationship( 

216 "CollectionCollectionRelationshipDB", 

217 primaryjoin="NendoCollectionDB.id==CollectionCollectionRelationshipDB.source_id", 

218 foreign_keys="[CollectionCollectionRelationshipDB.source_id]", 

219 back_populates="source", 

220 # cascade="all, delete-orphan", 

221 )