81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
"""Configuration management for refactored scripts."""
|
|
|
|
import logging
|
|
from configparser import ConfigParser
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DatabaseConfig:
|
|
"""Database configuration loader with validation."""
|
|
|
|
def __init__(self, config_file: Path | str = None, section: str = "mysql"):
|
|
"""
|
|
Initialize database configuration.
|
|
|
|
Args:
|
|
config_file: Path to the configuration file. Defaults to env/config.ini
|
|
section: Configuration section name. Defaults to 'mysql'
|
|
"""
|
|
if config_file is None:
|
|
# Default to env/config.ini relative to project root
|
|
config_file = Path(__file__).resolve().parent.parent.parent.parent / "env" / "config.ini"
|
|
|
|
self.config_file = Path(config_file)
|
|
self.section = section
|
|
self._config = self._load_config()
|
|
|
|
def _load_config(self) -> dict[str, str]:
|
|
"""Load and validate configuration from file."""
|
|
if not self.config_file.exists():
|
|
raise FileNotFoundError(f"Configuration file not found: {self.config_file}")
|
|
|
|
parser = ConfigParser()
|
|
parser.read(self.config_file)
|
|
|
|
if not parser.has_section(self.section):
|
|
raise ValueError(f"Section '{self.section}' not found in {self.config_file}")
|
|
|
|
config = dict(parser.items(self.section))
|
|
logger.info(f"Configuration loaded from {self.config_file}, section [{self.section}]")
|
|
|
|
return config
|
|
|
|
@property
|
|
def host(self) -> str:
|
|
"""Database host."""
|
|
return self._config.get("host", "localhost")
|
|
|
|
@property
|
|
def port(self) -> int:
|
|
"""Database port."""
|
|
return int(self._config.get("port", "3306"))
|
|
|
|
@property
|
|
def database(self) -> str:
|
|
"""Database name."""
|
|
return self._config["database"]
|
|
|
|
@property
|
|
def user(self) -> str:
|
|
"""Database user."""
|
|
return self._config["user"]
|
|
|
|
@property
|
|
def password(self) -> str:
|
|
"""Database password."""
|
|
return self._config["password"]
|
|
|
|
def as_dict(self) -> dict[str, any]:
|
|
"""Return configuration as dictionary compatible with aiomysql."""
|
|
return {
|
|
"host": self.host,
|
|
"port": self.port,
|
|
"db": self.database,
|
|
"user": self.user,
|
|
"password": self.password,
|
|
"autocommit": True,
|
|
}
|