Coverage for src/cstlcore/ydocs/models.py: 95%

1import uuid 

2from datetime import datetime 

3from typing import TYPE_CHECKING, List, Optional 

4 

5from pydantic import BaseModel 

6from sqlalchemy import CheckConstraint 

7from sqlalchemy.dialects.postgresql import JSONB 

8from sqlmodel import ( 

9 TIMESTAMP, 

10 Boolean, 

11 Column, 

12 FetchedValue, 

13 Field, 

14 LargeBinary, 

15 Relationship, 

16 SQLModel, 

17 UniqueConstraint, 

18 func, 

19 text, 

20) 

21 

22if TYPE_CHECKING: 

23 from cstlcore.collections.models import Collection 

24 from cstlcore.maps.models import MapLayer 

25 from cstlcore.users.models import User 

26 

27# TODO: Add DocType for elements, notes, etc... To be defined 

28 

29 

30class YDoc(SQLModel, table=True): 

31 ## Table args and constraints 

32 __table_args__ = ( 

33 UniqueConstraint( 

34 "collection_id", "parent_id", "name", name="uq_ydoc_parent_name" 

35 ), 

36 CheckConstraint( 

37 "(content IS NULL AND size IS NULL AND checksum IS NULL) OR " 

38 "(content IS NOT NULL AND size IS NOT NULL AND checksum IS NOT NULL)", 

39 name="ck_ydoc_content_size_checksum", 

40 ), 

41 CheckConstraint( 

42 "is_folder = TRUE AND content IS NULL AND size IS NULL AND checksum IS NULL OR " 

43 "is_folder = FALSE", 

44 name="ck_ydoc_folder_content", 

45 ), 

46 ) 

47 

48 ## Primary key and foreign keys 

49 id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) 

50 owner_id: uuid.UUID | None = Field(default=None, foreign_key="user.id", index=True) 

51 collection_id: uuid.UUID = Field( 

52 foreign_key="collection.id", index=True, ondelete="CASCADE" 

53 ) 

54 name: str = Field(index=True, nullable=False) 

55 parent_id: uuid.UUID | None = Field( 

56 default=None, 

57 foreign_key="ydoc.id", 

58 index=True, 

59 ondelete="CASCADE", 

60 ) 

61 is_folder: bool = Field( 

62 default=False, sa_column=Column(Boolean(), server_default=text("FALSE")) 

63 ) 

64 

65 ## file content and metadata 

66 size: int | None = Field(default=None) 

67 checksum: str | None = Field(default=None) 

68 metadata_: dict | None = Field( 

69 default=None, 

70 sa_column=Column(name="metadata", type_=JSONB), 

71 # alias="metadata", 

72 schema_extra={"serialization_alias": "metadata"}, 

73 ) 

74 content: bytes | None = Field(default=None, sa_type=LargeBinary) 

75 

76 ## Timestamps 

77 

78 created_at: datetime | None = Field( 

79 default=None, 

80 sa_column=Column( 

81 TIMESTAMP(timezone=True), server_default=func.now(), nullable=False 

82 ), 

83 ) 

84 updated_at: datetime | None = Field( 

85 default=None, 

86 sa_column=Column( 

87 TIMESTAMP(timezone=True), 

88 server_default=func.now(), 

89 server_onupdate=FetchedValue(), 

90 nullable=False, 

91 ), 

92 ) 

93 

94 ## Relationships 

95 

96 owner: "User" = Relationship(back_populates="owned_files") 

97 collection: "Collection" = Relationship(back_populates="files") 

98 layers: List["MapLayer"] = Relationship( 

99 back_populates="ydoc", 

100 cascade_delete=True, 

101 ) 

102 # Parent/children relationships for hierarchical YDocs (folders/files) 

103 parent: Optional["YDoc"] = Relationship( 

104 back_populates="children", 

105 sa_relationship_kwargs=dict(remote_side="YDoc.id", single_parent=True), 

106 ) 

107 children: List["YDoc"] = Relationship( 

108 back_populates="parent", 

109 sa_relationship_kwargs=dict( 

110 cascade="all, delete-orphan", 

111 ), 

112 ) 

113 

114 

115class YDocPublic(BaseModel): 

116 id: uuid.UUID 

117 name: str 

118 path: str | None = None 

119 owner_id: uuid.UUID 

120 collection_id: uuid.UUID 

121 parent_id: uuid.UUID | None = None 

122 is_folder: bool = False 

123 size: int | None = 0 

124 checksum: str | None = None 

125 metadata_: dict | None = Field(default=None, alias="metadata") 

126 created_at: datetime 

127 updated_at: datetime 

128 

129 

130class YDocCreate(BaseModel): 

131 name: str 

132 is_folder: bool = False 

133 parent_id: uuid.UUID | None = None 

134 # collection_id: uuid.UUID 

135 metadata_: dict | None = Field(default=None, alias="metadata") 

136 

137 

138class YDocUpdate(BaseModel): 

139 name: str | None = None 

140 parent_id: uuid.UUID | None = None 

141 # collection_id: uuid.UUID | None = None 

142 metadata_: dict | None = Field(default=None, alias="metadata") 

143 

144 

145class YDocContentUpdate(BaseModel): 

146 content: str | None = Field( 

147 None, description="Base64 encoded content of the YDoc file" 

148 ) 

149 

150 

151class YDocContentPublic(BaseModel): 

152 content: str | None = Field( 

153 None, description="Base64 encoded content of the YDoc file" 

154 ) 

155 size: int | None = None 

156 checksum: str | None = None