Enhance code quality

docs/source/conf.py

@@ -11,6 +11,7 @@
 # documentation root, use os.path.abspath to make it absolute, like shown here.
 
 import os
+import re
 
 from sphinx.ext import apidoc
 
@@ -62,14 +63,63 @@
 # so a file named "default.css" will overwrite the builtin "default.css".
 # html_static_path = ["_static"]
 
+
+# Specify which special members have to be kept
+special_members_dict = {
+    "Document": {"init"},
+    "ResponseError": {"init", "or"},
+    "PipelineBooleanDict": {"init", "or", "and"},
+    "PipelineAttribute": {"init", "or", "and", "eq", "gt", "ge", "lt", "le"},
+    "Pipelines": {"init", "add", "iadd"}
+}
+
+# Add trailing and leading "__" to all the aforementioned members
+for cls, methods in special_members_dict.items():
+    special_members_dict[cls] = {f"__{method}__" for method in methods}
+
+# Make a set of all allowed special members
+all_special_members = set()
+for methods in special_members_dict.values():
+    all_special_members |= methods
+
 autodoc_default_options = {
     "members": True,
     "member-order": "bysource",
     "private-members": True,
     "special-members": True,
-    "undoc-members": True,
+    "undoc-members": False,
 }
 
+
+def is_special_member(member_name: str) -> bool:
+    """Checks if the given member is special, i.e. its name has the following format ``__<some-str>__``."""
+    return bool(re.compile(r"^__\w+__$").match(member_name))
+
+
+def skip(app, typ, member_name, obj, flag, options):
+    """The filter function to determine whether to keep the member in the documentation.
+
+    ``True`` means skip the member.
+    """
+    if is_special_member(member_name):
+
+        if member_name not in all_special_members:
+            return True
+
+        obj_name = obj.__qualname__.split(".")[0]
+        if methods_set := special_members_dict.get(obj_name, None):
+            if member_name in methods_set:
+                return False  # Keep the member
+        return True
+
+    return None
+
+
+def setup(app):
+    """Sets up the sphinx app."""
+    app.connect("autodoc-skip-member", skip)
+
+
 root_doc = "index"
 
 output_dir = os.path.join(".")

docs/source/index.rst

@@ -1,4 +1,4 @@
-Welcome to Pytroll documentation!
+Pytroll-db Documentation
 ===========================================
 
 .. toctree::

trolldb/__init__.py

@@ -1 +1 @@
-"""trolldb package."""
+"""The database interface of the Pytroll package."""

trolldb/api/__init__.py

@@ -5,5 +5,5 @@
 
 For more information and documentation, please refer to the following sub-packages and modules:
   - :obj:`trolldb.api.routes`: The package which defines the API routes.
-  - :obj:`trollddb.api.api`: The module which defines the API server and how it is run via the given configuration.
+  - :obj:`trolldb.api.api`: The module which defines the API server and how it is run via the given configuration.
 """

trolldb/api/api.py

@@ -14,27 +14,28 @@
 """
 
 import asyncio
+import sys
 import time
 from contextlib import contextmanager
 from multiprocessing import Process
-from typing import Union
+from typing import Any, Generator, NoReturn, Union
 
 import uvicorn
 from fastapi import FastAPI, status
 from fastapi.responses import PlainTextResponse
 from loguru import logger
-from pydantic import FilePath, validate_call
+from pydantic import FilePath, ValidationError, validate_call
 
 from trolldb.api.routes import api_router
-from trolldb.config.config import AppConfig, Timeout, parse_config_yaml_file
+from trolldb.config.config import AppConfig, Timeout, parse_config
 from trolldb.database.mongodb import mongodb_context
 from trolldb.errors.errors import ResponseError
 
 API_INFO = dict(
     title="pytroll-db",
     summary="The database API of Pytroll",
     description=
-    "The API allows   you to perform CRUD operations as well as querying the database"
+    "The API allows you to perform CRUD operations as well as querying the database"
     "At the moment only MongoDB is supported. It is based on the following Python packages"
     "\n * **PyMongo** (https://github.com/mongodb/mongo-python-driver)"
     "\n * **motor** (https://github.com/mongodb/motor)",
@@ -43,9 +44,10 @@
         url="https://www.gnu.org/licenses/gpl-3.0.en.html"
     )
 )
-"""These will appear int the auto-generated documentation and are passed to the ``FastAPI`` class as keyword args."""
+"""These will appear in the auto-generated documentation and are passed to the ``FastAPI`` class as keyword args."""
 
 
+@logger.catch(onerror=lambda _: sys.exit(1))
 @validate_call
 def run_server(config: Union[AppConfig, FilePath], **kwargs) -> None:
     """Runs the API server with all the routes and connection to the database.
@@ -65,31 +67,27 @@
             `FastAPI class <https://fastapi.tiangolo.com/reference/fastapi/#fastapi.FastAPI>`_ and are directly passed
             to it. These keyword arguments will be first concatenated with the configurations of the API server which
             are read from the ``config`` argument. The keyword arguments which are passed explicitly to the function
-            take precedence over ``config``. Finally, ``API_INFO``, which are hard-coded information for the API server,
-            will be concatenated and takes precedence over all.
-
-    Raises:
-        ValidationError:
-            If the function is not called with arguments of valid type.
+            take precedence over ``config``. Finally, :obj:`API_INFO`, which are hard-coded information for the API
+            server, will be concatenated and takes precedence over all.
 
     Example:
         .. code-block:: python
 
-            from api.api import run_server
+            from trolldb.api.api import run_server
             if __name__ == "__main__":
                 run_server("config.yaml")
     """
     logger.info("Attempt to run the API server ...")
     if not isinstance(config, AppConfig):
-        config = parse_config_yaml_file(config)
+        config = parse_config(config)
 
     # Concatenate the keyword arguments for the API server in the order of precedence (lower to higher).
     app = FastAPI(**(config.api_server._asdict() | kwargs | API_INFO))
 
     app.include_router(api_router)
 
     @app.exception_handler(ResponseError)
-    async def auto_exception_handler(_, exc: ResponseError):
+    async def auto_handler_response_errors(_, exc: ResponseError) -> PlainTextResponse:
         """Catches all the exceptions raised as a ResponseError, e.g. accessing non-existing databases/collections."""
         status_code, message = exc.get_error_details()
         info = dict(
@@ -99,7 +97,13 @@
         logger.error(f"Response error caught by the API auto exception handler: {info}")
         return PlainTextResponse(**info)
 
-    async def _serve():
+    @app.exception_handler(ValidationError)
+    async def auto_handler_pydantic_validation_errors(_, exc: ValidationError) -> PlainTextResponse:
+        """Catches all the exceptions raised as a Pydantic ValidationError."""
+        logger.error(f"Response error caught by the API auto exception handler: {exc}")
+        return PlainTextResponse(str(exc), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    async def _serve() -> NoReturn:
         """An auxiliary coroutine to be used in the asynchronous execution of the FastAPI application."""
         async with mongodb_context(config.database):
             logger.info("Attempt to start the uvicorn server ...")
@@ -116,7 +120,8 @@
 
 
 @contextmanager
-def api_server_process_context(config: Union[AppConfig, FilePath], startup_time: Timeout = 2):
+def api_server_process_context(
+        config: Union[AppConfig, FilePath], startup_time: Timeout = 2) -> Generator[Process, Any, None]:
     """A synchronous context manager to run the API server in a separate process (non-blocking).
 
     It uses the `multiprocessing <https://docs.python.org/3/library/multiprocessing.html>`_ package. The main use case
@@ -133,7 +138,7 @@
     """
     logger.info("Attempt to run the API server process in a context manager ...")
     if not isinstance(config, AppConfig):
-        config = parse_config_yaml_file(config)
+        config = parse_config(config)
 
     process = Process(target=run_server, args=(config,))
     try:

trolldb/api/routes/__init__.py

@@ -1,4 +1,4 @@
-"""routes package."""
+"""The routes package of the API."""
 
 from .router import api_router
 

trolldb/api/routes/common.py

@@ -2,19 +2,11 @@
 
 from typing import Annotated, Union
 
-from fastapi import Depends, Query, Response
+from fastapi import Depends, Response
 from motor.motor_asyncio import AsyncIOMotorCollection, AsyncIOMotorDatabase
 
 from trolldb.database.mongodb import MongoDB
 
-exclude_defaults_query = Query(
-    True,
-    title="Query string",
-    description=
-    "A boolean to exclude default databases from a MongoDB instance. Refer to "
-    "`trolldb.database.mongodb.MongoDB.default_database_names` for more information."
-)
-
 
 async def check_database(database_name: str | None = None) -> AsyncIOMotorDatabase:
     """A dependency for route handlers to check for the existence of a database given its name.

trolldb/api/routes/databases.py

@@ -4,10 +4,12 @@
     For more information on the API server, see the automatically generated documentation by FastAPI.
 """
 
-from fastapi import APIRouter
+from typing import Annotated
+
+from fastapi import APIRouter, Query
 from pymongo.collection import _DocumentType
 
-from trolldb.api.routes.common import CheckCollectionDependency, CheckDataBaseDependency, exclude_defaults_query
+from trolldb.api.routes.common import CheckCollectionDependency, CheckDataBaseDependency
 from trolldb.config.config import MongoObjectId
 from trolldb.database.errors import (
     Databases,
@@ -23,7 +25,12 @@
 @router.get("/",
             response_model=list[str],
             summary="Gets the list of all database names")
-async def database_names(exclude_defaults: bool = exclude_defaults_query) -> list[str]:
+async def database_names(
+        exclude_defaults: Annotated[bool, Query(
+            title="Query parameter",
+            description="A boolean to exclude default databases from a MongoDB instance. Refer to "
+                        "`trolldb.database.mongodb.MongoDB.default_database_names` for more information."
+        )] = True) -> list[str]:
     """Please consult the auto-generated documentation by FastAPI."""
     db_names = await MongoDB.list_database_names()
 

trolldb/api/routes/queries.py

@@ -5,13 +5,14 @@
 """
 
 import datetime
+from typing import Annotated
 
 from fastapi import APIRouter, Query
 
 from trolldb.api.routes.common import CheckCollectionDependency
 from trolldb.database.errors import database_collection_error_descriptor
 from trolldb.database.mongodb import get_ids
-from trolldb.database.piplines import PipelineAttribute, Pipelines
+from trolldb.database.pipelines import PipelineAttribute, Pipelines
 
 router = APIRouter()
 
@@ -22,14 +23,11 @@
             summary="Gets the database UUIDs of the documents that match specifications determined by the query string")
 async def queries(
         collection: CheckCollectionDependency,
-        # We suppress ruff for the following four lines with `Query(default=None)`.
-        # Reason: This is the FastAPI way of defining optional queries and ruff is not happy about it!
-        platform: list[str] = Query(default=None),  # noqa: B008
-        sensor: list[str] = Query(default=None),  # noqa: B008
-        time_min: datetime.datetime = Query(default=None),  # noqa: B008
-        time_max: datetime.datetime = Query(default=None)) -> list[str]:  # noqa: B008
+        platform: Annotated[list[str] | None, Query()] = None,
+        sensor: Annotated[list[str] | None, Query()] = None,
+        time_min: Annotated[datetime.datetime, Query()] = None,
+        time_max: Annotated[datetime.datetime, Query()] = None) -> list[str]:
     """Please consult the auto-generated documentation by FastAPI."""
-    # We
     pipelines = Pipelines()
 
     if platform:
@@ -42,10 +40,7 @@ async def queries(
         start_time = PipelineAttribute("start_time")
         end_time = PipelineAttribute("end_time")
         pipelines += (
-                (start_time >= time_min) |
-                (start_time <= time_max) |
-                (end_time >= time_min) |
-                (end_time <= time_max)
+                ((start_time >= time_min) & (start_time <= time_max)) |
+                ((end_time >= time_min) & (end_time <= time_max))
         )
-
     return await get_ids(collection.aggregate(pipelines))

trolldb/cli.py

@@ -4,39 +4,68 @@
 import asyncio
 
 from loguru import logger
+from motor.motor_asyncio import AsyncIOMotorCollection
 from posttroll.message import Message
 from posttroll.subscriber import create_subscriber_from_dict_config
 from pydantic import FilePath
 
-from trolldb.config.config import AppConfig, parse_config_yaml_file
+from trolldb.config.config import AppConfig, parse_config
 from trolldb.database.mongodb import MongoDB, mongodb_context
 
 
-async def record_messages(config: AppConfig):
+async def delete_uri_from_collection(collection: AsyncIOMotorCollection, uri: str) -> int:
+    """Deletes a document from collection and logs the deletion.
+
+    Args:
+        collection:
+            The collection object which includes the document to delete.
+        uri:
+            The URI used to query the collection. It can be either a URI of a previously recorded file message or
+            a dataset message.
+
+    Returns:
+         Number of deleted documents.
+    """
+    del_result_file = await collection.delete_many({"uri": uri})
+    if del_result_file.deleted_count == 1:
+        logger.info(f"Deleted one document (file) with uri: {uri}")
+
+    del_result_dataset = await collection.delete_many({"dataset.uri": uri})
+    if del_result_dataset.deleted_count == 1:
+        logger.info(f"Deleted one document (dataset) with uri: {uri}")
+
+    return del_result_file.deleted_count + del_result_dataset.deleted_count
+
+
+async def record_messages(config: AppConfig) -> None:
     """Record the metadata of messages into the database."""
     async with mongodb_context(config.database):
         collection = await MongoDB.get_collection(
             config.database.main_database_name, config.database.main_collection_name
         )
         for m in create_subscriber_from_dict_config(config.subscriber).recv():
             msg = Message.decode(str(m))
-            if msg.type in ["file", "dataset"]:
-                await collection.insert_one(msg.data)
-            elif msg.type == "del":
-                deletion_result = await collection.delete_many({"uri": msg.data["uri"]})
-                if deletion_result.deleted_count != 1:
-                    logger.error("Recorder found multiple deletions!")  # TODO: Log some data related to the msg
-            else:
-                logger.debug(f"Don't know what to do with {msg.type} message.")
+            match msg.type:
+                case "file":
+                    await collection.insert_one(msg.data)
+                    logger.info(f"Inserted file with uri: {msg.data["uri"]}")
+                case "dataset":
+                    await collection.insert_one(msg.data)
+                    logger.info(f"Inserted dataset with {len(msg.data["dataset"])} elements.")
+                case "del":
+                    deletion_count = await delete_uri_from_collection(collection, msg.data["uri"])
+                    if deletion_count > 1:
+                        logger.error(f"Recorder found multiple deletions for uri: {msg.data["uri"]}!")
+                case _:
+                    logger.debug(f"Don't know what to do with {msg.type} message.")
 
 
-async def record_messages_from_config(config_file: FilePath):
+async def record_messages_from_config(config_file: FilePath) -> None:
     """Record messages into the database, getting the configuration from a file."""
-    config = parse_config_yaml_file(config_file)
-    await record_messages(config)
+    await record_messages(parse_config(config_file))
 
 
-async def record_messages_from_command_line(args=None):
+async def record_messages_from_command_line(args=None) -> None:
     """Record messages into the database, command-line interface."""
     parser = argparse.ArgumentParser()
     parser.add_argument(
@@ -47,6 +76,6 @@
     await record_messages_from_config(cmd_args.configuration_file)
 
 
-def run_sync():
+def run_sync() -> None:
     """Runs the interface synchronously."""
     asyncio.run(record_messages_from_command_line())