alex/config_manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop'

Browse Source
This commit is contained in:
zosimovaa
2025-10-31 18:54:18 +03:00
parent 4eb9327628 f491c65455
commit 5faea8f69f
8 changed files with 263 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1,3 +1,4 @@
__pycache__
venv/
.vscode/
log*.log

2
pyproject.toml
View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "config_manager"
version = "1.0.4"
version = "1.1.0"
description = "Config manager for building applications"
authors = [
{ name = "Aleksei Zosimov", email = "lesha.spb@gmail.com" }

2
src/basic_application/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from .config_manager import ConfigManager
from .log_manager import LogManager

150
src/basic_application/config_manager.py Normal file
View File

@@ -0,0 +1,150 @@
import logging
import logging.config
import asyncio
import json
import yaml
import os
from typing import Any, Optional
<<<<<<<< HEAD:src/config_manager.py
logger = logging.getLogger(__name__)
========
from .log_manager import LogManager
>>>>>>>> develop:src/basic_application/config_manager.py
class ConfigManager:
DEFAULT_UPDATE_INTERVAL = 5.0
DEFAULT_WORK_INTERVAL = 2.0
def __init__(self, path: str, log_manager: Optional[LogManager] = None):
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()
self._task: Optional[asyncio.Task] = None
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._log_manager = log_manager or LogManager()
self.logger = logging.getLogger(__name__)
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)
<<<<<<<< HEAD:src/config_manager.py
def _parse_config(self, data) -> Any:
========
def _parse_config(self, data) -> Any:
>>>>>>>> develop:src/basic_application/config_manager.py
extension = os.path.splitext(self.path)[1].lower()
if extension 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
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)
self.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)
self.logger.info(f"Work interval set to {self.work_interval} seconds")
else:
self.work_interval = self.DEFAULT_WORK_INTERVAL
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._log_manager.apply_config(new_config)
self._update_intervals_from_config()
except Exception as e:
self.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 _run(self) -> None:
"""Внутренняя корутина, запускающая все циклы"""
self._halt.clear()
self.logger.info("ConfigManager started")
try:
await asyncio.gather(
self._worker_loop(),
self._periodic_update_loop()
)
except asyncio.CancelledError:
self.logger.info("ConfigManager tasks cancelled")
finally:
self.logger.info("ConfigManager stopped")
def start(self) -> None:
"""Запускает менеджер конфигурации в текущем event loop"""
if self._task is not None and not self._task.done():
self.logger.warning("ConfigManager is already running")
return
try:
self._loop = asyncio.get_running_loop()
except RuntimeError:
self.logger.error("start() must be called from within an async context")
raise
self._task = self._loop.create_task(self._run())
self.logger.info("ConfigManager task created")
async def stop(self) -> None:
"""Останавливает менеджер конфигурации и ожидает завершения"""
if self._task is None:
self.logger.warning("ConfigManager is not running")
return
self.logger.info("ConfigManager stopping...")
self._halt.set()
try:
await self._task
except asyncio.CancelledError:
pass
self._task = None
<<<<<<<< HEAD:src/config_manager.py
logger.info("ConfigManager stopped successfully")
========
self.logger.info("ConfigManager stopped successfully")
>>>>>>>> develop:src/basic_application/config_manager.py

40
src/basic_application/log_manager.py Normal file
View File

@@ -0,0 +1,40 @@
import logging
from typing import Optional
class LogManager:
"""
Управляет конфигурацией логирования приложения.
Применяет конфигурацию из словаря с обработкой ошибок.
"""
def __init__(self):
self.logger = logging.getLogger(__name__)
self._last_valid_config: Optional[dict] = None
def apply_config(self, config: dict) -> None:
"""
Применяет конфигурацию логирования из словаря.
При ошибке восстанавливает последний валидный конфиг.
Args:
config: Словарь с настройками логирования (из файла конфига)
"""
logging_config = config.get("log")
if not logging_config:
return
try:
logging.config.dictConfig(logging_config)
self._last_valid_config = logging_config
self.logger.info("Logging configuration applied")
except Exception as e:
self.logger.error(f"Error applying logging config: {e}")
# Если был предыдущий валидный конфиг, восстанавливаем его
if self._last_valid_config:
try:
logging.config.dictConfig(self._last_valid_config)
self.logger.info("Previous logging configuration restored")
except Exception as restore_error:
self.logger.error(f"Error restoring previous config: {restore_error}")

6
src/config.yaml
View File

@@ -8,7 +8,7 @@ log:
formatters:
standard:
format: '%(asctime)s %(name)30s [%(levelname)8s]: %(message)s'
format: '%(asctime)s %(module)15s [%(levelname)8s]: %(message)s'
telegram:
format: '%(message)s'
@@ -40,12 +40,12 @@ log:
loggers:
'':
handlers: [console, file]
level: ERROR
level: INFO
propagate: False
__main__:
handlers: [console, file]
level: WARNING
level: DEBUG
propagate: False
config_manager:

58
src/config_manager.py
View File

@@ -1,19 +1,23 @@
import logging
import logging.config
import asyncio
import json
import yaml
import logging
import logging.config
from typing import Any, Optional
import os
from typing import Any, Optional
<<<<<<<< HEAD:src/config_manager.py
logger = logging.getLogger(__name__)
========
from .log_manager import LogManager
>>>>>>>> develop:src/basic_application/config_manager.py
class ConfigManager:
DEFAULT_UPDATE_INTERVAL = 5.0
DEFAULT_WORK_INTERVAL = 2.0
def __init__(self, path: str):
def __init__(self, path: str, log_manager: Optional[LogManager] = None):
self.path = path
self.config: Any = None
self._last_hash = None
@@ -22,6 +26,9 @@ class ConfigManager:
self._halt = asyncio.Event()
self._task: Optional[asyncio.Task] = None
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._log_manager = log_manager or LogManager()
self.logger = logging.getLogger(__name__)
def _read_file_sync(self) -> str:
with open(self.path, "r", encoding="utf-8") as f:
@@ -30,7 +37,11 @@ class ConfigManager:
async def _read_file_async(self) -> str:
return await asyncio.to_thread(self._read_file_sync)
<<<<<<<< HEAD:src/config_manager.py
def _parse_config(self, data) -> Any:
========
def _parse_config(self, data) -> Any:
>>>>>>>> develop:src/basic_application/config_manager.py
extension = os.path.splitext(self.path)[1].lower()
if extension in (".yaml", ".yml"):
return yaml.safe_load(data)
@@ -45,25 +56,16 @@ class ConfigManager:
if isinstance(upd, (int, float)) and upd > 0:
self.update_interval = float(upd)
logger.info(f"Update interval set to {self.update_interval} seconds")
self.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")
self.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()
@@ -73,12 +75,11 @@ class ConfigManager:
self.config = new_config
self._last_hash = current_hash
self._apply_logging_config(new_config)
self._log_manager.apply_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}")
self.logger.error(f"Error reading/parsing config file: {e}")
def execute(self) -> None:
"""
@@ -101,46 +102,49 @@ class ConfigManager:
async def _run(self) -> None:
"""Внутренняя корутина, запускающая все циклы"""
self._halt.clear()
logger.info("ConfigManager started")
self.logger.info("ConfigManager started")
try:
await asyncio.gather(
self._worker_loop(),
self._periodic_update_loop()
)
except asyncio.CancelledError:
logger.info("ConfigManager tasks cancelled")
self.logger.info("ConfigManager tasks cancelled")
finally:
logger.info("ConfigManager stopped")
self.logger.info("ConfigManager stopped")
def start(self) -> None:
"""Запускает менеджер конфигурации в текущем event loop"""
if self._task is not None and not self._task.done():
logger.warning("ConfigManager is already running")
self.logger.warning("ConfigManager is already running")
return
try:
self._loop = asyncio.get_running_loop()
except RuntimeError:
logger.error("start() must be called from within an async context")
self.logger.error("start() must be called from within an async context")
raise
self._task = self._loop.create_task(self._run())
logger.info("ConfigManager task created")
self.logger.info("ConfigManager task created")
async def stop(self) -> None:
"""Останавливает менеджер конфигурации и ожидает завершения"""
if self._task is None:
logger.warning("ConfigManager is not running")
self.logger.warning("ConfigManager is not running")
return
logger.info("ConfigManager stopping...")
self.logger.info("ConfigManager stopping...")
self._halt.set()
# Ждём корректного завершения задачи
try:
await self._task
except asyncio.CancelledError:
pass
self._task = None
<<<<<<<< HEAD:src/config_manager.py
logger.info("ConfigManager stopped successfully")
========
self.logger.info("ConfigManager stopped successfully")
>>>>>>>> develop:src/basic_application/config_manager.py

35
src/test.py Normal file
View File

@@ -0,0 +1,35 @@
from basic_application import ConfigManager
import logging
import asyncio
from typing import Optional
import os
os.chdir(os.path.dirname(__file__))
logger = logging.getLogger()
class MyApp(ConfigManager):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.iter = 0
def execute(self) -> None:
logger.info(f"current iteration {self.iter}")
self.iter += 1
async def main():
app = MyApp("config.yaml")
app.start()
logger.info("App started")
await asyncio.sleep(20)
await app.stop()
if __name__ == "__main__":
asyncio.run(main())