Coverage for dormatory/models/dormatory_model.py: 100%

1""" 

2DORMATORY SQLAlchemy Models 

3 

4This module defines the core models for storing structured hierarchical data 

5using a flat set of tables as shown in the ERD. 

6""" 

7 

8from datetime import datetime, UTC 

9from typing import Optional, List 

10from uuid import UUID, uuid4 

11 

12from sqlalchemy import ( 

13 Column, Integer, String, Text, DateTime, ForeignKey, 

14 create_engine, MetaData 

15) 

16from sqlalchemy.orm import declarative_base, relationship, sessionmaker, Session 

17from sqlalchemy.dialects.postgresql import UUID as PostgresUUID 

18 

19# Create declarative base 

20Base = declarative_base() 

21 

22 

23class Type(Base): 

24 """ 

25 Defines different categories or types for the object entities. 

26  

27 This table stores the various types that objects can have, providing 

28 a way to categorize and organize the hierarchical data structure. 

29 """ 

30 __tablename__ = "type" 

31 

32 id = Column(PostgresUUID(as_uuid=True), primary_key=True, default=uuid4) 

33 type_name = Column(Text, nullable=False) 

34 

35 # Relationships 

36 objects = relationship("Object", back_populates="type") 

37 

38 def __repr__(self) -> str: 

39 return f"<Type(id={self.id}, type_name='{self.type_name}')>" 

40 

41 

42class Object(Base): 

43 """ 

44 Core table representing the main data entities in the system. 

45  

46 This is the central entity that represents objects in the hierarchical 

47 data structure. Each object has a type and can participate in various 

48 relationships through the link table. 

49 """ 

50 __tablename__ = "object" 

51 

52 id = Column(Integer, primary_key=True, autoincrement=True) 

53 name = Column(String, nullable=False) 

54 version = Column(Integer, nullable=False, default=1) 

55 type_id = Column(PostgresUUID(as_uuid=True), ForeignKey("type.id"), nullable=False) 

56 created_on = Column(Text, nullable=False) 

57 created_by = Column(Text, nullable=False) 

58 

59 # Relationships 

60 type = relationship("Type", back_populates="objects") 

61 permissions = relationship("Permissions", back_populates="object") 

62 versioning_records = relationship("Versioning", back_populates="object") 

63 attributes = relationship("Attributes", back_populates="object") 

64 

65 # Self-referencing relationships through link table 

66 parent_links = relationship( 

67 "Link", 

68 foreign_keys="Link.child_id", 

69 back_populates="child" 

70 ) 

71 child_links = relationship( 

72 "Link", 

73 foreign_keys="Link.parent_id", 

74 back_populates="parent" 

75 ) 

76 

77 def __repr__(self) -> str: 

78 return f"<Object(id={self.id}, name='{self.name}', type_id={self.type_id})>" 

79 

80 

81class Link(Base): 

82 """ 

83 Manages hierarchical parent-child relationships between object entities. 

84  

85 This table allows for a flexible, flat representation of tree-like structures 

86 by establishing parent-child relationships between objects. An object can be 

87 both a parent and child in different relationships. 

88 """ 

89 __tablename__ = "link" 

90 

91 id = Column(Integer, primary_key=True, autoincrement=True) 

92 parent_id = Column(Integer, ForeignKey("object.id"), nullable=False) 

93 parent_type = Column(String, nullable=False) 

94 child_type = Column(String, nullable=False) 

95 r_name = Column(String, nullable=False) # Relationship name (e.g., "contains", "part_of") 

96 child_id = Column(Integer, ForeignKey("object.id"), nullable=False) 

97 

98 # Relationships 

99 parent = relationship("Object", foreign_keys=[parent_id], back_populates="child_links") 

100 child = relationship("Object", foreign_keys=[child_id], back_populates="parent_links") 

101 

102 def __repr__(self) -> str: 

103 return f"<Link(id={self.id}, parent_id={self.parent_id}, child_id={self.child_id}, r_name='{self.r_name}')>" 

104 

105 

106class Permissions(Base): 

107 """ 

108 Stores access control information for specific object entities. 

109  

110 This table manages permissions for objects, allowing fine-grained access 

111 control based on users and permission levels. 

112 """ 

113 __tablename__ = "permissions" 

114 

115 id = Column(Integer, primary_key=True, autoincrement=True) 

116 object_id = Column(Integer, ForeignKey("object.id"), nullable=False) 

117 user = Column(String, nullable=False) 

118 permission_level = Column(String, nullable=False) 

119 

120 # Relationships 

121 object = relationship("Object", back_populates="permissions") 

122 

123 def __repr__(self) -> str: 

124 return f"<Permissions(id={self.id}, object_id={self.object_id}, user='{self.user}', permission_level='{self.permission_level}')>" 

125 

126 

127class Versioning(Base): 

128 """ 

129 Tracks different historical versions of object entities. 

130  

131 This table maintains version history for objects, allowing tracking of 

132 changes over time with version strings and timestamps. 

133 """ 

134 __tablename__ = "versioning" 

135 

136 id = Column(Integer, primary_key=True, autoincrement=True) 

137 object_id = Column(Integer, ForeignKey("object.id"), nullable=False) 

138 version = Column(String, nullable=False) # Version string/tag 

139 created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC)) 

140 

141 # Relationships 

142 object = relationship("Object", back_populates="versioning_records") 

143 

144 def __repr__(self) -> str: 

145 return f"<Versioning(id={self.id}, object_id={self.object_id}, version='{self.version}')>" 

146 

147 

148class Attributes(Base): 

149 """ 

150 Stores arbitrary key-value attributes associated with object entities. 

151  

152 This table allows for flexible schema extension by storing arbitrary 

153 key-value pairs for objects, enabling dynamic attribute addition. 

154 """ 

155 __tablename__ = "attributes" 

156 

157 id = Column(Integer, primary_key=True, autoincrement=True) 

158 name = Column(String, nullable=False) # Attribute name (e.g., "color", "size") 

159 value = Column(String, nullable=False) # Attribute value (e.g., "red", "large") 

160 object_id = Column(Integer, ForeignKey("object.id"), nullable=False) 

161 created_on = Column(String, nullable=False) 

162 updated_on = Column(String, nullable=False) 

163 

164 # Relationships 

165 object = relationship("Object", back_populates="attributes") 

166 

167 def __repr__(self) -> str: 

168 return f"<Attributes(id={self.id}, object_id={self.object_id}, name='{self.name}', value='{self.value}')>" 

169 

170 

171# Database session management 

172def create_engine_and_session(database_url: str = None): 

173 """ 

174 Create SQLAlchemy engine and session factory. 

175  

176 Args: 

177 database_url: Database connection URL. If None, uses environment or default. 

178  

179 Returns: 

180 Tuple of (engine, SessionLocal) 

181 """ 

182 from .database_config import create_engine_and_session as create_engine_and_session_config 

183 return create_engine_and_session_config(database_url) 

184 

185 

186def create_tables(engine): 

187 """ 

188 Create all tables in the database. 

189  

190 Args: 

191 engine: SQLAlchemy engine instance 

192 """ 

193 Base.metadata.create_all(bind=engine) 

194 

195 

196def get_db_session(SessionLocal) -> Session: 

197 """ 

198 Get a database session. 

199  

200 Args: 

201 SessionLocal: Session factory 

202  

203 Returns: 

204 Database session 

205 """ 

206 db = SessionLocal() 

207 try: 

208 yield db 

209 finally: 

210 db.close()