Reorganized Mongo DB API

chat_server/blueprints/admin.py

@@ -30,11 +30,11 @@
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
-from chat_server.server_utils.db_utils import DbUtils
+from utils.database_utils.mongo_utils.queries.wrapper import MongoDocumentsAPI
 from utils.logging_utils import LOG
 from utils.http_utils import respond
 
-from chat_server.server_config import k8s_config, db_controller
+from chat_server.server_config import k8s_config
 from chat_server.server_utils.auth import login_required
 from chat_server.server_utils.k8s_utils import restart_deployment
 from chat_server.server_utils.admin_utils import run_mq_validation
@@ -79,7 +79,7 @@ async def refresh_state(
 @router.get("/chats/list")
 @login_required(tmp_allowed=False, required_roles=["admin"])
 async def chats_overview(request: Request, search_str: str = ""):
-    conversations_data = DbUtils.get_conversation_data(
+    conversations_data = MongoDocumentsAPI.CHATS.get_conversation_data(
         search_str=search_str,
         limit=100,
         allow_regex_search=True,

chat_server/blueprints/auth.py

@@ -31,13 +31,13 @@
 from fastapi import APIRouter, Form, Request
 from fastapi.responses import JSONResponse
 
-from chat_server.server_config import db_controller
 from utils.common import get_hash, generate_uuid
 from chat_server.server_utils.auth import (
     check_password_strength,
     get_current_user_data,
     generate_session_token,
 )
+from utils.database_utils.mongo_utils.queries.wrapper import MongoDocumentsAPI
 from utils.http_utils import respond
 
 router = APIRouter(
@@ -64,13 +64,7 @@ async def signup(
     :returns JSON response with status corresponding to the new user creation status,
              sets session cookies if creation is successful
     """
-    existing_user = db_controller.exec_query(
-        query={
-            "command": "find_one",
-            "document": "users",
-            "data": {"nickname": nickname},
-        }
-    )
+    existing_user = MongoDocumentsAPI.USERS.get_user(nickname=nickname)
     if existing_user:
         return respond("Nickname is already in use", 400)
     password_check = check_password_strength(password)
@@ -85,9 +79,7 @@ async def signup(
         date_created=int(time()),
         is_tmp=False,
     )
-    db_controller.exec_query(
-        query=dict(document="users", command="insert_one", data=new_user_record)
-    )
+    MongoDocumentsAPI.USERS.add_item(data=new_user_record)
 
     token = generate_session_token(user_id=new_user_record["_id"])
 
@@ -104,13 +96,7 @@ async def login(username: str = Form(...), password: str = Form(...)):
 
     :returns JSON response with status corresponding to authorization status, sets session cookie with response
     """
-    user = db_controller.exec_query(
-        query={
-            "command": "find_one",
-            "document": "users",
-            "data": {"nickname": username},
-        }
-    )
+    user = MongoDocumentsAPI.USERS.get_user(nickname=username)
     if not user or user.get("is_tmp", False):
         return respond("Invalid username or password", 400)
     db_password = user["password"]

chat_server/blueprints/chat.py

@@ -25,24 +25,20 @@
 # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 # NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 # SOFTWARE,  EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+import warnings
 from typing import Optional
 
 from time import time
 from fastapi import APIRouter, Request, Form
 from fastapi.responses import JSONResponse
 
 from chat_server.constants.conversations import ConversationSkins
-from chat_server.server_config import db_controller
 from chat_server.server_utils.auth import login_required
 from chat_server.server_utils.conversation_utils import build_message_json
-from chat_server.server_utils.db_utils import (
-    DbUtils,
-    MongoQuery,
-    MongoCommands,
-    MongoDocuments,
-)
 from chat_server.services.popularity_counter import PopularityCounter
 from utils.common import generate_uuid
+from utils.database_utils.mongo_utils.queries.mongo_queries import fetch_message_data
+from utils.database_utils.mongo_utils.queries.wrapper import MongoDocumentsAPI
 from utils.http_utils import respond
 from utils.logging_utils import LOG
 
@@ -56,7 +52,7 @@
 @login_required
 async def new_conversation(
     request: Request,
-    conversation_id: str = Form(""),
+    conversation_id: str = Form(""),  # DEPRECATED
     conversation_name: str = Form(...),
     is_private: str = Form(False),
     bound_service: str = Form(""),
@@ -65,34 +61,31 @@ async def new_conversation(
     Creates new conversation from provided conversation data
 
     :param request: Starlette Request object
-    :param conversation_id: new conversation id (optional)
+    :param conversation_id: new conversation id (DEPRECATED)
     :param conversation_name: new conversation name (optional)
     :param is_private: if new conversation should be private (defaults to False)
     :param bound_service: name of the bound service (ignored if empty value)
 
     :returns JSON response with new conversation data if added, 401 error message otherwise
     """
-
-    conversation_data = DbUtils.get_conversation_data(
-        search_str=[conversation_id, conversation_name]
+    if conversation_id:
+        warnings.warn(
+            "Param conversation id is no longer considered", DeprecationWarning
+        )
+    conversation_data = MongoDocumentsAPI.CHATS.get_conversation_data(
+        search_str=[conversation_name],
     )
     if conversation_data:
         return respond(f'Conversation "{conversation_name}" already exists', 400)
-    cid = conversation_id or generate_uuid()
+    cid = generate_uuid()
     request_data_dict = {
         "_id": cid,
         "conversation_name": conversation_name,
         "is_private": True if is_private == "1" else False,
         "bound_service": bound_service,
         "created_on": int(time()),
     }
-    db_controller.exec_query(
-        query=MongoQuery(
-            command=MongoCommands.INSERT_ONE,
-            document=MongoDocuments.CHATS,
-            data=request_data_dict,
-        )
-    )
+    MongoDocumentsAPI.CHATS.add_item(data=request_data_dict)
     PopularityCounter.add_new_chat(cid=cid, name=conversation_name)
     return JSONResponse(content=request_data_dict)
 
@@ -119,13 +112,15 @@ async def get_matching_conversation(
 
     :returns conversation data if found, 401 error code otherwise
     """
-    conversation_data = DbUtils.get_conversation_data(search_str=search_str)
+    conversation_data = MongoDocumentsAPI.CHATS.get_conversation_data(
+        search_str=search_str
+    )
 
     if not conversation_data:
         return respond(f'No conversation matching = "{search_str}"', 404)
 
     message_data = (
-        DbUtils.fetch_skin_message_data(
+        fetch_message_data(
             skin=skin,
             conversation_data=conversation_data,
             start_idx=chat_history_from,

chat_server/blueprints/files_api.py

@@ -31,10 +31,9 @@
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
-from chat_server.server_config import db_controller
 from chat_server.server_utils.auth import login_required
-from chat_server.server_utils.db_utils import DbUtils
 from chat_server.server_utils.http_utils import get_file_response, save_file
+from utils.database_utils.mongo_utils.queries.wrapper import MongoDocumentsAPI
 from utils.http_utils import respond
 from utils.logging_utils import LOG
 
@@ -50,11 +49,11 @@ async def get_audio_message(
     message_id: str,
 ):
     """Gets file based on the name"""
-    matching_shouts = DbUtils.fetch_shouts(shout_ids=[message_id], fetch_senders=False)
-    if matching_shouts and matching_shouts[0].get("is_audio", "0") == "1":
+    matching_shout = MongoDocumentsAPI.SHOUTS.get_item(item_id=message_id)
+    if matching_shout and matching_shout.get("is_audio", "0") == "1":
         LOG.info(f"Fetching audio for message_id={message_id}")
         return get_file_response(
-            matching_shouts[0]["message_text"],
+            matching_shout["message_text"],
             location_prefix="audio",
             media_type="audio/wav",
         )
@@ -70,12 +69,7 @@ async def get_avatar(user_id: str):
     :param user_id: target user id
     """
     LOG.debug(f"Getting avatar of user id: {user_id}")
-    user_data = (
-        db_controller.exec_query(
-            query={"document": "users", "command": "find_one", "data": {"_id": user_id}}
-        )
-        or {}
-    )
+    user_data = MongoDocumentsAPI.USERS.get_user(user_id=user_id) or {}
     if user_data.get("avatar", None):
         num_attempts = 0
         try:
@@ -101,13 +95,11 @@ async def get_message_attachment(request: Request, msg_id: str, filename: str):
     :param filename: name of the file to get
     """
     LOG.debug(f"{msg_id} - {filename}")
-    message_files = db_controller.exec_query(
-        query={"document": "shouts", "command": "find_one", "data": {"_id": msg_id}}
-    )
-    if message_files:
+    shout_data = MongoDocumentsAPI.SHOUTS.get_item(item_id=msg_id)
+    if shout_data:
         attachment_data = [
             attachment
-            for attachment in message_files["attachments"]
+            for attachment in shout_data["attachments"]
             if attachment["name"] == filename
         ][0]
         media_type = attachment_data["mime"]

chat_server/blueprints/preferences.py

@@ -27,9 +27,8 @@
 # SOFTWARE,  EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 from fastapi import APIRouter, Request, Form
-from chat_server.server_config import db_controller
 from chat_server.server_utils.auth import get_current_user, login_required
-from chat_server.server_utils.db_utils import DbUtils
+from utils.database_utils.mongo_utils.queries.wrapper import MongoDocumentsAPI
 from utils.http_utils import respond
 from utils.logging_utils import LOG
 
@@ -50,7 +49,7 @@ async def update_language(
     except Exception as ex:
         LOG.error(ex)
         return respond(f"Failed to update language of {cid}/{input_type} to {lang}")
-    DbUtils.set_user_preferences(
+    MongoDocumentsAPI.USERS.set_preferences(
         user_id=current_user_id,
         preferences_mapping={f"chat_language_mapping.{cid}.{input_type}": lang},
     )

chat_server/blueprints/users.py

@@ -38,9 +38,9 @@
     get_current_user_data,
     login_required,
 )
-from chat_server.server_utils.db_utils import DbUtils
 from chat_server.server_utils.http_utils import save_file
 from utils.common import get_hash
+from utils.database_utils.mongo_utils.queries.wrapper import MongoDocumentsAPI
 from utils.http_utils import respond
 from utils.logging_utils import LOG
 
@@ -69,9 +69,7 @@ async def get_user(
     """
     session_token = ""
     if user_id:
-        user = db_controller.exec_query(
-            query={"document": "users", "command": "find_one", "data": {"_id": user_id}}
-        )
+        user = MongoDocumentsAPI.USERS.get_user(user_id=user_id)
         user.pop("password", None)
         user.pop("date_created", None)
         user.pop("tokens", None)
@@ -109,9 +107,8 @@ async def fetch_received_user_ids(
     if nicknames:
         filter_data["nickname"] = {"$in": nicknames.split(",")}
 
-    users = db_controller.exec_query(
-        query={"document": "users", "command": "find", "data": filter_data},
-        as_cursor=False,
+    users = MongoDocumentsAPI.USERS.list_items(
+        filters=filter_data, result_as_cursor=False
     )
     for user in users:
         user.pop("password", None)
@@ -209,7 +206,7 @@ async def update_settings(
     """
     user = get_current_user(request=request)
     preferences_mapping = {"minify_messages": minify_messages}
-    DbUtils.set_user_preferences(
+    MongoDocumentsAPI.USERS.set_preferences(
         user_id=user["_id"], preferences_mapping=preferences_mapping
     )
     return respond(msg="OK")

chat_server/server_config.py

@@ -29,25 +29,24 @@
 import os
 from typing import Optional
 
-from utils.logging_utils import LOG
 
 from config import Configuration
 from chat_server.server_utils.sftp_utils import init_sftp_connector
 from chat_server.server_utils.rmq_utils import RabbitMQAPI
 
 from utils.logging_utils import LOG
 from utils.database_utils import DatabaseController
+from utils.database_utils.mongo_utils.queries.wrapper import MongoDocumentsAPI
 
-server_config_path = os.path.expanduser(os.environ.get(
-    "CHATSERVER_CONFIG", "~/.local/share/neon/credentials.json"
-))
-database_config_path = os.path.expanduser(os.environ.get(
-    "DATABASE_CONFIG", "~/.local/share/neon/credentials.json"
-))
+server_config_path = os.path.expanduser(
+    os.environ.get("CHATSERVER_CONFIG", "~/.local/share/neon/credentials.json")
+)
+database_config_path = os.path.expanduser(
+    os.environ.get("DATABASE_CONFIG", "~/.local/share/neon/credentials.json")
+)
 
 
 def _init_db_controller(db_config: dict) -> Optional[DatabaseController]:
-    from chat_server.server_utils.db_utils import DbUtils
 
     # Determine configured database dialect
     dialect = db_config.pop("dialect", "mongo")
@@ -57,42 +56,40 @@ def _init_db_controller(db_config: dict) -> Optional[DatabaseController]:
         db_controller = DatabaseController(config_data=db_config)
         db_controller.attach_connector(dialect=dialect)
         db_controller.connect()
+        return db_controller
     except Exception as e:
         LOG.exception(f"DatabaseController init failed: {e}")
         return None
 
-    # Initialize convenience class
-    DbUtils.init(db_controller)
-    return db_controller
-
 
 if os.path.isfile(server_config_path) or os.path.isfile(database_config_path):
     LOG.warning(f"Using legacy configuration at {server_config_path}")
     LOG.warning(f"Using legacy configuration at {database_config_path}")
     LOG.info(f"KLAT_ENV : {Configuration.KLAT_ENV}")
-    config = Configuration(from_files=[server_config_path,
-                                       database_config_path])
+    config = Configuration(from_files=[server_config_path, database_config_path])
     app_config = config.get("CHAT_SERVER", {}).get(Configuration.KLAT_ENV, {})
     db_controller = config.get_db_controller(name="pyklatchat_3333")
 else:
     # ovos-config has built-in mechanisms for loading configuration files based
     # on envvars, so the configuration structure is simplified
     from ovos_config.config import Configuration
+
     config = Configuration()
     app_config = config.get("CHAT_SERVER") or dict()
     env_spec = os.environ.get("KLAT_ENV")
     if env_spec and app_config.get(env_spec):
         LOG.warning("Legacy configuration handling KLAT_ENV envvar")
         app_config = app_config.get(env_spec)
-    db_controller = _init_db_controller(app_config.get("connection_properties",
-                                                       config.get(
-                                                           "DATABASE_CONFIG",
-                                                           {})))
+    db_controller = _init_db_controller(
+        app_config.get("connection_properties", config.get("DATABASE_CONFIG", {}))
+    )
 
 LOG.info(f"App config: {app_config}")
 
 sftp_connector = init_sftp_connector(config=app_config.get("SFTP", {}))
 
+MongoDocumentsAPI.init(db_controller=db_controller, sftp_connector=sftp_connector)
+
 mq_api = None
 mq_management_config = config.get("MQ_MANAGEMENT", {})
 if mq_management_url := mq_management_config.get("MQ_MANAGEMENT_URL"):