""" 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 from utils_v2.mail.mail_parser_v2 import parse_addr 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_object import ( CoreFileObjectInfoModel, CoreFileObjectSharingModel, CoreFileObjectPermissionsModel ) # To work with MongoDB: from bson import ObjectId # To work with datatypes: from typing import Any, Literal, List # To make deep copies: import copy # To make deep-copies: import copy # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class FileObjectManagementModel: # Define class-level variables: FILE_OBJECTS_COLLECTION = "_fileObjects" # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ def match_lists( self, list_a: list, list_b: list ): """ To match items in a List. The lists should have values that can be sorted. Mixed value dicts will raise exceptions. :param list_a: One of the lists to check. :param list_b: The other list to check. :return: True if they match, else False. """ # Start by assuming success: are_matching = True # If the lengths don't match, # there is no way the lists are exactly the same: if len(list_a) != len(list_b): are_matching = False return are_matching # Sort the lists first for better comparison: sorted_list_a = sorted(copy.deepcopy(list_a)) sorted_list_b = sorted(copy.deepcopy(list_b)) # Now loop through them and compare elements at the same index: for index in range(len(sorted_list_a)): av = sorted_list_a[index] bv = sorted_list_b[index] # If both of them are some sorts of lists: if isinstance(av, (list, tuple, set)): if not self.match_lists(list_a = list(av), list_b = list(bv)): are_matching = False break # If both of them are dicts, we go recursive: elif isinstance(av, dict): if not self.match_dicts(dict_a = av, dict_b = bv): are_matching = False break # Finally, we make direct comparison for things like strings, ints, etc. else: if av != bv: are_matching = False break # Done here: return are_matching def match_dicts( self, dict_a: dict, dict_b: dict ): """ To match items in a dict. Any lists in the dict should have values that can be sorted. Mixed value dicts will raise exceptions. :param dict_a: One of the dicts to check. :param dict_b: The other dict to check. :return: True if they match, else False. """ # Start by assuming success: are_matching = True # Iterate through the required items: for ak, av in dict_a.items(): # Fetch the corresponding value from the other dict: bv = dict_b.get(ak) # If the types are themselves mismatched, # no point in further comparison: if not isinstance(av, type(bv)): are_matching = False break # If both of them are some sorts of lists: if isinstance(av, (list, tuple, set)): if not self.match_lists(list_a = list(av), list_b = list(bv)): are_matching = False break # If both of them are dicts, we go recursive: elif isinstance(av, dict): if not self.match_dicts(dict_a = av, dict_b = bv): are_matching = False break # Finally, we make direct comparison for things like strings, ints, etc. else: if av != bv: are_matching = False break # Done here: return are_matching # ┏┓ • ┓ ┏┓ • # ┃┃┓┏┓┏┃┏ ┃┃┓┏┏┓┏┓┓┏┓┏ # ┗┻┗┻┗┗┛┗ ┗┻┗┻┗ ┛ ┗┗ ┛ async def exists( self, mongo_conn: AsyncMongoStorage, file_object_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_object_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_OBJECTS_COLLECTION, filter = {"_id": ObjectId(file_object_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_object_id: ObjectId ) -> CoreFileObjectInfoModel | None: """ To get the information about this file. :param mongo_conn: The instance of the database connection to perform this action. :param file_object_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_OBJECTS_COLLECTION, filter = {"_id": ObjectId(file_object_id)} ) # return the results: if not record: return None else: return CoreFileObjectInfoModel(**record) async def is_private( self, mongo_conn: AsyncMongoStorage, file_object_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_object_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_OBJECTS_COLLECTION, filter = {"_id": ObjectId(file_object_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_object_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_object_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_object_id = file_object_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_object_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_object_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_OBJECTS_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "_id": ObjectId(file_object_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, permission: Literal["read", "delete", "changePermissions"], file_object_id: ObjectId | str = None, check_ownership: bool = True ) -> bool | None: """ To check if a particular user has permissions to a given file obj. You may pass either an instance of the file's info, or you may send the file's id to check. If you pass just the file's id, a database call will be needed. :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 permission: The name of the permission that the said user must have on this file. :param file_object_id: The id of the file to check. :param check_ownership: Owners of files have all rights to their own files. If this field is set to True, ownership will be tested, and if the user is the owner of the file, the permission is assumed to be granted. If this field is set to False, only the sharing section of the file's record will be checked. :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 we have to test ownership: if check_ownership: is_owner = await self.is_owner( mongo_conn = mongo_conn, user_info = user_info, file_object_id = file_object_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_OBJECTS_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "_id": ObjectId(file_object_id), "sharedWith.user": user_conditions, }), projection = { "sharedWith.$": True } ) # Return the result: if record is None: return False else: return record["sharedWith"][0]["permissions"][permission] # ┓ • • # ┃ ┓┏╋┓┏┓┏┓ # ┗┛┗┛┗┗┛┗┗┫ # ┛ async def list_owned_dirs( self, mongo_conn: AsyncMongoStorage, user_info: CoreUserInfoModel, limit: int = 50, skip: int = 0, ) -> List[CoreFileObjectInfoModel] | None: """ To list all the dirs that are owned by the described user. :param mongo_conn: The connection instance to use to make the check. :param user_info: The information about the user that we must match. :param limit: How many max. records to fetch. :param skip: How many initial records to skip. Useful for pagination. :return: The list of files owned by the user (can be empty), or None if something goes wrong. """ # Run the query: records = await mongo_conn.find_many( collection = self.FILE_OBJECTS_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "user": {k: v for k, v in user_info.model_dump().items() if v is not None}, "isDir": True }), limit = limit, skip = skip ) # If something went wrong, we receive null for the records. # We pass that null on: if records is None: return None # Otherwise, we format and return the records: return [CoreFileObjectInfoModel(**record) for record in records] async def list_shared_dirs( self, mongo_conn: AsyncMongoStorage, user_info: CoreUserInfoModel, limit: int = 50, skip: int = 0, ) -> List[CoreFileObjectInfoModel] | None: """ To list all the dirs that have been shared with the described user. :param mongo_conn: The connection instance to use to make the check. :param user_info: The information about the user that we must match. :param limit: How many max. records to fetch. :param skip: How many initial records to skip. Useful for pagination. :return: The list of files owned by the user (can be empty), or None if something goes wrong. """ # Run the query: records = await mongo_conn.find_many( collection = self.FILE_OBJECTS_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "sharedWith.user": {k: v for k, v in user_info.model_dump().items() if v is not None}, "isDir": True }), limit = limit, skip = skip ) # If something went wrong, we receive null for the records. # We pass that null on: if records is None: return None # Otherwise, we format and return the records: return [CoreFileObjectInfoModel(**record) for record in records] async def list_owned_files( self, mongo_conn: AsyncMongoStorage, user_info: CoreUserInfoModel, limit: int = 50, skip: int = 0, ) -> List[CoreFileObjectInfoModel] | None: """ To list all the files that are owned by the described user. :param mongo_conn: The connection instance to use to make the check. :param user_info: The information about the user that we must match. :param limit: How many max. records to fetch. :param skip: How many initial records to skip. Useful for pagination. :return: The list of files owned by the user (can be empty), or None if something goes wrong. """ # Run the query: records = await mongo_conn.find_many( collection = self.FILE_OBJECTS_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "user": {k: v for k, v in user_info.model_dump().items() if v is not None}, "isDir": False }), limit = limit, skip = skip ) # If something went wrong, we receive null for the records. # We pass that null on: if records is None: return None # Otherwise, we format and return the records: return [CoreFileObjectInfoModel(**record) for record in records] async def list_shared_files( self, mongo_conn: AsyncMongoStorage, user_info: CoreUserInfoModel, limit: int = 50, skip: int = 0, ) -> List[CoreFileObjectInfoModel] | None: """ To list all the files that have been shared with the described user. :param mongo_conn: The connection instance to use to make the check. :param user_info: The information about the user that we must match. :param limit: How many max. records to fetch. :param skip: How many initial records to skip. Useful for pagination. :return: The list of files owned by the user (can be empty), or None if something goes wrong. """ # Run the query: records = await mongo_conn.find_many( collection = self.FILE_OBJECTS_COLLECTION, filter = mongo_conn.dict_to_dot_notation({ "sharedWith.user": {k: v for k, v in user_info.model_dump().items() if v is not None}, "isDir": False }), limit = limit, skip = skip ) # If something went wrong, we receive null for the records. # We pass that null on: if records is None: return None # Otherwise, we format and return the records: return [CoreFileObjectInfoModel(**record) for record in records] # ┳┓ ┓• # ┣┫┏┓┏┓┏┫┓┏┓┏┓ # ┛┗┗ ┗┻┗┻┗┛┗┗┫ # ┛ @staticmethod async def download_file( 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. WARNING: Check for permissions before using this method. :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 @staticmethod async def get_file_download_stream( 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. Once the stream is created you can read from it, and terminate it like this: READ: await stream.read(chunk_size) CLOSE (without awaiting): stream.close() WARNING: Check for permissions before using this method. :param mongo_conn: The instance of the database connection to perform this action. :param file_id: The id of the file whose stream you would like to fetch. :return: Returns the stream object that will allow more efficient downloads of files on the user's end. """ return await mongo_conn.get_download_stream(file_id = file_id) # ┓ ┏ • • # ┃┃┃┏┓┓╋┓┏┓┏┓ # ┗┻┛┛ ┗┗┗┛┗┗┫ # ┛ async def make_dir(self): pass async def upload_file(self): pass async def get_file_upload_stream(self): pass # ┳┓ ┓ • # ┃┃┏┓┃┏┓╋┓┏┓┏┓ # ┻┛┗ ┗┗ ┗┗┛┗┗┫ # ┛ async def delete_dir(self): pass async def delete_file(self): pass # ┏┓ • • # ┃┃┏┓┏┓┏┳┓┓┏┏┓┏┓┏┓┏ # ┣┛┗ ┛ ┛┗┗┗┛┛┗┗┛┛┗┛ async def update_permissions(self): pass async def make_public(self): pass async def mak_private(self): 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 = FileObjectManagementModel() user_a = CoreUserInfoModel( userId = 1 ) user_b = CoreUserInfoModel( userId = 6 ) owned_files = await my_fs.list_shared_files( mongo_conn = files_mongo, user_info = user_b ) if owned_files is None: print("ERROR!") else: print(f"FILES ({len(owned_files)}):", json.to_string(owned_files, default = str)) asyncio.run(main())