added is_production and overrides

monitor_utils/config.py

@@ -1,40 +1,32 @@
 import json
-from enum import Enum
-from functools import cache
 import pathlib
+from functools import cache
 
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-from monitor_utils.constants import PROXIES_PATH, PROJECT_ROOT
+from monitor_utils.constants import (
+    PROXIES_PATH,
+    PROJECT_ROOT,
+    DEVELOPMENT_MODE,
+    PRODUCTION_BOT_TOKEN,
+    DEVELOPMENT_BOT_TOKEN,
+    DevelopmentMode,
+)
 
 
-class DevelopmentMode(str, Enum):
-    PRODUCTION = "PRODUCTION"
-    DEVELOPMENT = "DEVELOPMENT"
+def is_production(override_production=False) -> bool:
+    if override_production:
+        return override_production
+
+    return DEVELOPMENT_MODE == DevelopmentMode.PRODUCTION
 
 
-class Settings(BaseSettings):
-    mongo_url: str
-    mongo_database_name: str
-    production_bot_token: str
-    profit_api_key: str
-    development_bot_token: str | None = None
-    logging_level: str = "INFO"
-    development_mode: DevelopmentMode = DevelopmentMode.PRODUCTION
-
-    model_config = SettingsConfigDict(env_file=(".env", ".env.local"), extra="ignore")
-
-    @property
-    def bot_token(self) -> str:
-        if self.development_mode == DevelopmentMode.PRODUCTION:
-            return self.production_bot_token
-
-        return self.development_bot_token
+IS_PRODUCTION = is_production()
 
 
-@cache
-def get_settings() -> Settings:
-    return Settings()
+def get_bot_token() -> str:
+    if IS_PRODUCTION:
+        return PRODUCTION_BOT_TOKEN
+
+    return DEVELOPMENT_BOT_TOKEN
 
 
 @cache

monitor_utils/constants.py

@@ -1,5 +1,6 @@
 import asyncio
 import pathlib
+from enum import Enum
 
 import aiohttp
 import requests
@@ -8,7 +9,26 @@ from pymongo.errors import PyMongoError
 
 PROJECT_ROOT = pathlib.Path(".")
 PROXIES_PATH = pathlib.Path("proxies.txt")
+MONGO_DATABASE_NAME = "tpv_woot"
+MONGO_URL = (
+    "mongodb://tpv:n3%23QI07GnbZCbvGJ@zelda.disqualifier.me:27017/?authSource=tpv_woot"
+)
+PRODUCTION_BOT_TOKEN = (
+    "MTQwMDE4NDk2MzY4NDYzMDUyOQ.GOrqOb.6frT3upUQ5B9XubVOIT9pdzVcYVXSrnEMK7RlA"
+)
+DEVELOPMENT_BOT_TOKEN = (
+    "MTUxNzYwNDI4NDEwMTk1MTUxOA.GO3b1i.aXdPICnvlY37m67OqESVcQsZhFcGOcCrn_1F-U"
+)
+PROFIT_API_KEY = "dev-XvIu8S83BJ2N1C9ZCsoLAAav5jan7kj5"
+LOGGING_LEVEL = "INFO"
 
+
+class DevelopmentMode(str, Enum):
+    PRODUCTION = "PRODUCTION"
+    DEVELOPMENT = "DEVELOPMENT"
+
+
+DEVELOPMENT_MODE = DevelopmentMode.PRODUCTION
 RETRY_LIMIT = 5
 RETRY_WAIT = 3
 BACKOFF_DELAY = 0.001

monitor_utils/database.py

@@ -7,23 +7,20 @@ from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
 from pymongo import UpdateOne
 from pymongo.errors import PyMongoError
 
-from monitor_utils.config import get_settings
 from monitor_utils.models import Product
-from monitor_utils.constants import STATUS_PENDING
+from monitor_utils.constants import STATUS_PENDING, MONGO_DATABASE_NAME, MONGO_URL
 
 LOGGER = logging.getLogger(__name__)
 
 
 def get_collection(collection_name: str) -> AsyncIOMotorCollection:
-    settings = get_settings()
-    return get_mongo_client()[settings.mongo_database_name][collection_name]
+    return get_mongo_client()[MONGO_DATABASE_NAME][collection_name]
 
 
 @cache
 def get_mongo_client() -> AsyncIOMotorClient:
-    settings = get_settings()
     return AsyncIOMotorClient(
-        settings.mongo_url,
+        MONGO_URL,
         maxPoolSize=200,
         minPoolSize=10,
         retryWrites=True,

monitor_utils/runner.py

@@ -11,8 +11,9 @@ from monitor_utils.constants import (
     EXPECTED_TASK_ERRORS,
     FLAG_WEBHOOK_MAP,
     BUCKET_WEBHOOK_MAP,
+    LOGGING_LEVEL,
 )
-from monitor_utils.config import get_settings, get_config
+from monitor_utils.config import get_config, is_production
 from monitor_utils.models import (
     Product,
     ProductMetadata,
@@ -28,21 +29,23 @@ LOGGER = logging.getLogger(__name__)
 
 def configure_logging():
     logging.basicConfig(
-        level=get_settings().logging_level,
+        level=LOGGING_LEVEL,
         format="%(asctime)s - %(levelname)s - %(module)s - %(funcName)s - %(message)s",
     )
 
 
-def setup_config(monitor_name: str) -> tuple[list[str], dict, AsyncIOMotorCollection]:
-    is_production = get_settings().development_mode == "PRODUCTION"
+def setup_config(
+    monitor_name: str, override_production=False
+) -> tuple[list[str], dict, AsyncIOMotorCollection]:
+    use_production = is_production(override_production)
     config, _ = get_config(
         monitor_name=monitor_name,
-        use_production_path=is_production,
+        use_production_path=use_production,
     )
 
     category_names = config["categories"]
     collection = get_collection(
-        monitor_name if is_production else f"{monitor_name}_testing"
+        monitor_name if use_production else f"{monitor_name}_testing"
     )
     return category_names, config["channels"], collection