Coverage for src/paperap/const.py: 100%

235 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-03-22 16:02 -0400

1""" 

2---------------------------------------------------------------------------- 

3 

4 METADATA: 

5 

6 File: const.py 

7 Project: paperap 

8 Created: 2025-03-04 

9 Version: 0.0.9 

10 Author: Jess Mann 

11 Email: jess@jmann.me 

12 Copyright (c) 2025 Jess Mann 

13 

14---------------------------------------------------------------------------- 

15 

16 LAST MODIFIED: 

17 

18 2025-03-04 By Jess Mann 

19 

20""" 

21 

22from __future__ import annotations 

23 

24import logging 

25from datetime import datetime 

26from enum import Enum, IntEnum, StrEnum 

27from string import Template 

28from typing import ( 

29 Any, 

30 Iterator, 

31 Literal, 

32 NotRequired, 

33 Protocol, 

34 Required, 

35 Self, 

36 TypeAlias, 

37 TypedDict, 

38 override, 

39 runtime_checkable, 

40) 

41 

42import pydantic 

43from pydantic import ConfigDict, Field 

44 

45logger = logging.getLogger(__name__) 

46 

47 

48class StrEnumWithUnknown(StrEnum): 

49 @override 

50 @classmethod 

51 def _missing_(cls, value: object) -> str: 

52 logger.debug("Handling unknown enum value", extra={"enum_class": cls.__name__, "value": value}) 

53 return cls.UNKNOWN # type: ignore # subclasses will define unknown 

54 

55 

56class IntEnumWithUnknown(IntEnum): 

57 @override 

58 @classmethod 

59 def _missing_(cls, value: object) -> int: 

60 logger.debug("Handling unknown enum value", extra={"enum_class": cls.__name__, "value": value}) 

61 return cls.UNKNOWN # type: ignore # subclasses will define unknown 

62 

63 

64class ConstModel(pydantic.BaseModel): 

65 model_config = ConfigDict( 

66 from_attributes=True, 

67 extra="forbid", 

68 use_enum_values=True, 

69 validate_default=True, 

70 validate_assignment=True, 

71 ) 

72 

73 @override 

74 def __eq__(self, other: Any) -> bool: 

75 if isinstance(other, dict): 

76 # Ensure the dictionary keys match the model fields 

77 expected_keys = set(self.model_fields.keys()) 

78 if set(other.keys()) != expected_keys: 

79 return False 

80 return all(getattr(self, key) == other.get(key) for key in expected_keys) 

81 

82 # This check probably isn't necessary before calling super (TODO?) 

83 if isinstance(other, self.__class__): 

84 # Compare all fields of the model 

85 return self.model_dump() == other.model_dump() 

86 

87 return super().__eq__(other) 

88 

89 

90class URLS: 

91 # May be deprecated in the future. Used for reference currently. 

92 index: Template = Template("/api/") 

93 token: Template = Template("/api/token/") 

94 list: Template = Template("/api/${resource}/") 

95 detail: Template = Template("/api/${resource}/${pk}/") 

96 create: Template = Template("/api/${resource}/") 

97 update: Template = Template("/api/${resource}/${pk}/") 

98 delete: Template = Template("/api/${resource}/${pk}/") 

99 meta: Template = Template("/api/document/${pk}/metadata/") 

100 next_asn: Template = Template("/api/document/next_asn/") 

101 notes: Template = Template("/api/document/${pk}/notes/") 

102 post: Template = Template("/api/documents/post_document/") 

103 single: Template = Template("/api/document/${pk}/") 

104 suggestions: Template = Template("/api/${resource}/${pk}/suggestions/") 

105 preview: Template = Template("/api/${resource}/${pk}/preview/") 

106 thumbnail: Template = Template("/api/${resource}/${pk}/thumb/") 

107 download: Template = Template("/api/${resource}/${pk}/download/") 

108 

109 

110CommonEndpoints: TypeAlias = Literal["list", "detail", "create", "update", "delete"] 

111Endpoints: TypeAlias = dict[CommonEndpoints | str, Template] 

112 

113 

114class FilteringStrategies(StrEnum): 

115 WHITELIST = "whitelist" 

116 BLACKLIST = "blacklist" 

117 ALLOW_ALL = "allow_all" 

118 ALLOW_NONE = "allow_none" 

119 

120 

121class ModelStatus(StrEnum): 

122 INITIALIZING = "initializing" 

123 UPDATING = "updating" 

124 SAVING = "saving" 

125 READY = "ready" 

126 ERROR = "error" 

127 

128 

129class CustomFieldTypes(StrEnumWithUnknown): 

130 STRING = "string" 

131 BOOLEAN = "boolean" 

132 INTEGER = "integer" 

133 FLOAT = "float" 

134 MONETARY = "monetary" 

135 DATE = "date" 

136 URL = "url" 

137 DOCUMENT_LINK = "documentlink" 

138 UNKNOWN = "unknown" 

139 

140 

141class CustomFieldValues(ConstModel): 

142 field: int 

143 value: Any 

144 

145 

146class CustomFieldTypedDict(TypedDict): 

147 field: int 

148 value: Any 

149 

150 

151# Possibly not used after refactoring 

152class DocumentMetadataType(ConstModel): 

153 namespace: str | None = None 

154 prefix: str | None = None 

155 key: str | None = None 

156 value: str | None = None 

157 

158 

159class DocumentSearchHitType(ConstModel): 

160 score: float | None = None 

161 highlights: str | None = None 

162 note_highlights: str | None = None 

163 rank: int | None = None 

164 

165 

166class MatchingAlgorithmType(IntEnumWithUnknown): 

167 NONE = 0 

168 ANY = 1 

169 ALL = 2 

170 LITERAL = 3 

171 REGEX = 4 

172 FUZZY = 5 

173 AUTO = 6 

174 UNKNOWN = -1 

175 

176 @override 

177 @classmethod 

178 def _missing_(cls, value: object) -> "Literal[MatchingAlgorithmType.UNKNOWN]": 

179 logger.debug("Handling unknown enum value", extra={"enum_class": cls.__name__, "value": value}) 

180 return cls.UNKNOWN 

181 

182 

183class PermissionSetType(ConstModel): 

184 users: list[int] = Field(default_factory=list) 

185 groups: list[int] = Field(default_factory=list) 

186 

187 

188class PermissionTableType(ConstModel): 

189 view: PermissionSetType = Field(default_factory=PermissionSetType) 

190 change: PermissionSetType = Field(default_factory=PermissionSetType) 

191 

192 

193class RetrieveFileMode(StrEnum): 

194 DOWNLOAD = "download" 

195 PREVIEW = "preview" 

196 THUMBNAIL = "thumb" 

197 

198 

199class SavedViewFilterRuleType(ConstModel): 

200 rule_type: int 

201 value: str | None = None 

202 saved_view: int | None = None 

203 

204 

205class ShareLinkFileVersionType(StrEnumWithUnknown): 

206 ARCHIVE = "archive" 

207 ORIGINAL = "original" 

208 UNKNOWN = "unknown" 

209 

210 @override 

211 @classmethod 

212 def _missing_(cls, value: object) -> "Literal[ShareLinkFileVersionType.UNKNOWN]": 

213 logger.debug("Handling unknown enum value", extra={"enum_class": cls.__name__, "value": value}) 

214 return cls.UNKNOWN 

215 

216 

217class StatusType(StrEnumWithUnknown): 

218 OK = "OK" 

219 ERROR = "ERROR" 

220 UNKNOWN = "UNKNOWN" 

221 

222 @override 

223 @classmethod 

224 def _missing_(cls, value: object) -> "Literal[StatusType.UNKNOWN]": 

225 logger.debug("Handling unknown enum value", extra={"enum_class": cls.__name__, "value": value}) 

226 return cls.UNKNOWN 

227 

228 

229class StatusDatabaseMigrationStatusType(ConstModel): 

230 latest_migration: str | None = None 

231 unapplied_migrations: list[str] = Field(default_factory=list) 

232 

233 

234class StatusDatabaseType(ConstModel): 

235 type: str | None = None 

236 url: str | None = None 

237 status: StatusType | None = None 

238 error: str | None = None 

239 migration_status: StatusDatabaseMigrationStatusType | None = None 

240 

241 

242class StatusStorageType(ConstModel): 

243 total: int | None = None 

244 available: int | None = None 

245 

246 

247class StatusTasksType(ConstModel): 

248 redis_url: str | None = None 

249 redis_status: StatusType | None = None 

250 redis_error: str | None = None 

251 celery_status: StatusType | None = None 

252 index_status: StatusType | None = None 

253 index_last_modified: datetime | None = None 

254 index_error: str | None = None 

255 classifier_status: StatusType | None = None 

256 classifier_last_trained: datetime | None = None 

257 classifier_error: str | None = None 

258 

259 

260class TaskStatusType(StrEnumWithUnknown): 

261 PENDING = "PENDING" 

262 STARTED = "STARTED" 

263 SUCCESS = "SUCCESS" 

264 FAILURE = "FAILURE" 

265 UNKNOWN = "UNKNOWN" 

266 

267 

268class TaskTypeType(StrEnumWithUnknown): 

269 AUTO = "auto_task" 

270 SCHEDULED_TASK = "scheduled_task" 

271 MANUAL_TASK = "manual_task" 

272 UNKNOWN = "unknown" 

273 

274 

275class WorkflowActionType(IntEnumWithUnknown): 

276 ASSIGNMENT = 1 

277 REMOVAL = 2 

278 EMAIL = 3 

279 WEBHOOK = 4 

280 UNKNOWN = -1 

281 

282 

283class WorkflowTriggerType(IntEnumWithUnknown): 

284 CONSUMPTION = 1 

285 DOCUMENT_ADDED = 2 

286 DOCUMENT_UPDATED = 3 

287 UNKNOWN = -1 

288 

289 

290class WorkflowTriggerSourceType(IntEnumWithUnknown): 

291 CONSUME_FOLDER = 1 

292 API_UPLOAD = 2 

293 MAIL_FETCH = 3 

294 UNKNOWN = -1 

295 

296 

297class WorkflowTriggerMatchingType(IntEnumWithUnknown): 

298 NONE = 0 

299 ANY = 1 

300 ALL = 2 

301 LITERAL = 3 

302 REGEX = 4 

303 FUZZY = 5 

304 UNKNOWN = -1 

305 

306 

307class ScheduleDateFieldType(StrEnumWithUnknown): 

308 ADDED = "added" 

309 CREATED = "created" 

310 MODIFIED = "modified" 

311 CUSTOM_FIELD = "custom_field" 

312 UNKNOWN = "unknown" 

313 

314 

315class WorkflowTriggerScheduleDateFieldType(StrEnumWithUnknown): 

316 ADDED = "added" 

317 CREATED = "created" 

318 MODIFIED = "modified" 

319 CUSTOM_FIELD = "custom_field" 

320 UNKNOWN = "unknown" 

321 

322 

323class SavedViewDisplayModeType(StrEnumWithUnknown): 

324 TABLE = "table" 

325 SMALL_CARDS = "smallCards" 

326 LARGE_CARDS = "largeCards" 

327 UNKNOWN = "unknown" 

328 

329 

330class SavedViewDisplayFieldType(StrEnumWithUnknown): 

331 TITLE = "title" 

332 CREATED = "created" 

333 ADDED = "added" 

334 TAGS = "tag" 

335 CORRESPONDENT = "correspondent" 

336 DOCUMENT_TYPE = "documenttype" 

337 STORAGE_PATH = "storagepath" 

338 NOTES = "note" 

339 OWNER = "owner" 

340 SHARED = "shared" 

341 ASN = "asn" 

342 PAGE_COUNT = "pagecount" 

343 CUSTOM_FIELD = "custom_field_%d" 

344 UNKNOWN = "unknown" 

345 

346 

347class DocumentStorageType(StrEnumWithUnknown): 

348 UNENCRYPTED = "unencrypted" 

349 GPG = "gpg" 

350 UNKNOWN = "unknown" 

351 

352 

353class TaskNameType(StrEnumWithUnknown): 

354 CONSUME_FILE = "consume_file" 

355 TRAIN_CLASSIFIER = "train_classifier" 

356 CHECK_SANITY = "check_sanity" 

357 INDEX_OPTIMIZE = "index_optimize" 

358 UNKNOWN = "unknown"