""" AUTHOR: Khushal P Soonderji DATE: Thursday, 12th Dec., 2024. OBJECTIVE: To define how information about files will be stored in the database. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For making data behaviour_models: from pydantic import BaseModel, Field, field_validator, PastDatetime, AwareDatetime from typing import Optional, Literal, Union, Any, Dict, List # My utils: from utils_v2.string import regex from utils_v2.date_time import date_time # To work with MongoDB: from bson.objectid import ObjectId # Other data models: from models.data.core.user import CoreUserInfoModel # To work with date and time: import datetime # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** class CoreFileInfoModel(BaseModel): fileId: ObjectId = Field( description = "the id of the document in mongodb that holds this information", frozen = False, default = None, alias = "_id" ) user: CoreUserInfoModel = Field( description = "to identify the user who owns this file", frozen = True ) filename: str = Field( description = "the name of this file", frozen = True ) uploadTs: AwareDatetime = Field( description = "the time (utc) at which this file was uploaded", frozen = True ) length: int | None = Field( description = "to note the size of the file in bytes", frozen = False, default = None ) hash: str | None = Field( description = "a simple hash to verify the integrity of the uploaded data; irrelevant for dirs", frozen = True, default = None ) metadata: Dict[str, Any] = Field( description = "any additional data about this file to filter it later", frozen = False, default = {} ) tags: List[str] = Field( description = "a list of keywords to apply to this file to filter it later", frozen = False, default = [], examples = ["Bank Statement", "PDF", "bhopli@orange.com"] ) isPrivate: bool = Field( description = "whether this file is a private file, or publicly available", frozen = True ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "allow" arbitrary_types_allowed = True def model_dump(self, *args, **kwargs): return super().model_dump(*args, by_alias = True, **kwargs) # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ @field_validator("uploadTs", mode = "before") def parse_date_time(cls, value): return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC) @field_validator("metadata", mode = "before") def validate_metadata(cls, value): if value is None: value = {} return value @field_validator("tags", mode = "before") def validate_tags(cls, value): if value is None: value = [] return value @field_validator("fileId", mode = "before") def parse_oid(cls, value): try: if isinstance(value, str): value = ObjectId(value) except: pass return value # ┏┓ • # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏ # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛ # ┛ @property def summary(self): return { "user": self.user.model_dump(), "name": self.name, "createTs": self.createTs, "parentId": self.parentId, "isPrivate": self.isPrivate, } # --------------------------------------------------------------------------------------------------------------------- class CoreFileAccessResponseModel(BaseModel): success: bool = Field( description = "whether, or not, the operation was successful", frozen = False, default = False ) message: str | None = Field( description = "useful to note the reason in case of an unsuccessful operation", frozen = False, default = "message not documented" ) data: Any = Field( description = "the result of the operation", frozen = False, default = None ) exception: Exception | None = Field( description = "to document any exceptions that came up", frozen = False, default = None ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "allow" arbitrary_types_allowed = True # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": from utils_v2.string import json file_info = CoreFileInfoModel( _id = "67519cf3a7804fcbc6f12452", user = CoreUserInfoModel( fullName = "Bhopli Narangi", userId = 1, entityId = 2, billingAccountId = 3, departmentId = 4, branchId = 5, industry = "technology" ), filename = "Graduation Certificate.png", uploadTs = date_time.get_current_utc_date_time(as_string = False), length = 1024, hash = "abcdefgh12345678", metadata = { "camera": "iPhone 1000 Pro Max XS" }, tags = [ "important" ], parentId = None, isPrivate = True ) print("FILE INFO MODEL:", json.to_string(file_info.model_dump(), default = str))