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

(20241219) SMS module fully re-organized.

Browse Source
This commit is contained in:
khushal
2024-12-19 17:26:18 +05:30
parent d89a88ab8e
commit 79b771b48c
28 changed files with 1086 additions and 827 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controllers_v2/core/auth_token.py
+176 -38
View File
@@ -31,6 +31,9 @@
# To make sibling directories accessible for imports:
import sys
import httpx
sys.path.append(".")
sys.path.append("..")
@@ -39,9 +42,10 @@ 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
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
# Base model:
from controllers.base import BaseModel
from controllers_v2.core.base import CoreBaseModel
# Data models:
from models.core.auth_token import CoreAuthTokenModel
@@ -90,7 +94,7 @@ from bson import ObjectId
# *****************************************************************************************************************
class CoreAuthTokenController(BaseModel):
class CoreAuthTokenController(CoreBaseModel):
# ┏┓┓ ┓┏
# ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏
@@ -99,10 +103,53 @@ class CoreAuthTokenController(BaseModel):
# For MongoDB:
AUTH_COLLECTION = "_authTokens"
async def get_token_key(
# ┏┓
# ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
# ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
def __init__(
self,
db_conn: AsyncMySQL,
mongo_conn: AsyncMongo,
cache: AsyncRedisCache = None,
http_client: httpx.AsyncClient = None,
alert_url: str = None,
base_filter: dict = None,
debug = True,
debug_prefix = "Core Base (C) | ",
debug_only_errors = True
):
"""
This is the core controller for all authorization and token activities.
:param cache: The object to use for caching results from database calls.
:param debug: Whether, or not, you would like to print debugging messages:
:param debug_prefix: The prefix to print with the debugging messages.
:param debug_only_errors: Whether you would like to print only error messages or all messages.
:return: None.
"""
# Accept the base filter:
self._base_filter = base_filter or {}
# Invoke the parent's constructor:
CoreBaseModel.__init__(
self,
cache = cache,
http_client = http_client,
alert_url = alert_url,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
# ┏┓┳┓┳┳┳┓ ┏┓ ┓ ┳┳ ┓
# ┃ ┣┫┃┃┃┃ ━━ ┃ ┏┓┏┓┏┓╋┏┓ ┏┓┏┓┏┫ ┃┃┏┓┏┫┏┓╋┏┓
# ┗┛┛┗┗┛┻┛ ┗┛┛ ┗ ┗┻┗┗ ┗┻┛┗┗┻ ┗┛┣┛┗┻┗┻┗┗
# ┛
async def generate_token_key(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
auth_token: CoreAuthTokenModel,
token_notes: dict,
session_token: str = None,
@@ -111,8 +158,8 @@ class CoreAuthTokenController(BaseModel):
"""
Stores params from the session info and gives an identifier to use in the authorization URL. Use this when the
user requests an authorization URL to link your service to another service (like GMail).
:param db_conn: The database connection (MariaDB) to use to perform the action.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param sql_conn: The database connection (MariaDB) to use to perform the action.
:param mongo_data_conn: The database connection (MongoDB) to use to perform the action.
:param auth_token: An instance of the core auth-token model that holds data in the database.
:param token_notes: Any notes to feed into MariaDB with the token identifier.
:param session_token: The session token of the user who requested this service.
@@ -124,9 +171,9 @@ class CoreAuthTokenController(BaseModel):
# Get the identifier from the database if it already exists, else create one.
# BE CAREFUL WITH THE KEYS HERE, THEY SHOULD MATCH THE FIELDS OF THE CORE AUTH-TOKEN MODEL:
mongo_json = await mongo_conn.find_one_and_update(
mongo_json = await mongo_data_conn.find_one_and_update(
collection = self.AUTH_COLLECTION,
filter = mongo_conn.dict_to_dot_notation({
filter = mongo_data_conn.dict_to_dot_notation({
"serviceType": auth_token.serviceType,
"user": {
"entityId": auth_token.user.entityId,
@@ -166,7 +213,7 @@ class CoreAuthTokenController(BaseModel):
db_json = {}
if mongo_json is not None:
db_json = await self.call_procedure(
db_conn = db_conn,
sql_conn = sql_conn,
proc_name = "entity_integration_save",
proc_args = (
auth_token.user.entityId, # ..................................... 'p_entity_id'
@@ -187,8 +234,8 @@ class CoreAuthTokenController(BaseModel):
async def set_token(
self,
db_conn: AsyncMySQL,
mongo_conn: AsyncMongo,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
token_key: ObjectId | str,
auth_token: CoreAuthTokenModel,
token_notes: dict,
@@ -199,9 +246,9 @@ class CoreAuthTokenController(BaseModel):
This method is to be called when the end user authorizes your service to connect to his third-party account. For
example, when the end user allows you to access his GMail account. USE THIS FOR UPDATING (REFRESHING) TOKENS
ALSO.
:param db_conn: The database connection (MariaDB) to use to perform the action.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param token_key: The identifier granted by the 'get_token_key' method.
:param sql_conn: The database connection (MariaDB) to use to perform the action.
:param mongo_data_conn: The database connection (MongoDB) to use to perform the action.
:param token_key: The identifier granted by the 'generate_token_key' method.
:param auth_token: The actual auth/token data to be saved to the database.
:param token_notes: Any notes to feed into MariaDB with the token identifier.
:param session_token: The session token of the user who requested this service.
@@ -216,9 +263,9 @@ class CoreAuthTokenController(BaseModel):
# Save the token to MongoDB.
# BE CAREFUL WITH THE KEYS HERE, THEY SHOULD MATCH THE FIELDS OF THE CORE AUTH-TOKEN MODEL:
mongo_json = await mongo_conn.find_one_and_update(
mongo_json = await mongo_data_conn.find_one_and_update(
collection = self.AUTH_COLLECTION,
filter = mongo_conn.dict_to_dot_notation({
filter = mongo_data_conn.dict_to_dot_notation({
"key": ObjectId(token_key),
"clientUserId": auth_token.clientUserId
}),
@@ -250,7 +297,7 @@ class CoreAuthTokenController(BaseModel):
# Tell MariaDB that the token was saved:
if mongo_json is not None:
db_json = await self.call_procedure(
db_conn = db_conn,
sql_conn = sql_conn,
proc_name = "entity_integration_save",
proc_args = (
mongo_json["user"]["entityId"], # ............................... 'p_entity_id'
@@ -270,24 +317,82 @@ class CoreAuthTokenController(BaseModel):
# Done here:
return token_saved
async def set_token_direct(
self,
sql_conn: AsyncMySQL,
mongo_data_conn: AsyncMongo,
auth_token: CoreAuthTokenModel,
token_notes: dict,
session_token: str = None
) -> bool:
"""
Some authorizations don't need two steps, but our core system works on the 2-step approach that was developed to
work with Google's GMail OAuth2.0 mechanism.
:param sql_conn: The database connection (MariaDB) to use to perform the action.
:param mongo_data_conn: The database connection (MongoDB) to use to perform the action.
:param auth_token: The actual auth/token data to be saved to the database.
:param token_notes: Any notes to feed into MariaDB with the token identifier.
:param session_token: The session token of the user who requested this service.
:return: True if saved, False if failed.
"""
# Start by assuming failure:
success = False
# Get a token id (and receive its key):
token_key = await self.generate_token_key(
sql_conn = sql_conn,
mongo_data_conn = mongo_data_conn,
auth_token = auth_token,
token_notes = token_notes,
session_token = session_token
)
# Immediately save the details against that token id:
success = await self.set_token(
sql_conn = sql_conn,
mongo_data_conn = mongo_data_conn,
token_key = token_key,
auth_token = auth_token,
token_notes = token_notes,
session_token = session_token
)
# Done here:
return success
# ┏┓┳┓┳┳┳┓ ┳┓ •
# ┃ ┣┫┃┃┃┃ ━━ ┣┫┏┓╋┏┓┓┏┓┓┏┏┓
# ┗┛┛┗┗┛┻┛ ┛┗┗ ┗┛ ┗┗ ┗┛┗
async def get_token_from_id(
self,
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
token_id: ObjectId | str = None,
additional_filter: dict = None
) -> CoreAuthTokenModel | None:
"""
To retrieve stored tokens from the database. One token at a time.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param mongo_data_conn: The database connection (MongoDB) to use to perform the action.
:param token_id: The identifier of the document that holds the token's details.
:param additional_filter: Any addition filters to use.
:return: The retrieved record that has the token, and information about the service and client if found, else
None when there is no matching record.
"""
# Prepare the filter:
filter_json = {"_id": ObjectId(token_id)}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items(): filter_json[k] = v
# If there is some filtering possible, we fetch the token:
token = await mongo_conn.find_one(
token = await mongo_data_conn.find_one(
collection = self.AUTH_COLLECTION,
filter = {"_id": ObjectId(token_id)}
filter = filter_json
)
# Done here:
@@ -295,22 +400,31 @@ class CoreAuthTokenController(BaseModel):
async def get_token_from_key(
self,
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
token_key: ObjectId | str = None,
additional_filter: dict = None
) -> CoreAuthTokenModel | None:
"""
To retrieve stored tokens from the database. One token at a time.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param token_key: The identifier granted by the 'get_token_key' method.
:param mongo_data_conn: The database connection (MongoDB) to use to perform the action.
:param token_key: The identifier granted by the 'generate_token_key' method.
:param additional_filter: Any addition filters to use.
:return: The retrieved record that has the token, and information about the service and client if found, else
None when there is no matching record.
"""
# Prepare the filter:
filter_json = {"key": ObjectId(token_key)}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items(): filter_json[k] = v
# If there is some filtering possible, we fetch the token:
token = await mongo_conn.find_one(
token = await mongo_data_conn.find_one(
collection = self.AUTH_COLLECTION,
filter = {"key": ObjectId(token_key)}
filter = filter_json
)
# Done here:
@@ -318,24 +432,33 @@ class CoreAuthTokenController(BaseModel):
async def get_tokens_from_ids(
self,
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
token_ids: List[ObjectId | str] = None,
limit: int = 100
limit: int = 100,
additional_filter: dict = None
) -> List[CoreAuthTokenModel]:
"""
To retrieve stored tokens from the database. Multiple tokens at a time.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param mongo_data_conn: The database connection (MongoDB) to use to perform the action.
:param token_ids: The identifier of the document that holds the token's details.
:param limit: The max. no. of records to pick.
:param additional_filter: Any addition filters to use.
:return: The retrieved record that has the token, and information about the service and client if found, else
None when there is no matching record.
"""
# Prepare the filter:
filter_json = {"_id": {"$in": [ObjectId(k) for k in token_ids]}}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items(): filter_json[k] = v
# If there is some filtering possible, we fetch the token:
tokens = await mongo_conn.find_many(
tokens = await mongo_data_conn.find_many(
collection = self.AUTH_COLLECTION,
filter = {"_id": {"$in": [ObjectId(k) for k in token_ids]}},
filter = filter_json,
limit = limit
)
@@ -344,30 +467,45 @@ class CoreAuthTokenController(BaseModel):
async def get_tokens_from_keys(
self,
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
token_keys: List[ObjectId | str] = None,
limit: int = 100
limit: int = 100,
additional_filter: dict = None
) -> List[CoreAuthTokenModel]:
"""
To retrieve stored tokens from the database. Multiple tokens at a time.
:param mongo_conn: The database connection (MongoDB) to use to perform the action.
:param token_keys: the identifiers granted by the 'get_token_key' method.
:param mongo_data_conn: The database connection (MongoDB) to use to perform the action.
:param token_keys: the identifiers granted by the 'generate_token_key' method.
:param limit: The max. no. of records to pick.
:param additional_filter: Any addition filters to use.
:return: The retrieved record that has the token, and information about the service and client if found, else
None when there is no matching record.
"""
# Prepare the filter:
filter_json = {"key": {"$in": [ObjectId(k) for k in token_keys]}}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items(): filter_json[k] = v
# If there is some filtering possible, we fetch the token:
tokens = await mongo_conn.find_many(
tokens = await mongo_data_conn.find_many(
collection = self.AUTH_COLLECTION,
filter = {"key": {"$in": [ObjectId(k) for k in token_keys]}},
filter = filter_json,
limit = limit
)
# Done here:
return [CoreAuthTokenModel(**token) for token in tokens]
# ┏┓┳┓┳┳┳┓ ┳┓ ┓
# ┃ ┣┫┃┃┃┃ ━━ ┃┃┏┓┃┏┓╋┏┓
# ┗┛┛┗┗┛┻┛ ┻┛┗ ┗┗ ┗┗
pass
# *****************************************************************************************************************
# ***** ****
controllers_v2/core/base.py
+15 -21
View File
@@ -42,9 +42,6 @@ from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
from utils_v2.api.codes import StatusCodes
# For asynchronous activities:
import asyncio
# For debugging:
from icecream import IceCreamDebugger
@@ -89,7 +86,7 @@ from typing import List
# *****************************************************************************************************************
class BaseModel:
class CoreBaseModel:
PREVIEW_LENGTH = 250
@@ -99,12 +96,12 @@ class BaseModel:
alert_url = None,
http_client = None,
debug = True,
debug_prefix = "Model | ",
debug_prefix = "Core Base (C) | ",
debug_only_errors = True
):
"""
This is the base model.
This is the base controller.
:param cache: The object to use for caching results from database calls.
:param debug: Whether, or not, you would like to print debugging messages:
:param debug_prefix: The prefix to print with the debugging messages.
@@ -125,9 +122,6 @@ class BaseModel:
if not debug: self._printer.disable()
self._debug_only_errors = debug_only_errors
# A semaphore for activities that must absolutely be done one at a time:
self.__exclusive_semaphore = asyncio.Semaphore(1)
# A simple debugging output:
self._printer("Model initialized.")
@@ -175,7 +169,7 @@ class BaseModel:
cache: AsyncRedisCache,
cache_key: str,
cache_expiry: int,
db_conn: AsyncMySQL,
sql_conn: AsyncMySQL,
proc_name: str,
proc_args: tuple,
retry_count: int = 1,
@@ -189,7 +183,7 @@ class BaseModel:
:param cache: The caching object to use to set the session in cache memory.
:param cache_key: The string to use as the key when caching the response.
:param cache_expiry: The no. of seconds after which this information will be deleted from the cache.
:param db_conn: The connection instance to use to call the procedure.
:param sql_conn: The connection instance to use to call the procedure.
:param proc_name: The name of the stored procedure that must be called.
:param proc_args: The args to be sent to the stored procedure.
:param retry_count: The max. number of times to try in case one or more attempts fail.
@@ -209,7 +203,7 @@ class BaseModel:
# Make the database call:
data = await self.call_procedure(
db_conn = db_conn,
sql_conn = sql_conn,
proc_name = proc_name,
proc_args = proc_args,
retry_count = retry_count,
@@ -227,7 +221,7 @@ class BaseModel:
async def call_procedure(
self,
db_conn: AsyncMySQL,
sql_conn: AsyncMySQL,
proc_name: str,
proc_args: tuple,
retry_count: int = 1,
@@ -238,7 +232,7 @@ class BaseModel:
"""
Calls a stored procedure and returns the response as a JSON-like object (dict or list).
:param db_conn: The connection instance to use to call the procedure.
:param sql_conn: The connection instance to use to call the procedure.
:param proc_name: The name of the stored procedure that must be called.
:param proc_args: The args to be sent to the stored procedure.
:param retry_count: The max. number of times to try in case one or more attempts fail.
@@ -251,7 +245,7 @@ class BaseModel:
"""
# Call the stored procedure:
db_json, exception = await db_conn.call_procedure_and_get_json(
db_json, exception = await sql_conn.call_procedure_and_get_json(
proc_name,
proc_args,
retry_count = retry_count,
@@ -290,14 +284,14 @@ class BaseModel:
async def execute_one(
self,
db_conn: AsyncMySQL,
sql_conn: AsyncMySQL,
query: str,
session_token: str = None
):
"""
Runs one query and sends an alert if that fails.
:param db_conn: The connection to use to run the query.
:param sql_conn: The connection to use to run the query.
:param query: The query to run.
:param session_token: A session token to share with the tech module when alerts need to be sent out for any
occurrence of exceptions. If this is passed, the tech module will be able to tell you which user faced the
@@ -306,7 +300,7 @@ class BaseModel:
"""
# Run the query:
rows_affected, db_response, exception = await db_conn.execute_one(query = query, return_exception = True)
rows_affected, db_response, exception = await sql_conn.execute_one(query = query, return_exception = True)
# Send an alert out on exceptions:
if exception is not None:
@@ -329,7 +323,7 @@ class BaseModel:
async def execute_many(
self,
db_conn: AsyncMySQL,
sql_conn: AsyncMySQL,
query: str,
data: List[tuple],
session_token: str = None
@@ -337,7 +331,7 @@ class BaseModel:
"""
Runs many queries and sends an alert if that fails.
:param db_conn: The connection to use to run the query.
:param sql_conn: The connection to use to run the query.
:param query: The query to run.
:param data: The data to feed into the query.
:param session_token: A session token to share with the tech module when alerts need to be sent out for any
@@ -347,7 +341,7 @@ class BaseModel:
"""
# Run the query:
rows_affected, db_response, exception = await db_conn.execute_many(
rows_affected, db_response, exception = await sql_conn.execute_many(
query = query,
data = data,
return_exception = True
controllers_v2/core/message.py
+156 -78
View File
@@ -6,7 +6,7 @@
DATE:
Thursday, 12th Dec., 2024
Thursday, 19th Dec., 2024
OBJECTIVE:
@@ -34,41 +34,25 @@ import sys
sys.path.append(".")
sys.path.append("..")
# For Quart:
from quart import current_app
# 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
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
# Base model:
from controllers.base import BaseModel
# Controllers:
from controllers_v2.core.base import CoreBaseModel
from controllers_v2.core.auth_token import CoreAuthTokenController
# Data models:
from models.core.auth_token import CoreAuthTokenModel
# Models:
from models.core.message import CoreMessageModel
from models.core.user import CoreUserInfoModel
# To work with MongoDB:
from bson import ObjectId
from pymongo import InsertOne, UpdateOne, ReplaceOne
# To work with datatypes:
from typing import Literal, List, Dict, Any
from typing import List
# To make deep-copies:
import copy
# To work with base-64 encoding:
import base64
# To work with date and time:
import datetime
# For asynchronous activities:
import asyncio
# To make HTTP requests:
import httpx
# *****************************************************************************************************************
@@ -108,7 +92,7 @@ import asyncio
# *****************************************************************************************************************
class CoreMessageController(BaseModel):
class CoreMessageController(CoreAuthTokenController):
# ┏┓┓ ┓┏
# ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏
@@ -117,45 +101,92 @@ class CoreMessageController(BaseModel):
# For MongoDB:
MESSAGES_COLLECTION = "_messages"
# ┏┓
# ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
# ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
def __init__(
self,
cache: AsyncRedisCache = None,
http_client: httpx.AsyncClient = None,
alert_url: str = None,
base_filter: dict = None,
debug: bool = True,
debug_prefix: str = "Message (C) | ",
debug_only_errors: bool = True
):
"""
This is the foundational controller of all message controllers. You must structure individual message
controllers through this structure. Individual message controllers would be for things like mails, SMS messages,
chat app messages, etc.
:param cache: The object to use for caching results from database calls.
:param http_client: The HTTP client
:param base_filter: The basic filter that will be applied to all fetching/updating queries. WARNING: THE BASE
FILTER WILL ALWAYS BE APPLIED AUTOMATICALLY. SET THIS UP WISELY.
:param debug: Whether, or not, you would like to print debugging messages:
:param debug_prefix: The prefix to print with the debugging messages.
:param debug_only_errors: Whether you would like to print only error messages or all messages.
:return: None.
"""
# Accept the base filter:
self._base_filter = base_filter or {}
# Invoke the parent's constructor:
CoreAuthTokenController.__init__(
self,
cache = cache,
http_client = http_client,
alert_url = alert_url,
base_filter = base_filter,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
# ┏┓┳┓┳┳┳┓ ┏┓
# ┃ ┣┫┃┃┃┃ ━━ ┃ ┏┓┏┓┏┓╋┏┓
# ┗┛┛┗┗┛┻┛ ┗┛┛ ┗ ┗┻┗┗
async def insert(
async def save_one_message(
self,
mongo_conn: AsyncMongo,
message: CoreMessageModel
mongo_data_conn: AsyncMongo,
message: CoreMessageModel,
session = None
) -> ObjectId:
"""
Simply insert one message document into the database.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param message: The message to save into the database.
:param session: In case you need to perform this operation as a transaction, pass a session here.
:return: The object id of the inserted document.
"""
# Simply insert the document:
return await mongo_conn.insert_one(
return await mongo_data_conn.insert_one(
collection = self.MESSAGES_COLLECTION,
document = message,
raise_exception = True
document = message.model_dump(),
raise_exception = True,
session = session
)
async def bulk_write(
async def bulk_operate_messages(
self,
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
mongo_operations: list
) -> int:
"""
Needed in cases like forcing re-sync of mails where you need to perform actions like bulk replacements of
existing documents. Not recommended to use. Please use very carefully to ensure document integrity.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param mongo_operations: The list operations that are supported by MongoDB's Bulk Write system.
:return: The no. of documents affected.
"""
return await mongo_conn.bulk_write(
return await mongo_data_conn.bulk_write(
collection = self.MESSAGES_COLLECTION,
requests = mongo_operations,
raise_exception = True
@@ -167,29 +198,35 @@ class CoreMessageController(BaseModel):
async def count_messages(
self,
mongo_conn: AsyncMongo,
token_ids: List[ObjectId | str],
mongo_data_conn: AsyncMongo,
token_ids: List[ObjectId | str] = None,
additional_filter: dict = None
) -> int:
"""
Just counts the no. of messages that match a given set of conditions.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param token_ids: The token ids of the accounts from which these messages must be fetched.
:param additional_filter: Any addition filters to use.
:return: The no. of messages that match the given conditions.
"""
# We cannot allow counting without any filter whatsoever:
if token_ids is None and not additional_filter:
raise ValueError("Cannot operate without some filter.")
# Prepare the filter:
if not isinstance(token_ids, list): token_ids = [token_ids]
token_ids = [ObjectId(t) for t in token_ids]
filter_json = {"tokenId": {"$in": token_ids}}
filter_json = {}
if token_ids is not None:
if not isinstance(token_ids, list): token_ids = [token_ids]
filter_json["tokenId"] = {"$in": token_ids}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items():
filter_json[k] = v
for k, v in additional_filter.items(): filter_json[k] = v
# Get the count of the documents that match the criteria:
count = await mongo_conn.count(
count = await mongo_data_conn.count(
collection = self.MESSAGES_COLLECTION,
filter = filter_json,
raise_exception = True
@@ -198,42 +235,51 @@ class CoreMessageController(BaseModel):
# Done here:
return count
async def get_previews(
async def get_message_previews(
self,
mongo_conn: AsyncMongo,
token_ids: List[ObjectId | str],
mongo_data_conn: AsyncMongo,
token_ids: List[ObjectId | str] = None,
limit: int = 100,
skip: int = 0,
additional_filter: dict = None
additional_filter: dict = None,
projection: dict = None
) -> List[CoreMessageModel] | None:
"""
Fetches many messages in one call, but leaves out the full payloads.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param token_ids: The token ids of the accounts from which these messages must be fetched.
:param limit: The max. no. of messages to retrieve in this call.
:param skip: The no. of initial messages to skip. Useful for pagination.
:param additional_filter: Any addition filters to use.
:param projection: To decide what is picked from each document. WARNING: THIS MAY BREAK THE BEHAVIOUR OF THE
CORE MESSAGE MODEL. USE CAREFULLY.
:return: The list of messages (as the message model). This list can be empty.
"""
# We cannot allow counting without any filter whatsoever:
if token_ids is None and not additional_filter:
raise ValueError("Cannot operate without some filter.")
# Prepare the filter:
if not isinstance(token_ids, list): token_ids = [token_ids]
token_ids = [ObjectId(t) for t in token_ids]
filter_json = {"tokenId": {"$in": token_ids}}
filter_json = {}
if token_ids is not None:
if not isinstance(token_ids, list): token_ids = [token_ids]
filter_json["tokenId"] = {"$in": token_ids}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items():
filter_json[k] = v
for k, v in additional_filter.items(): filter_json[k] = v
# We fetch the messages that are identified by a specific token id,
# with the specified fetching limits, while enforcing the sorting condition:
records = await mongo_conn.find_many(
records = await mongo_data_conn.find_many(
collection = self.MESSAGES_COLLECTION,
filter = filter_json,
limit = limit,
skip = skip,
sort = {"ts": -1},
projection = {
projection = projection or {
"_id": True,
"ts": True,
"syncTs": True,
@@ -260,39 +306,49 @@ class CoreMessageController(BaseModel):
async def get_messages(
self,
mongo_conn: AsyncMongo,
token_ids: List[ObjectId | str],
mongo_data_conn: AsyncMongo,
token_ids: List[ObjectId | str] = None,
limit: int = 100,
skip: int = 0,
additional_filter: dict = None
additional_filter: dict = None,
projection: dict = None
) -> List[CoreMessageModel] | None:
"""
Fetches many full messages in one call.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param token_ids: The token ids of the accounts from which these messages must be fetched.
:param limit: The max. no. of messages to retrieve in this call.
:param skip: The no. of initial messages to skip. Useful for pagination.
:param additional_filter: Any addition filters to use.
:param projection: To decide what is picked from each document. WARNING: THIS MAY BREAK THE BEHAVIOUR OF THE
CORE MESSAGE MODEL. USE CAREFULLY.
:return: The list of messages (as the message model). This list can be empty.
"""
# We cannot allow counting without any filter whatsoever:
if token_ids is None and not additional_filter:
raise ValueError("Cannot operate without some filter.")
# Prepare the filter:
if not isinstance(token_ids, list): token_ids = [token_ids]
token_ids = [ObjectId(t) for t in token_ids]
filter_json = {"tokenId": {"$in": token_ids}}
filter_json = {}
if token_ids is not None:
if not isinstance(token_ids, list): token_ids = [token_ids]
filter_json["tokenId"] = {"$in": token_ids}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items():
filter_json[k] = v
for k, v in additional_filter.items(): filter_json[k] = v
# We fetch the messages that are identified by a specific token id,
# with the specified fetching limits, while enforcing the sorting condition:
records = await mongo_conn.find_many(
records = await mongo_data_conn.find_many(
collection = self.MESSAGES_COLLECTION,
filter = filter_json,
limit = limit,
skip = skip,
sort = {"ts": -1},
projection = projection,
raise_exception = True
)
@@ -301,21 +357,34 @@ class CoreMessageController(BaseModel):
async def get_message(
self,
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
message_id: ObjectId | str,
additional_filter: dict = None,
projection: dict = None
) -> CoreMessageModel | None:
"""
Gets one message if you know its message id.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param message_id: The id of the message that needs to be read.
:param additional_filter: Any addition filters to use.
:param projection: To decide what is picked from each document. WARNING: THIS MAY BREAK THE BEHAVIOUR OF THE
CORE MESSAGE MODEL. USE CAREFULLY.
:return: The contents of that one message in a structured format.
"""
# Start by preparing the filter:
filter_json = {"_id": ObjectId(message_id)}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items(): filter_json[k] = v
# We fetch the whole payload of that one message:
record = await mongo_conn.find_one(
record = await mongo_data_conn.find_one(
collection = self.MESSAGES_COLLECTION,
filter = {"_id": ObjectId(message_id)},
filter = filter_json,
projection = projection,
raise_exception = True
)
@@ -336,25 +405,34 @@ class CoreMessageController(BaseModel):
async def update_tags(
self,
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
message_id: ObjectId | str,
unset_tags: List[str] = None,
set_tags: List[str] = None
set_tags: List[str] = None,
additional_filter: dict = None
) -> bool:
"""
Updates the tags on one message. The tags to remove are processed first, the ones to add are processed later.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param message_id: The id of the message that needs to be read.
:param unset_tags: The tags to remove from the message.
:param set_tags: The tags to add to the message.
:param additional_filter: Any addition filters to use.
:return: True if the update was successful, else False.
"""
# Start by preparing the filter:
filter_json = {"_id": ObjectId(message_id)}
if self._base_filter:
for k, v in self._base_filter.items(): filter_json[k] = v
if additional_filter:
for k, v in additional_filter.items(): filter_json[k] = v
# Update the tags:
return await mongo_conn.update_one(
return await mongo_data_conn.update_one(
collection = self.MESSAGES_COLLECTION,
filter = {"_id": ObjectId(message_id)},
filter = filter_json,
update = [{
"$set": {
"tags": {