add to git ignore

src/basic_application/basic_application.py

@@ -1,134 +0,0 @@
-import asyncio
-import json
-import yaml  # pip install pyyaml
-import logging
-import logging.config
-from typing import Any
-import os
-
-
-logger = logging.getLogger(__name__)
-
-
-class ConfigManager:
-    DEFAULT_UPDATE_INTERVAL = 5.0
-    DEFAULT_WORK_INTERVAL = 2.0
-
-    def __init__(self, path: str):
-        self.path = path
-        self.config: Any = None
-        self._last_hash = None
-        self.update_interval = self.DEFAULT_UPDATE_INTERVAL
-        self.work_interval = self.DEFAULT_WORK_INTERVAL
-        self._halt = asyncio.Event()
-
-    def _read_file_sync(self) -> str:
-        with open(self.path, "r", encoding="utf-8") as f:
-            return f.read()
-
-    async def _read_file_async(self) -> str:
-        return await asyncio.to_thread(self._read_file_sync)
-
-    def _parse_config(self, data: str) -> Any:
-        ext = os.path.splitext(self.path)[1].lower()
-        if ext in (".yaml", ".yml"):
-            return yaml.safe_load(data)
-        else:
-            return json.loads(data)
-
-    def _update_intervals_from_config(self) -> None:
-        if not self.config:
-            return
-        # Берём интервалы из секции config обновления, с контролем типа и значений
-        upd = self.config.get("update_interval")
-        wrk = self.config.get("work_interval")
-
-        if isinstance(upd, (int, float)) and upd > 0:
-            self.update_interval = float(upd)
-            logger.info(f"Update interval set to {self.update_interval} seconds")
-        else:
-            self.update_interval = self.DEFAULT_UPDATE_INTERVAL
-
-        if isinstance(wrk, (int, float)) and wrk > 0:
-            self.work_interval = float(wrk)
-            logger.info(f"Work interval set to {self.work_interval} seconds")
-        else:
-            self.work_interval = self.DEFAULT_WORK_INTERVAL
-
-    def _apply_logging_config(self, config: dict) -> None:
-        try:
-            logging_config = config.get("logging")
-            if logging_config:
-                logging.config.dictConfig(logging_config)
-                logger.info("Logging configuration applied")
-        except Exception as e:
-            logger.error(f"Error applying logging config: {e}")
-
-    async def _update_config(self) -> None:
-        try:
-            data = await self._read_file_async()
-            current_hash = hash(data)
-            if current_hash != self._last_hash:
-                new_config = self._parse_config(data)
-                self.config = new_config
-                self._last_hash = current_hash
-
-                self._apply_logging_config(new_config)
-                self._update_intervals_from_config()
-
-                logger.info("Config updated: %s", self.config)
-        except Exception as e:
-            logger.error(f"Error reading/parsing config file: {e}")
-
-    def execute(self) -> None:
-        """
-        Метод для переопределения в подклассах.
-        Здесь может быть блокирующая работа.
-        Запускается в отдельном потоке.
-        """
-        pass
-
-    async def _worker_loop(self) -> None:
-        while not self._halt.is_set():
-            await asyncio.to_thread(self.execute)
-            await asyncio.sleep(self.work_interval)
-
-    async def _periodic_update_loop(self) -> None:
-        while not self._halt.is_set():
-            await self._update_config()
-            await asyncio.sleep(self.update_interval)
-
-    async def start(self) -> None:
-        self._halt.clear()
-        logger.info("ConfigManager started")
-        await asyncio.gather(
-            self._worker_loop(),
-            self._periodic_update_loop()
-        )
-
-    def stop(self) -> None:
-        self._halt.set()
-        logger.info("ConfigManager stopping...")
-
-
-
-
-# Пример наследования и переопределения execute
-class MyApp(ConfigManager):
-    def execute(self) -> None:
-        logger.info("Executing blocking work with config: %s", self.config)
-
-
-async def main():
-    app = MyApp("config.yaml")  # Можно config.json или config.yaml
-    task = asyncio.create_task(app.start())
-    await asyncio.sleep(20)
-    app.stop()
-    await task
-    logger.info("Work finished.")
-
-
-if __name__ == "__main__":
-    logging.basicConfig(level=logging.INFO,
-                        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
-    asyncio.run(main())