khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241211) Started more work on the files part and added validation to the phone no. in the API exposed for Nimbus.

Browse Source
This commit is contained in:
khushal
2024-12-11 13:20:54 +05:30
parent 570a28ef38
commit 897c32c595
8 changed files with 792 additions and 433 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/behaviour/file/file.py
-371
View File
@@ -1,371 +0,0 @@
"""
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())
models/behaviour/file/file_object.py
+672
View File
@@ -0,0 +1,672 @@
"""
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())
models/data/api/sms/send.py
+24 -2
View File
@@ -82,7 +82,7 @@ class NimbusSMSIndiaMessage(BaseModel):
recipientNo: str = Field( recipientNo: str = Field(
description = "the phone no. of the target recipient", description = "the phone no. of the target recipient",
min_length = 1, pattern = r"\+?\d{0,3}\s*\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}",
frozen = True frozen = True
) )
@@ -106,6 +106,17 @@ class NimbusSMSIndiaMessage(BaseModel):
class Config: class Config:
extra = "forbid" extra = "forbid"
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("recipientNo", mode = "before")
def validate_contact_nos(cls, value):
value = regex.replace(text = str(value), pattern = r"[^\d]", substitute_text = "")
if len(value) > 10: value = regex.replace(text = str(value), pattern = r"^(91)", substitute_text = "")
value = regex.find_first(text = str(value), pattern = r"^[\d]{10}")
return value
# --------------------------------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------------------------------
@@ -194,4 +205,15 @@ class SMSSendRequestData(BaseModel):
if __name__ == "__main__": if __name__ == "__main__":
pass from utils_v2.string import json
my_request = SMSSendRequestData(
tokenId = "670f580d7cda4ebc1adc3444",
message = NimbusSMSIndiaMessage(
recipientNo = "+.9.1 ---> 93261 3642fgnn193261",
text = "Hello, Nimbus!",
templateId = "12345678"
)
)
print("NIMBUS SMS API MODEL:", json.to_string(my_request.model_dump(), default = str))
models/data/core/auth_token.py
+7 -4
View File
@@ -82,11 +82,11 @@ import datetime
class CoreAuthTokenModel(BaseModel): class CoreAuthTokenModel(BaseModel):
version: str = Field( authTokenId: ObjectId = Field(
description = "a hint about the version no. of this message", description = "the id of the document in mongodb that holds this information",
min_length = 1,
frozen = True, frozen = True,
default = "2.0.0" default = None,
alias = "_id"
) )
serviceType: Literal["email", "sms", "chat"] = Field( serviceType: Literal["email", "sms", "chat"] = Field(
@@ -176,6 +176,9 @@ class CoreAuthTokenModel(BaseModel):
extra = "allow" extra = "allow"
arbitrary_types_allowed = True arbitrary_types_allowed = True
def model_dump(self, *args, **kwargs):
return super().model_dump(*args, by_alias = True, **kwargs)
# ┓┏ ┓• ┓ • # ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
models/data/core/file.py → models/data/core/file_object.py
+75 -41
View File
@@ -80,22 +80,28 @@ import datetime
# ***************************************************************************************************************** # *****************************************************************************************************************
class CoreFilePermissionsModel(BaseModel): class CoreFileObjectPermissionsModel(BaseModel):
read: bool = Field( read: bool = Field(
description = "grants permissions to read/view this file", description = "grants permissions to read/view this file/dir",
frozen = True, frozen = True,
default = True default = True
) )
write: bool = Field(
description = "grants permissions to write new files this dir; irrelevant for files",
frozen = True,
default = False
)
delete: bool = Field( delete: bool = Field(
description = "grants permissions to delete this file entirely", description = "grants permissions to delete this file/dir entirely",
frozen = True, frozen = True,
default = False default = False
) )
changePermissions: bool = Field( changePermissions: bool = Field(
description = "grants permissions to modify the permissions of this file", description = "grants permissions to modify the permissions of this file/dir",
frozen = True, frozen = True,
default = False default = False
) )
@@ -113,14 +119,14 @@ class CoreFilePermissionsModel(BaseModel):
# --------------------------------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------------------------------
class CoreFileSharingModel(BaseModel): class CoreFileObjectSharingModel(BaseModel):
user: CoreUserInfoModel = Field( user: CoreUserInfoModel = Field(
description = "to identify the user who has access to this file", description = "to identify the user who has access to this file",
frozen = True frozen = True
) )
permissions: CoreFilePermissionsModel = Field( permissions: CoreFileObjectPermissionsModel = Field(
description = "the permissions that the above mentioned user has to this file", description = "the permissions that the above mentioned user has to this file",
frozen = True frozen = True
) )
@@ -138,66 +144,73 @@ class CoreFileSharingModel(BaseModel):
# --------------------------------------------------------------------------------------------------------------------- # ---------------------------------------------------------------------------------------------------------------------
class CoreFileInfoModel(BaseModel): class CoreFileObjectInfoModel(BaseModel):
version: str = Field( fileObjectId: ObjectId = Field(
description = "a hint about the version no. of this message", description = "the id of the document in mongodb that holds this information",
min_length = 1,
frozen = True, frozen = True,
default = "2.0.0" default = None,
alias = "_id"
) )
user: CoreUserInfoModel = Field( user: CoreUserInfoModel = Field(
description = "to identify the user who owns this file", description = "to identify the user who owns this file/dir",
frozen = True frozen = True
) )
filename: str = Field( isDir: bool = Field(
description = "the name of this file", description = "to know whether this object is s file or a directory",
frozen = True frozen = True,
)
uploadTs: AwareDatetime = Field(
description = "the time (utc) at which this message was sent by the sender",
frozen = True
)
length: int = Field(
description = "to note when the user has marked this message as unread",
frozen = False,
default = False default = False
) )
hash: str = Field( name: str = Field(
description = "a simple hash to verify the integrity of the uploaded data", description = "the name of this file/dir",
frozen = True frozen = True
) )
createTs: AwareDatetime = Field(
description = "the time (utc) at which this file/dir was created",
frozen = True
)
length: int | None = Field(
description = "to note the size of the file; irrelevant for dirs",
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( metadata: Dict[str, Any] = Field(
description = "any addition data about this file to filter it later", description = "any addition data about this file/dir to filter it later",
frozen = False, frozen = False,
default = {} default = {}
) )
tags: List[str] = Field( tags: List[str] = Field(
description = "a list of keywords to apply to this file to filter it later", description = "a list of keywords to apply to this file/dir to filter it later",
frozen = False, frozen = False,
default = [], default = [],
examples = ["Bank Statement", "PDF", "bhopli@orange.com"] examples = ["Bank Statement", "PDF", "bhopli@orange.com"]
) )
parentId: ObjectId | None = Field( parentId: ObjectId | None = Field(
description = "to identify the parent directory of this file; null means root directory", description = "to identify the parent dir of this file/dir; null means root dir",
frozen = False frozen = False
) )
isPrivate: bool = Field( isPrivate: bool = Field(
description = "whether, or not, this file is a private file", description = "whether, or not, this file/dir is a private file",
frozen = True frozen = True
) )
sharedWith: List[CoreFileSharingModel] = Field( sharedWith: List[CoreFileObjectSharingModel] = Field(
description = "sharing settings; specially relevant when the file is private", description = "sharing settings; specially relevant when the file/dir is private",
frozen = True, frozen = True,
default = [] default = []
) )
@@ -211,20 +224,39 @@ class CoreFileInfoModel(BaseModel):
extra = "allow" extra = "allow"
arbitrary_types_allowed = True arbitrary_types_allowed = True
def model_dump(self, *args, **kwargs):
return super().model_dump(*args, by_alias = True, **kwargs)
# ┓┏ ┓• ┓ • # ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("uploadTs", mode = "before") @field_validator("createTs", mode = "before")
def parse_date_time(cls, value): def parse_date_time(cls, value):
return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC) return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC)
@field_validator("parentId", mode = "before") @field_validator("fileObjectId", "parentId", mode = "before")
def parse_oid(cls, value): def parse_oid(cls, value):
try: value = ObjectId(value) try: value = ObjectId(value)
except: pass except: pass
return value return value
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
@property
def summary(self):
return {
"user": self.user,
"isDir": self.isDir,
"name": self.name,
"createTs": self.createTs,
"parentId": self.parentId,
"isPrivate": self.isPrivate,
}
# ***************************************************************************************************************** # *****************************************************************************************************************
# ***** **** # ***** ****
@@ -237,7 +269,8 @@ if __name__ == "__main__":
from utils_v2.string import json from utils_v2.string import json
file_info = CoreFileInfoModel( file_info = CoreFileObjectInfoModel(
# _id = "67519cf3a7804fcbc6f12452",
user = CoreUserInfoModel( user = CoreUserInfoModel(
fullName = "Bhopli Narangi", fullName = "Bhopli Narangi",
userId = 1, userId = 1,
@@ -247,10 +280,10 @@ if __name__ == "__main__":
branchId = 5, branchId = 5,
industry = "fashion" industry = "fashion"
), ),
filename = "Giga-Cat.png", name = "Midnight-Snack.png",
uploadTs = date_time.get_current_utc_date_time(as_string = False), createTs = date_time.get_current_utc_date_time(as_string = False),
length = 1024, length = 1024,
hash = "abcdefgh", hash = "abcdefgh12345678",
metadata = { metadata = {
"camera": "iPhone 1000 Pro Max XS" "camera": "iPhone 1000 Pro Max XS"
}, },
@@ -262,7 +295,7 @@ if __name__ == "__main__":
parentId = "67519cf3a7804fcbc6f12452", parentId = "67519cf3a7804fcbc6f12452",
isPrivate = True, isPrivate = True,
sharedWith = [ sharedWith = [
CoreFileSharingModel( CoreFileObjectSharingModel(
user = CoreUserInfoModel( user = CoreUserInfoModel(
fullName = "Polki Muchhwaali", fullName = "Polki Muchhwaali",
userId = 6, userId = 6,
@@ -272,8 +305,9 @@ if __name__ == "__main__":
branchId = 10, branchId = 10,
industry = "entertainment" industry = "entertainment"
), ),
permissions = CoreFilePermissionsModel( permissions = CoreFileObjectPermissionsModel(
read = True, read = True,
write = False,
delete = False, delete = False,
changePermissions = False changePermissions = False
) )
models/data/core/message.py
+7 -4
View File
@@ -79,11 +79,11 @@ import datetime
class CoreMessageModel(BaseModel): class CoreMessageModel(BaseModel):
version: str = Field( messageId: ObjectId = Field(
description = "a hint about the version no. of this message", description = "the id of the document in mongodb that holds this information",
min_length = 1,
frozen = True, frozen = True,
default = "2.0.0" default = None,
alias = "_id"
) )
ts: AwareDatetime = Field( ts: AwareDatetime = Field(
@@ -170,6 +170,9 @@ class CoreMessageModel(BaseModel):
extra = "allow" extra = "allow"
arbitrary_types_allowed = True arbitrary_types_allowed = True
def model_dump(self, *args, **kwargs):
return super().model_dump(*args, by_alias = True, **kwargs)
# ┓┏ ┓• ┓ • # ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
models/data/core/payment.py
+7 -4
View File
@@ -126,11 +126,11 @@ class PaymentEvent(BaseModel):
class CorePaymentModel(BaseModel): class CorePaymentModel(BaseModel):
version: str = Field( paymentId: ObjectId = Field(
description = "a hint about the version no. of this message", description = "the id of the document in mongodb that holds this information",
min_length = 1,
frozen = True, frozen = True,
default = "2.0.0" default = None,
alias = "_id"
) )
paymentStatus: Literal[ paymentStatus: Literal[
@@ -196,6 +196,9 @@ class CorePaymentModel(BaseModel):
extra = "allow" extra = "allow"
arbitrary_types_allowed = True arbitrary_types_allowed = True
def model_dump(self, *args, **kwargs):
return super().model_dump(*args, by_alias = True, **kwargs)
# ┓┏ ┓• ┓ • # ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
models/data/core/user.py
-7
View File
@@ -79,13 +79,6 @@ import datetime
class CoreUserInfoModel(BaseModel): class CoreUserInfoModel(BaseModel):
version: str = Field(
description = "a hint about the version no. of this message",
min_length = 1,
frozen = True,
default = "2.0.0"
)
fullName: str | None = Field( fullName: str | None = Field(
description = "the full name of the user as found in the database", description = "the full name of the user as found in the database",
frozen = True, frozen = True,