Remove load_envs module, refactor database implementations and upgrad…

…e pydantic

Deleted the 'load_envs.py' module as its functions are no longer needed. Modified the 'reporter.py' and 'media_cache.py' classes in the database package with refactoring to improve readability and structure. Upgraded the Pydantic package to a newer version for improved data validation and settings management. The 'poetry.lock' file updated to reflect the package changes.

app/commands/__init__.py

@@ -1,3 +1,3 @@
-from app.commands.commands import commands
+from .commands import commands
 
 connect_commands = commands.connect_commands

app/commands/commands.py

@@ -2,6 +2,7 @@
 import random
 from collections.abc import Callable
 
+from constants import settings
 from telegram import (
     ChatAdministratorRights,
     KeyboardButton,
@@ -11,11 +12,10 @@
 )
 from telegram.constants import ChatType
 
-from app import constants, settings
 from app.commands.registrator import CommandRegistrator
-from app.constants import DEFAULT_LOCALE
 from app.context import CallbackContext
 from app.parsers.base import Parser
+from app.settings import command_handler
 from app.utils import a, b
 from app.utils.i18n import _
 
@@ -56,24 +56,23 @@ def get(key: str) -> str | None:
                 return res
             if res := obj.get(f"{key}_{ctx.user_lang.split('-')[0]}"):
                 return res
-            if res := obj.get(f"{key}_{DEFAULT_LOCALE}"):
+            if res := obj.get(f"{key}_{settings.default_locale}"):
                 return res
             return obj.get(key)
 
         return get
 
+    contacts_list = "\n".join(
+        f'- {g("type")}: {a(g("text"), g("url"))}'  # type: ignore[arg-type]
+        for c in settings.contacts
+        if all(map(c.get, ("type", "text", "url")))
+        if (g := get_by_lang(c))  # type: ignore[arg-type]
+    )
     contacts = ""
-    if constants.CONTACTS:
-        contacts_list = "\n".join(
-            f'- {g("type")}: {a(g("text"), g("url"))}'  # type: ignore[arg-type]
-            for c in constants.CONTACTS
-            if all(map(c.get, ("type", "text", "url")))
-            if (g := get_by_lang(c))  # type: ignore[arg-type]
+    if contacts_list:
+        contacts = _("\n\nContacts:\n{contacts_list}").format(
+            contacts_list=contacts_list,
         )
-        if contacts_list:
-            contacts = _("\n\nContacts:\n{contacts_list}").format(
-                contacts_list=contacts_list,
-            )
 
     reply_markup = None
     rights = ChatAdministratorRights(
@@ -146,4 +145,4 @@ async def clear_history(update: Update, ctx: CallbackContext) -> None:
     await update.message.reply_text(_("History cleared."))
 
 
-commands.add_handler(settings.command_handler(), _("Bot settings"))
+commands.add_handler(command_handler(), _("Bot settings"))

app/constants/__init__.py

@@ -1,76 +1,5 @@
-import json
-import os
-import sys
-from pathlib import Path
-
-import pytz
-
-from .init_logger import init_logger_config
-from .json_logger import CONTEXT_VARS
-from .load_envs import load_envs
+from .context_vars import *
+from .init_logger import *
+from .paths import *
+from .settings import *
 from .types import *
-
-# region Base paths
-APP_PATH = Path(__file__).resolve().parent.parent
-PROJECT_PATH = APP_PATH.parent
-BASE_PATH = Path(os.getenv("BASE_PATH", PROJECT_PATH))
-
-CONFIG_PATH = BASE_PATH / "config"
-DATA_PATH = BASE_PATH / "data"
-LOG_PATH = DATA_PATH / "logs"
-
-CONFIG_PATH.mkdir(parents=True, exist_ok=True)
-DATA_PATH.mkdir(parents=True, exist_ok=True)
-LOG_PATH.mkdir(parents=True, exist_ok=True)
-# endregion
-
-# Load envs
-load_envs(BASE_PATH, CONFIG_PATH)
-
-# region Localizations
-LOCALE_PATH = PROJECT_PATH / "locales"
-DEFAULT_LOCALE = os.getenv("DEFAULT_LOCALE", "en")
-DOMAIN = os.getenv("LOCALE_DOMAIN", "messages")
-# endregion
-
-# region Other
-TOKEN = os.getenv("TG_TOKEN")  # Telegram token
-TIME_ZONE = pytz.timezone(os.getenv("TZ", "Europe/Moscow"))
-
-# Contacts for help command
-CONTACTS_PATH = Path(os.getenv("CONTACTS_PATH", CONFIG_PATH / "contacts.json"))
-REPORT_PATH = Path(os.getenv("REPORT_PATH", CONFIG_PATH / "report.json"))
-NOTIFY_PATH = Path(os.getenv("NOTIFY_PATH", CONFIG_PATH / "notify.json"))
-
-if not REPORT_PATH.parent.exists():
-    REPORT_PATH.parent.mkdir(parents=True)
-
-CONTACTS: list[CONTACT] = []
-
-if CONTACTS_PATH.exists():
-    with open(CONTACTS_PATH) as f:
-        CONTACTS = json.load(f)
-
-# Telegram file limit
-TG_FILE_LIMIT = 20 * 1024 * 1024  # 20 MB
-
-# LamadavaSaas API Token
-LAMADAVA_SAAS_TOKEN = os.getenv("LAMADAVA_SAAS_TOKEN", None)
-
-# region MongoDB support
-# mongodb://user:password@localhost:27017/database
-# mongodb://user:password@localhost:27017/
-MONGO_URL = os.getenv("MONGO_URL", None)
-MONGO_DB = os.getenv("MONGO_DB", None)
-
-ENABLE_MONGO = MONGO_URL and MONGO_DB
-if not ENABLE_MONGO:
-    print(
-        "Bot requires MongoDB to work.\n" "Please, set MONGO_URL and MONGO_DB envs.",
-        file=sys.stderr,
-    )
-    exit(1)
-# endregion
-
-# Load custom logger config
-init_logger_config(LOG_PATH, TIME_ZONE)

app/constants/context_vars.py

@@ -0,0 +1,20 @@
+from contextvars import ContextVar
+
+__all__ = (
+    "USER_ID",
+    "USERNAME",
+    "QUERY",
+    "DATA_TYPE",
+    "CONTEXT_VARS",
+)
+
+USER_ID: ContextVar[int] = ContextVar("USER_ID", default=0)
+USERNAME: ContextVar[str] = ContextVar("USERNAME", default="")
+QUERY: ContextVar[str] = ContextVar("QUERY", default="")
+DATA_TYPE: ContextVar[str] = ContextVar("DATA_TYPE", default="")
+CONTEXT_VARS: list[ContextVar] = [
+    USER_ID,
+    USERNAME,
+    QUERY,
+    DATA_TYPE,
+]

app/constants/fmt_logger.py

@@ -0,0 +1,23 @@
+import json
+
+from .json_logger import JsonFormatter
+
+__all__ = ("FmtFormatter",)
+
+
+def escape_string(string: str) -> str:
+    if " " in string or '"' in string or "'" in string:
+        return json.dumps(string)
+    return string
+
+
+class FmtFormatter(JsonFormatter):
+    def format(self, record) -> str:
+        """
+        Mostly the same as the parent's class method, the difference
+        being that a dict is manipulated and dumped as JSON
+        instead of a string.
+        """
+        message_dict: dict[str, str] = json.loads(super().format(record))
+
+        return " ".join(f"{k}={escape_string(v)}" for k, v in message_dict.items() if v)

app/constants/init_logger.py

@@ -5,7 +5,12 @@
 
 import pytz
 
+from .fmt_logger import FmtFormatter
 from .json_logger import JsonFormatter
+from .paths import LOG_PATH
+from .settings import settings
+
+__all__ = ("init_logger_config",)
 
 
 def init_logger_config(log_path: Path, time_zone: tzinfo = pytz.timezone("Europe/Moscow")) -> None:
@@ -15,16 +20,21 @@ def init_logger_config(log_path: Path, time_zone: tzinfo = pytz.timezone("Europe
         "disable_existing_loggers": False,
         "formatters": {
             "default": {
-                "format": ("%(asctime)s - %(name)s - %(levelname)s - %(message)s"),
-                "datefmt": "%Y-%m-%d %H:%M:%S",
+                "()": FmtFormatter,
+                "fmt_dict": {
+                    "timestamp": "asctime",
+                    "level": "levelname",
+                    "loggerName": "name",
+                    "message": "message",
+                },
             },
             "json": {
                 "()": JsonFormatter,
                 "fmt_dict": {
                     "timestamp": "asctime",
                     "level": "levelname",
-                    "message": "message",
                     "loggerName": "name",
+                    "message": "message",
                 },
             },
         },
@@ -76,3 +86,8 @@ def init_logger_config(log_path: Path, time_zone: tzinfo = pytz.timezone("Europe
         )
     else:
         logging.config.dictConfig(config)
+
+
+# Load custom logger config
+init_logger_config(LOG_PATH, settings.time_zone)
+logging.getLogger("httpx").setLevel(level=logging.ERROR)

app/constants/json_logger.py

@@ -1,20 +1,9 @@
 import json
 import logging.config
-from contextvars import ContextVar
 
-USER_ID: ContextVar[int] = ContextVar("USER_ID", default=0)
-USERNAME: ContextVar[str] = ContextVar("USERNAME", default="")
-QUERY: ContextVar[str] = ContextVar("QUERY", default="")
-DATA_TYPE: ContextVar[str] = ContextVar("DATA_TYPE", default="")
+from .context_vars import CONTEXT_VARS
 
-CONTEXT_VARS: list[ContextVar] = [
-    USER_ID,
-    USERNAME,
-    QUERY,
-    DATA_TYPE,
-]
-
-logger = logging.getLogger(__name__)
+__all__ = ("JsonFormatter",)
 
 
 class JsonFormatter(logging.Formatter):

app/constants/paths.py

@@ -0,0 +1,25 @@
+import os
+from pathlib import Path
+
+__all__ = (
+    "APP_PATH",
+    "PROJECT_PATH",
+    "BASE_PATH",
+    "DATA_PATH",
+    "LOG_PATH",
+    "CONFIG_PATH",
+    "LOCALE_PATH",
+)
+
+APP_PATH = Path(__file__).resolve().parent.parent
+PROJECT_PATH = APP_PATH.parent
+BASE_PATH = Path(os.getenv("BASE_PATH", PROJECT_PATH))
+
+DATA_PATH = BASE_PATH / "data"
+LOG_PATH = DATA_PATH / "logs"
+CONFIG_PATH = BASE_PATH / "config"
+LOCALE_PATH = PROJECT_PATH / "locales"
+
+DATA_PATH.mkdir(parents=True, exist_ok=True)
+LOG_PATH.mkdir(parents=True, exist_ok=True)
+CONFIG_PATH.mkdir(parents=True, exist_ok=True)

app/constants/settings.py

@@ -0,0 +1,63 @@
+import functools
+import json
+from pathlib import Path
+from typing import Self
+
+import pytz
+from pydantic import ByteSize, Field, MongoDsn, SecretStr, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from .paths import BASE_PATH, CONFIG_PATH
+from .types import CONTACT
+
+__all__ = (
+    "Settings",
+    "settings",
+)
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=[
+            CONFIG_PATH / ".env.local",
+            CONFIG_PATH / ".env",
+            BASE_PATH / ".env.local",
+            BASE_PATH / ".env",
+        ],
+        env_ignore_empty=True,
+        extra="ignore",
+    )
+
+    default_locale: str = Field("en")
+    domain: str = Field("messages")
+
+    token: SecretStr = Field(alias="TG_TOKEN", description="Telegram Token")
+    time_zone: pytz.tzinfo.BaseTzInfo = Field(pytz.timezone("Europe/Moscow"), alias="TZ")
+
+    tg_file_size: ByteSize = Field("20 MiB", description="Telegram file size")
+
+    report_path: Path = Field(CONFIG_PATH / "report.json")
+    contacts_path: Path = Field(CONFIG_PATH / "contacts.json")
+
+    @functools.cached_property
+    def contacts(self) -> list[CONTACT]:
+        if self.contacts_path.exists():
+            with self.contacts_path.open() as f:
+                return json.load(f)
+        return []
+
+    mongo_url: MongoDsn | None = Field(None)
+    mongo_db: str | None = Field(None)
+
+    @model_validator(mode="after")
+    def mongo_require(self) -> Self:
+        if self.mongo_url is None or self.mongo_db is None:
+            raise ValueError("Bot requires MongoDB to work. Please, set MONGO_URL and MONGO_DB.")
+        self.report_path.parent.mkdir(exist_ok=True, parents=True)
+        return self
+
+
+settings = Settings()
+
+if __name__ == "__main__":
+    print(repr(settings))