""" AUTHOR: Khushal P Soonderji DATE: Tuesday, 10th Dec., 2024 OBJECTIVE: To define all file-management activities in one place. REFERENCES: N/A DOWNLOADS: N/A """ import io # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys from sqlalchemy.orm.collections import collection sys.path.append(".") sys.path.append("..") # My async utils: from utils_v2.string import json from utils_v2.date_time import date_time from utils_v2.database.async_mysql_v2 import AsyncMySQL from utils_v2.database.async_mongo_v2 import AsyncMongo, AsyncMongoStorage # Base model: from models.behaviour.base import BaseModel # Data models: from models.data.core.user import CoreUserInfoModel from models.data.core.file import CoreFileInfoModel, CoreFileSharingModel, CoreFilePermissionsModel # To work with MongoDB: from bson import ObjectId # To work with datatypes: from typing import Any, Literal # To make deep-copies: import copy # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class FileManagementModel: # Define class-level variables: USER_COLLECTION = "_filesUser" FILE_COLLECTION = "_filesInfo" DIRS_COLLECTION = "_dirsInfo" LOGS_COLLECTION = "_filesLogs" # ┳┓ ┓• # ┣┫┏┓┏┓┏┫┓┏┓┏┓ # ┛┗┗ ┗┻┗┻┗┛┗┗┫ # ┛ async def exists( self, mongo_conn: AsyncMongoStorage, file_id: ObjectId | str ) -> bool: """ To check whether, or not, a particular file's record exists in the database. :param mongo_conn: The instance of the database connection to perform this action. :param file_id: The id of the file to check. :return: True if it exists, else False. """ # Run the query: record = await mongo_conn.find_one( collection = self.FILE_COLLECTION, filter = {"_id": ObjectId(file_id)}, projection = {"_id": True, "user": True} ) # Return the result: if not record: return False else: return True async def info( self, mongo_conn: AsyncMongoStorage, file_id: ObjectId ) -> CoreFileInfoModel | None: """ To get the information about this file. :param mongo_conn: The instance of the database connection to perform this action. :param file_id: The id of the file to check. :return: None if the file doesn't exist, else its summary. """ # Run the query: record = await mongo_conn.find_one( collection = self.FILE_COLLECTION, filter = {"_id": ObjectId(file_id)} ) # return the results: if not record: return None else: return CoreFileInfoModel(**record) async def is_private( self, mongo_conn: AsyncMongoStorage, file_id: ObjectId | str ) -> bool | None: """ To check whether, or not, a particular file is publicly readable. :param mongo_conn: The instance of the database connection to perform this action. :param file_id: The id of the file to check. :return: True if private, else False. None if it doesn't exist at all. """ # Run the query: record = await mongo_conn.find_one( collection = self.FILE_COLLECTION, filter = {"_id": ObjectId(file_id)}, projection = {"_id": False, "isPrivate": True} ) # Return the result: if not record: return None else: return record["isPrivate"] async def is_public( self, mongo_conn: AsyncMongoStorage, file_id: ObjectId | str ) -> bool | None: """ A wrapper around the is_private, method that returns the opposite value. :param mongo_conn: The instance of the database connection to perform this action. :param file_id: The id of the file to check. :return: True if public, else False. None if it doesn't exist at all. """ # Call the existing function: is_private = await self.is_private( mongo_conn = mongo_conn, file_id = file_id ) # Return the opposite result: if is_private is None: return is_private else: return not is_private async def is_owner( self, mongo_conn: AsyncMongoStorage, user_info: CoreUserInfoModel, file_id: ObjectId | str, ) -> bool | None: """ To check if a specific user is the owner of a specific file. :param mongo_conn: The instance of the database connection to perform this action. :param user_info: The details of the user who needs to have permissions to this file. :param file_id: The id of the file to check. :return: True if owner, else False. None if something goes wrong. """ # Parse the user conditions by ignoring the null values: user_conditions = {k: v for k, v in user_info.model_dump().items() if v is not None} # Run the query: record = await mongo_conn.find_one( collection = self.FILE_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "_id": ObjectId(file_id), "user": user_conditions, }), projection = { "user": True } ) # Return the result: if record is None: return False else: return True async def has_permission( self, mongo_conn: AsyncMongoStorage, user_info: CoreUserInfoModel, file_id: ObjectId | str, permission: Literal["read", "delete", "changePermissions"] ) -> bool | None: """ To check if a particular user has permissions to a given file. :param mongo_conn: The instance of the database connection to perform this action. :param user_info: The details of the user who needs to have permissions to this file. :param file_id: The id of the file to check. :param permission: The name of the permission that the said user must have on this file. :return: True if the user has said permission, else False. None if something goes wrong. """ # Parse the user conditions by ignoring the null values: user_conditions = {k: v for k, v in user_info.model_dump().items() if v is not None} # # If the user is the owner, # # he has all the permissions: # is_owner = await self.is_owner( # mongo_conn = mongo_conn, # user_info = user_info, # file_id = file_id # ) # if is_owner: return True # otherwise, we run the query to check for granted permissions: record = await mongo_conn.find_one( collection = self.FILE_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "_id": ObjectId(file_id), "sharedWith.user": user_conditions, }), projection = { "sharedWith.$": True } ) # Return the result: if record is None: return False else: return record["sharedWith"][0]["permissions"][permission] @staticmethod async def download( mongo_conn: AsyncMongoStorage, file_id: ObjectId ) -> io.BytesIO | None: """ To quickly download small files. Do not use this for larger files because the file will be held in RAM first and any large file will end up filling RAM fast. It's okay for smaller files that won't block up the memory. :param mongo_conn: The instance of the database connection to perform this action. :param file_id: The id of the file to fetch. :return: The file in a BytesIO buffer, or None if the file doesn't exist. """ # Get the file from the database: buffer = io.BytesIO() success = await mongo_conn.easy_download( destination = buffer, file_id = ObjectId(file_id), raise_exception = True ) buffer.seek(0) # Return the result: if not success: return None else: return buffer async def download_from_stream( self, mongo_conn: AsyncMongoStorage, file_id: ObjectId ) -> Any: """ To download any file as a stream. Better than the simple 'download' method because it doesn't block RAM. :param mongo_conn: The instance of the database connection to perform this action. :param file_id: The id of the file to fetch. :return: YET TO BE IMPLEMENTED. """ pass # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": import asyncio import time async def main(): files_mongo = AsyncMongoStorage( connection_string = r"mongodb://del.ditscentre.in:27017,wtt.ditscentre.in:27017,mum.arh.001.ditscentre.in:27017/admin?tls=true&tlsCAFile=%2Fetc%2Fssl%2Fcerts%2Fmongo_data_ca.pem&tlsCertificateKeyFile=%2Fetc%2Fssl%2Fcerts%2Fmongo_data_cert.pem&replicaSet=dits_mongod_rep&readPreference=primary&authMechanism=MONGODB-X509&authSource=%24external", database_name = "converseStore", max_connections = 10, debug = True ) await files_mongo.connect() my_fs = FileManagementModel() start_time = time.time() tasks = [my_fs.has_permission( mongo_conn = files_mongo, user_info = CoreUserInfoModel( userId = _, # industry = "fashion" ), file_id = "67583b64f35e2c7d3cab7955", permission = "read" ) for _ in range(1_000)] results = await asyncio.gather(*tasks) for i, r in enumerate(results): print(f"{i: >4}: {r}") print(f"Done in {time.time() - start_time:.5f} seconds") asyncio.run(main())