266 lines
8.5 KiB
Python
266 lines
8.5 KiB
Python
"""
|
|
|
|
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)) |