nikdoof/leggen
1
1
Fork 0
mirror of https://github.com/elisiariocouto/leggen.git synced 2025-12-13 14:52:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(config): Add Pydantic validation and fix telegram config field mappings.

Browse Source 
* Add Pydantic models for configuration validation in leggen/models/config.py
* Fix telegram config field aliases (api-key -> token, chat-id -> chat_id)
* Update config.py to use Pydantic validation with proper error handling
* Fix TOML serialization by excluding None values with exclude_none=True
* Update notification service to use correct telegram field names
* Enhance notification service with actual Discord/Telegram implementations
* Fix all failing configuration tests to work with Pydantic validation
* Add pydantic dependency to pyproject.toml

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Elisiário Couto
2025-09-14 20:28:18 +01:00
committed by Elisiário Couto
parent 990d0295b3
commit 2c6e099596
6 changed files with 236 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
leggen/models/config.py Normal file
View File

@@ -0,0 +1,66 @@
from typing import Optional, List
from pydantic import BaseModel, Field
class GoCardlessConfig(BaseModel):
key: str = Field(..., description="GoCardless API key")
secret: str = Field(..., description="GoCardless API secret")
url: str = Field(
default="https://bankaccountdata.gocardless.com/api/v2",
description="GoCardless API URL",
)
class DatabaseConfig(BaseModel):
sqlite: bool = Field(default=True, description="Enable SQLite database")
class DiscordNotificationConfig(BaseModel):
webhook: str = Field(..., description="Discord webhook URL")
enabled: bool = Field(default=True, description="Enable Discord notifications")
class TelegramNotificationConfig(BaseModel):
token: str = Field(..., alias="api-key", description="Telegram bot token")
chat_id: int = Field(..., alias="chat-id", description="Telegram chat ID")
enabled: bool = Field(default=True, description="Enable Telegram notifications")
class NotificationConfig(BaseModel):
discord: Optional[DiscordNotificationConfig] = None
telegram: Optional[TelegramNotificationConfig] = None
class FilterConfig(BaseModel):
case_insensitive: Optional[List[str]] = Field(
default_factory=list, alias="case-insensitive"
)
case_sensitive: Optional[List[str]] = Field(
default_factory=list, alias="case-sensitive"
)
class SyncScheduleConfig(BaseModel):
enabled: bool = Field(default=True, description="Enable sync scheduling")
hour: int = Field(default=3, ge=0, le=23, description="Hour to run sync (0-23)")
minute: int = Field(default=0, ge=0, le=59, description="Minute to run sync (0-59)")
cron: Optional[str] = Field(
default=None, description="Custom cron expression (overrides hour/minute)"
)
class SchedulerConfig(BaseModel):
sync: SyncScheduleConfig = Field(default_factory=SyncScheduleConfig)
class Config(BaseModel):
gocardless: GoCardlessConfig
database: DatabaseConfig = Field(default_factory=DatabaseConfig)
notifications: Optional[NotificationConfig] = None
filters: Optional[FilterConfig] = None
scheduler: SchedulerConfig = Field(default_factory=SchedulerConfig)
class Config:
allow_population_by_field_name = (
True # Allow both 'case_insensitive' and 'case-insensitive'
)

108
leggen/services/notification_service.py
View File

@@ -109,26 +109,68 @@ class NotificationService:
"""Check if Telegram notifications are enabled"""
telegram_config = self.notifications_config.get("telegram", {})
return bool(
telegram_config.get("api-key")
and telegram_config.get("chat-id")
telegram_config.get("token")
and telegram_config.get("chat_id")
and telegram_config.get("enabled", True)
)
async def _send_discord_notifications(
self, transactions: List[Dict[str, Any]]
) -> None:
"""Send Discord notifications - placeholder implementation"""
# Would import and use leggen.notifications.discord
logger.info(f"Sending {len(transactions)} transaction notifications to Discord")
"""Send Discord notifications for transactions"""
try:
from leggen.notifications.discord import send_transactions_message
import click
# Create a mock context with the webhook
ctx = click.Context(click.Command("notifications"))
ctx.obj = {
"notifications": {
"discord": {
"webhook": self.notifications_config.get("discord", {}).get(
"webhook"
)
}
}
}
# Send transaction notifications using the actual implementation
send_transactions_message(ctx, transactions)
logger.info(
f"Sent {len(transactions)} transaction notifications to Discord"
)
except Exception as e:
logger.error(f"Failed to send Discord transaction notifications: {e}")
raise
async def _send_telegram_notifications(
self, transactions: List[Dict[str, Any]]
) -> None:
"""Send Telegram notifications - placeholder implementation"""
# Would import and use leggen.notifications.telegram
"""Send Telegram notifications for transactions"""
try:
from leggen.notifications.telegram import send_transaction_message
import click
# Create a mock context with the telegram config
ctx = click.Context(click.Command("notifications"))
telegram_config = self.notifications_config.get("telegram", {})
ctx.obj = {
"notifications": {
"telegram": {
"api-key": telegram_config.get("token"),
"chat-id": telegram_config.get("chat_id"),
}
}
}
# Send transaction notifications using the actual implementation
send_transaction_message(ctx, transactions)
logger.info(
f"Sending {len(transactions)} transaction notifications to Telegram"
f"Sent {len(transactions)} transaction notifications to Telegram"
)
except Exception as e:
logger.error(f"Failed to send Telegram transaction notifications: {e}")
raise
async def _send_discord_test(self, message: str) -> None:
"""Send Discord test notification"""
@@ -173,8 +215,8 @@ class NotificationService:
ctx.obj = {
"notifications": {
"telegram": {
"api-key": telegram_config.get("api-key"),
"chat-id": telegram_config.get("chat-id"),
"api-key": telegram_config.get("token"),
"chat-id": telegram_config.get("chat_id"),
}
}
}
@@ -194,8 +236,50 @@ class NotificationService:
async def _send_discord_expiry(self, notification_data: Dict[str, Any]) -> None:
"""Send Discord expiry notification"""
logger.info(f"Sending Discord expiry notification: {notification_data}")
try:
from leggen.notifications.discord import send_expire_notification
import click
# Create a mock context with the webhook
ctx = click.Context(click.Command("expiry"))
ctx.obj = {
"notifications": {
"discord": {
"webhook": self.notifications_config.get("discord", {}).get(
"webhook"
)
}
}
}
# Send expiry notification using the actual implementation
send_expire_notification(ctx, notification_data)
logger.info(f"Sent Discord expiry notification: {notification_data}")
except Exception as e:
logger.error(f"Failed to send Discord expiry notification: {e}")
raise
async def _send_telegram_expiry(self, notification_data: Dict[str, Any]) -> None:
"""Send Telegram expiry notification"""
logger.info(f"Sending Telegram expiry notification: {notification_data}")
try:
from leggen.notifications.telegram import send_expire_notification
import click
# Create a mock context with the telegram config
ctx = click.Context(click.Command("expiry"))
telegram_config = self.notifications_config.get("telegram", {})
ctx.obj = {
"notifications": {
"telegram": {
"api-key": telegram_config.get("token"),
"chat-id": telegram_config.get("chat_id"),
}
}
}
# Send expiry notification using the actual implementation
send_expire_notification(ctx, notification_data)
logger.info(f"Sent Telegram expiry notification: {notification_data}")
except Exception as e:
logger.error(f"Failed to send Telegram expiry notification: {e}")
raise

40
leggen/utils/config.py
View File

@@ -7,14 +7,17 @@ from typing import Dict, Any, Optional
import click
from loguru import logger
from pydantic import ValidationError
from leggen.utils.text import error
from leggen.utils.paths import path_manager
from leggen.models.config import Config as ConfigModel
class Config:
_instance = None
_config = None
_config_model = None
_config_path = None
def __new__(cls):
@@ -35,8 +38,18 @@ class Config:
try:
with open(config_path, "rb") as f:
self._config = tomllib.load(f)
raw_config = tomllib.load(f)
logger.info(f"Configuration loaded from {config_path}")
# Validate configuration using Pydantic
try:
self._config_model = ConfigModel(**raw_config)
self._config = self._config_model.dict(by_alias=True, exclude_none=True)
logger.info("Configuration validation successful")
except ValidationError as e:
logger.error(f"Configuration validation failed: {e}")
raise ValueError(f"Invalid configuration: {e}") from e
except FileNotFoundError:
logger.error(f"Configuration file not found: {config_path}")
raise
@@ -65,15 +78,24 @@ class Config:
if config_data is None:
raise ValueError("No config data to save")
# Validate the configuration before saving
try:
validated_model = ConfigModel(**config_data)
validated_config = validated_model.dict(by_alias=True, exclude_none=True)
except ValidationError as e:
logger.error(f"Configuration validation failed before save: {e}")
raise ValueError(f"Invalid configuration: {e}") from e
# Ensure directory exists
Path(config_path).parent.mkdir(parents=True, exist_ok=True)
try:
with open(config_path, "wb") as f:
tomli_w.dump(config_data, f)
tomli_w.dump(validated_config, f)
# Update in-memory config
self._config = config_data
self._config = validated_config
self._config_model = validated_model
self._config_path = config_path
logger.info(f"Configuration saved to {config_path}")
except Exception as e:
@@ -146,8 +168,16 @@ class Config:
def load_config(ctx: click.Context, _, filename):
try:
with click.open_file(str(filename), "rb") as f:
# TODO: Implement configuration file validation (use pydantic?)
ctx.obj = tomllib.load(f)
raw_config = tomllib.load(f)
# Validate configuration using Pydantic
try:
validated_model = ConfigModel(**raw_config)
ctx.obj = validated_model.dict(by_alias=True, exclude_none=True)
except ValidationError as e:
error(f"Configuration validation failed: {e}")
sys.exit(1)
except FileNotFoundError:
error(
"Configuration file not found. Provide a valid configuration file path with leggen --config <path> or LEGGEN_CONFIG=<path> environment variable."

1
pyproject.toml
View File

@@ -34,6 +34,7 @@ dependencies = [
"apscheduler>=3.10.0,<4",
"tomli-w>=1.0.0,<2",
"httpx>=0.28.1",
"pydantic>=2.0.0,<3",
]
[project.urls]

41
tests/unit/test_config.py
View File

@@ -37,10 +37,14 @@ class TestConfig:
# Reset singleton state for testing
config._config = None
config._config_path = None
config._config_model = None
result = config.load_config(str(config_file))
assert result == config_data
# Result should contain validated config data
assert result["gocardless"]["key"] == "test-key"
assert result["gocardless"]["secret"] == "test-secret"
assert result["database"]["sqlite"] is True
assert config.gocardless_config["key"] == "test-key"
assert config.database_config["sqlite"] is True
@@ -54,11 +58,19 @@ class TestConfig:
def test_save_config_success(self, temp_config_dir):
"""Test successful configuration saving."""
config_data = {"gocardless": {"key": "new-key", "secret": "new-secret"}}
config_data = {
"gocardless": {
"key": "new-key",
"secret": "new-secret",
"url": "https://bankaccountdata.gocardless.com/api/v2",
},
"database": {"sqlite": True},
}
config_file = temp_config_dir / "new_config.toml"
config = Config()
config._config = None
config._config_model = None
config.save_config(config_data, str(config_file))
@@ -70,12 +82,18 @@ class TestConfig:
with open(config_file, "rb") as f:
saved_data = tomllib.load(f)
assert saved_data == config_data
assert saved_data["gocardless"]["key"] == "new-key"
assert saved_data["gocardless"]["secret"] == "new-secret"
assert saved_data["database"]["sqlite"] is True
def test_update_config_success(self, temp_config_dir):
"""Test updating configuration values."""
initial_config = {
"gocardless": {"key": "old-key"},
"gocardless": {
"key": "old-key",
"secret": "old-secret",
"url": "https://bankaccountdata.gocardless.com/api/v2",
},
"database": {"sqlite": True},
}
@@ -87,6 +105,7 @@ class TestConfig:
config = Config()
config._config = None
config._config_model = None
config.load_config(str(config_file))
config.update_config("gocardless", "key", "new-key")
@@ -102,7 +121,14 @@ class TestConfig:
def test_update_section_success(self, temp_config_dir):
"""Test updating entire configuration section."""
initial_config = {"database": {"sqlite": True}}
initial_config = {
"gocardless": {
"key": "test-key",
"secret": "test-secret",
"url": "https://bankaccountdata.gocardless.com/api/v2",
},
"database": {"sqlite": True},
}
config_file = temp_config_dir / "config.toml"
with open(config_file, "wb") as f:
@@ -112,12 +138,13 @@ class TestConfig:
config = Config()
config._config = None
config._config_model = None
config.load_config(str(config_file))
new_db_config = {"sqlite": False, "path": "./custom.db"}
new_db_config = {"sqlite": False}
config.update_section("database", new_db_config)
assert config.database_config == new_db_config
assert config.database_config["sqlite"] is False
def test_scheduler_config_defaults(self):
"""Test scheduler configuration with defaults."""

2
uv.lock generated
View File

@@ -229,6 +229,7 @@ dependencies = [
{ name = "fastapi" },
{ name = "httpx" },
{ name = "loguru" },
{ name = "pydantic" },
{ name = "requests" },
{ name = "tabulate" },
{ name = "tomli-w" },
@@ -257,6 +258,7 @@ requires-dist = [
{ name = "fastapi", specifier = ">=0.104.0,<1" },
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "loguru", specifier = ">=0.7.2,<0.8" },
{ name = "pydantic", specifier = ">=2.0.0,<3" },
{ name = "requests", specifier = ">=2.31.0,<3" },
{ name = "tabulate", specifier = ">=0.9.0,<0.10" },
{ name = "tomli-w", specifier = ">=1.0.0,<2" },