Refactor backend services and utilities

- Update service layer implementations
- Improve configuration handling utilities
- Update documentation tasks

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
2026-04-25 18:39:30 +02:00
parent 83452ffc23
commit 420ea18fd9
12 changed files with 52 additions and 83 deletions

View File

@@ -43,7 +43,11 @@ from app.models.ban import (
from app.repositories import fail2ban_db_repo
from app.repositories import history_archive_repo as default_history_archive_repo
from app.services.fail2ban_metadata_service import default_fail2ban_metadata_service
from app.utils.constants import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE
from app.utils.constants import (
DEFAULT_PAGE_SIZE,
FAIL2BAN_SOCKET_TIMEOUT,
MAX_PAGE_SIZE,
)
from app.utils.fail2ban_client import (
Fail2BanClient,
)
@@ -73,10 +77,6 @@ async def get_fail2ban_db_path(socket_path: str) -> str:
# Constants
# ---------------------------------------------------------------------------
_SOCKET_TIMEOUT: float = 5.0
async def ban_ip(socket_path: str, jail: str, ip: str) -> None:
"""Ban an IP address in the specified jail."""
@@ -85,7 +85,7 @@ async def ban_ip(socket_path: str, jail: str, ip: str) -> None:
except ValueError as exc:
raise ValueError(f"Invalid IP address: {ip!r}") from exc
client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT)
client = Fail2BanClient(socket_path=socket_path, timeout=FAIL2BAN_SOCKET_TIMEOUT)
try:
ok(await client.send(["set", jail, "banip", ip]))
@@ -102,7 +102,7 @@ async def unban_ip(socket_path: str, ip: str, jail: str | None = None) -> None:
except ValueError as exc:
raise ValueError(f"Invalid IP address: {ip!r}") from exc
client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT)
client = Fail2BanClient(socket_path=socket_path, timeout=FAIL2BAN_SOCKET_TIMEOUT)
if jail is None:
ok(await client.send(["unban", ip]))
@@ -254,7 +254,7 @@ async def get_active_bans(
cannot be reached.
"""
client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT)
client = Fail2BanClient(socket_path=socket_path, timeout=FAIL2BAN_SOCKET_TIMEOUT)
global_status = to_dict(ok(await client.send(["status"])))
jail_list_raw: str = str(global_status.get("Jail list", "") or "").strip()