Refactor blocklist schedule management into service

This commit is contained in:
2026-04-14 15:25:36 +02:00
parent 58bb769a35
commit b70dc6fa7a
6 changed files with 161 additions and 75 deletions

View File

@@ -208,7 +208,11 @@ Lazy imports mask circular dependencies at import time but surface as confusing
--- ---
### TASK-07 — Move schedule management out of `routers/blocklist.py` into a service 🟠 ### TASK-07 — Move schedule management out of `routers/blocklist.py` into a service
**Status:** Completed ✅
**Summary:** Router schedule endpoints now delegate schedule persistence and APScheduler job management to `blocklist_service`, removing the router's direct dependency on `app.tasks.blocklist_import`.
**Where:** **Where:**
`backend/app/routers/blocklist.py` — line 51: `backend/app/routers/blocklist.py` — line 51:
@@ -418,4 +422,4 @@ Additionally note in section 6 (Authentication): the `_session_cache` in `depend
- Cross-reference TASK-05 once it is complete (the task should delegate to a service). - Cross-reference TASK-05 once it is complete (the task should delegate to a service).
**Why:** **Why:**
The architecture document is a contract between contributors. Undocumented modules lead to inconsistent patterns and duplicate implementations when new developers add features that the undocumented module already handles. The architecture document is a contract between contributors. Undocumented modules lead to inconsistent patterns and duplicate implementations when new developers add features that the undocumented module already handlzttttttttt nbbbbbbbbbbes.

View File

@@ -28,12 +28,12 @@ import aiosqlite
from fastapi import APIRouter, Depends, HTTPException, Query, status from fastapi import APIRouter, Depends, HTTPException, Query, status
from app.dependencies import ( from app.dependencies import (
AppDep,
AuthDep, AuthDep,
Fail2BanSocketDep, Fail2BanSocketDep,
GeoBatchLookupDep, GeoBatchLookupDep,
HttpSessionDep, HttpSessionDep,
SchedulerDep, SchedulerDep,
SettingsDep,
get_db, get_db,
) )
from app.models.blocklist import ( from app.models.blocklist import (
@@ -48,7 +48,6 @@ from app.models.blocklist import (
ScheduleInfo, ScheduleInfo,
) )
from app.services import blocklist_service, geo_service, jail_service from app.services import blocklist_service, geo_service, jail_service
from app.tasks import blocklist_import as blocklist_import_task
router: APIRouter = APIRouter(prefix="/api/blocklists", tags=["Blocklists"]) router: APIRouter = APIRouter(prefix="/api/blocklists", tags=["Blocklists"])
@@ -162,7 +161,6 @@ async def get_schedule(
The ``next_run_at`` field is read from APScheduler if the job is active. The ``next_run_at`` field is read from APScheduler if the job is active.
Args: Args:
request: Incoming request (used to query the scheduler).
db: Application database connection (injected). db: Application database connection (injected).
_auth: Validated session — enforces authentication. _auth: Validated session — enforces authentication.
@@ -170,12 +168,7 @@ async def get_schedule(
:class:`~app.models.blocklist.ScheduleInfo` with config and run :class:`~app.models.blocklist.ScheduleInfo` with config and run
times. times.
""" """
job = scheduler.get_job(blocklist_import_task.JOB_ID) return await blocklist_service.get_schedule_info_with_runtime(db, scheduler)
next_run_at: str | None = None
if job is not None and job.next_run_time is not None:
next_run_at = job.next_run_time.isoformat()
return await blocklist_service.get_schedule_info(db, next_run_at)
@router.put( @router.put(
@@ -188,29 +181,29 @@ async def update_schedule(
db: DbDep, db: DbDep,
_auth: AuthDep, _auth: AuthDep,
scheduler: SchedulerDep, scheduler: SchedulerDep,
app: AppDep, http_session: HttpSessionDep,
settings: SettingsDep,
) -> ScheduleInfo: ) -> ScheduleInfo:
"""Persist a new schedule configuration and reschedule the import job. """Persist a new schedule configuration and reschedule the import job.
Args: Args:
payload: New :class:`~app.models.blocklist.ScheduleConfig`. payload: New :class:`~app.models.blocklist.ScheduleConfig`.
app: FastAPI application instance used to reschedule the import job.
db: Application database connection (injected). db: Application database connection (injected).
_auth: Validated session — enforces authentication. _auth: Validated session — enforces authentication.
scheduler: Shared APScheduler instance (injected).
http_session: Shared HTTP session used by the scheduler job.
settings: Current application settings used by the scheduler job.
Returns: Returns:
Updated :class:`~app.models.blocklist.ScheduleInfo`. Updated :class:`~app.models.blocklist.ScheduleInfo`.
""" """
await blocklist_service.set_schedule(db, payload) return await blocklist_service.update_schedule(
# Reschedule the background job immediately. db,
blocklist_import_task.reschedule(app) scheduler,
http_session,
job = scheduler.get_job(blocklist_import_task.JOB_ID) settings,
next_run_at: str | None = None payload,
if job is not None and job.next_run_time is not None: )
next_run_at = job.next_run_time.isoformat()
return await blocklist_service.get_schedule_info(db, next_run_at)
@router.get( @router.get(

View File

@@ -30,6 +30,7 @@ from app.models.blocklist import (
ImportSourceResult, ImportSourceResult,
PreviewResponse, PreviewResponse,
ScheduleConfig, ScheduleConfig,
ScheduleFrequency,
ScheduleInfo, ScheduleInfo,
) )
from app.repositories import blocklist_repo, import_log_repo, settings_repo from app.repositories import blocklist_repo, import_log_repo, settings_repo
@@ -40,7 +41,9 @@ if TYPE_CHECKING:
import aiohttp import aiohttp
import aiosqlite import aiosqlite
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from app.config import Settings
from app.models.geo import GeoBatchLookup from app.models.geo import GeoBatchLookup
log: structlog.stdlib.BoundLogger = structlog.get_logger() log: structlog.stdlib.BoundLogger = structlog.get_logger()
@@ -505,6 +508,66 @@ async def import_all(
_DEFAULT_SCHEDULE = ScheduleConfig() _DEFAULT_SCHEDULE = ScheduleConfig()
#: Stable APScheduler job id for the blocklist import job.
JOB_ID: str = "blocklist_import"
def _get_job_next_run_at(scheduler: AsyncIOScheduler) -> str | None:
"""Return the next scheduled run time as an ISO 8601 string."""
job = scheduler.get_job(JOB_ID)
if job is None or job.next_run_time is None:
return None
return job.next_run_time.isoformat()
def schedule_blocklist_job(
scheduler: AsyncIOScheduler,
settings: Settings,
http_session: aiohttp.ClientSession,
config: ScheduleConfig,
) -> None:
"""Register or replace the scheduled blocklist import job."""
from app.tasks import blocklist_import as blocklist_import_task
if scheduler.get_job(JOB_ID):
scheduler.remove_job(JOB_ID)
kwargs: dict[str, object] = {
"settings": settings,
"http_session": http_session,
}
if config.frequency == ScheduleFrequency.hourly:
trigger_type = "interval"
trigger_kwargs = {"hours": config.interval_hours}
elif config.frequency == ScheduleFrequency.weekly:
trigger_type = "cron"
trigger_kwargs = {
"day_of_week": config.day_of_week,
"hour": config.hour,
"minute": config.minute,
}
else:
trigger_type = "cron"
trigger_kwargs = {
"hour": config.hour,
"minute": config.minute,
}
scheduler.add_job(
blocklist_import_task._run_import_with_resources,
trigger=trigger_type,
id=JOB_ID,
kwargs=kwargs,
**trigger_kwargs,
)
log.info(
"blocklist_import_scheduled",
frequency=config.frequency,
trigger=trigger_type,
trigger_kwargs=trigger_kwargs,
)
async def get_schedule(db: aiosqlite.Connection) -> ScheduleConfig: async def get_schedule(db: aiosqlite.Connection) -> ScheduleConfig:
"""Read the import schedule config from the settings table. """Read the import schedule config from the settings table.
@@ -576,6 +639,28 @@ async def get_schedule_info(
) )
async def get_schedule_info_with_runtime(
db: aiosqlite.Connection,
scheduler: AsyncIOScheduler,
) -> ScheduleInfo:
"""Return schedule info enriched with runtime scheduler metadata."""
next_run_at = _get_job_next_run_at(scheduler)
return await get_schedule_info(db, next_run_at)
async def update_schedule(
db: aiosqlite.Connection,
scheduler: AsyncIOScheduler,
http_session: aiohttp.ClientSession,
settings: Settings,
config: ScheduleConfig,
) -> ScheduleInfo:
"""Persist a new schedule config and re-register the scheduled job."""
await set_schedule(db, config)
schedule_blocklist_job(scheduler, settings, http_session, config)
return await get_schedule_info(db, _get_job_next_run_at(scheduler))
async def list_import_logs( async def list_import_logs(
db: aiosqlite.Connection, db: aiosqlite.Connection,
*, *,

View File

@@ -18,7 +18,6 @@ from typing import TYPE_CHECKING, Any
import structlog import structlog
from app.db import open_db from app.db import open_db
from app.models.blocklist import ScheduleFrequency
from app.services import blocklist_service, jail_service from app.services import blocklist_service, jail_service
from app.utils.runtime_state import get_effective_settings from app.utils.runtime_state import get_effective_settings
@@ -130,46 +129,11 @@ def _apply_schedule(app: FastAPI, config: Any) -> None:
app: FastAPI application instance. app: FastAPI application instance.
config: :class:`~app.models.blocklist.ScheduleConfig` to apply. config: :class:`~app.models.blocklist.ScheduleConfig` to apply.
""" """
scheduler = app.state.scheduler from app.services import blocklist_service
kwargs: dict[str, Any] = { blocklist_service.schedule_blocklist_job(
"settings": get_effective_settings(app), app.state.scheduler,
"http_session": app.state.http_session, get_effective_settings(app),
} app.state.http_session,
trigger_type: str config,
trigger_kwargs: dict[str, Any]
if config.frequency == ScheduleFrequency.hourly:
trigger_type = "interval"
trigger_kwargs = {"hours": config.interval_hours}
elif config.frequency == ScheduleFrequency.weekly:
trigger_type = "cron"
trigger_kwargs = {
"day_of_week": config.day_of_week,
"hour": config.hour,
"minute": config.minute,
}
else: # daily (default)
trigger_type = "cron"
trigger_kwargs = {
"hour": config.hour,
"minute": config.minute,
}
# Remove existing job if it exists, then add new one.
if scheduler.get_job(JOB_ID):
scheduler.remove_job(JOB_ID)
scheduler.add_job(
_run_import_with_resources,
trigger=trigger_type,
id=JOB_ID,
kwargs=kwargs,
**trigger_kwargs,
)
log.info(
"blocklist_import_scheduled",
frequency=config.frequency,
trigger=trigger_type,
trigger_kwargs=trigger_kwargs,
) )

View File

@@ -361,7 +361,7 @@ class TestGetSchedule:
async def test_schedule_returns_200(self, bl_client: AsyncClient) -> None: async def test_schedule_returns_200(self, bl_client: AsyncClient) -> None:
"""GET /api/blocklists/schedule returns 200.""" """GET /api/blocklists/schedule returns 200."""
with patch( with patch(
"app.routers.blocklist.blocklist_service.get_schedule_info", "app.routers.blocklist.blocklist_service.get_schedule_info_with_runtime",
new=AsyncMock(return_value=_make_schedule_info()), new=AsyncMock(return_value=_make_schedule_info()),
): ):
resp = await bl_client.get("/api/blocklists/schedule") resp = await bl_client.get("/api/blocklists/schedule")
@@ -370,7 +370,7 @@ class TestGetSchedule:
async def test_schedule_response_has_config(self, bl_client: AsyncClient) -> None: async def test_schedule_response_has_config(self, bl_client: AsyncClient) -> None:
"""Schedule response includes the config sub-object.""" """Schedule response includes the config sub-object."""
with patch( with patch(
"app.routers.blocklist.blocklist_service.get_schedule_info", "app.routers.blocklist.blocklist_service.get_schedule_info_with_runtime",
new=AsyncMock(return_value=_make_schedule_info()), new=AsyncMock(return_value=_make_schedule_info()),
): ):
resp = await bl_client.get("/api/blocklists/schedule") resp = await bl_client.get("/api/blocklists/schedule")
@@ -396,7 +396,7 @@ class TestGetSchedule:
last_run_errors=True, last_run_errors=True,
) )
with patch( with patch(
"app.routers.blocklist.blocklist_service.get_schedule_info", "app.routers.blocklist.blocklist_service.get_schedule_info_with_runtime",
new=AsyncMock(return_value=info_with_errors), new=AsyncMock(return_value=info_with_errors),
): ):
resp = await bl_client.get("/api/blocklists/schedule") resp = await bl_client.get("/api/blocklists/schedule")
@@ -425,13 +425,8 @@ class TestUpdateSchedule:
last_run_at=None, last_run_at=None,
) )
with patch( with patch(
"app.routers.blocklist.blocklist_service.set_schedule", "app.routers.blocklist.blocklist_service.update_schedule",
new=AsyncMock(),
), patch(
"app.routers.blocklist.blocklist_service.get_schedule_info",
new=AsyncMock(return_value=new_info), new=AsyncMock(return_value=new_info),
), patch(
"app.routers.blocklist.blocklist_import_task.reschedule",
): ):
resp = await bl_client.put( resp = await bl_client.put(
"/api/blocklists/schedule", "/api/blocklists/schedule",

View File

@@ -9,7 +9,11 @@ import aiosqlite
import pytest import pytest
from app.db import init_db from app.db import init_db
from app.models.blocklist import BlocklistSource, ScheduleConfig, ScheduleFrequency from app.models.blocklist import (
BlocklistSource,
ScheduleConfig,
ScheduleFrequency,
)
from app.services import blocklist_service from app.services import blocklist_service
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -346,6 +350,47 @@ class TestSchedule:
info = await blocklist_service.get_schedule_info(db, None) info = await blocklist_service.get_schedule_info(db, None)
assert info.last_run_errors is True assert info.last_run_errors is True
async def test_get_schedule_info_with_runtime_uses_scheduler_metadata(
self, db: aiosqlite.Connection
) -> None:
"""get_schedule_info_with_runtime derives next_run_at from the scheduler."""
next_run = MagicMock()
next_run.isoformat.return_value = "2099-01-01T00:00:00+00:00"
scheduler = MagicMock()
scheduler.get_job.return_value = MagicMock(next_run_time=next_run)
info = await blocklist_service.get_schedule_info_with_runtime(db, scheduler)
assert info.next_run_at == "2099-01-01T00:00:00+00:00"
async def test_update_schedule_persists_and_schedules_job(
self, db: aiosqlite.Connection
) -> None:
"""update_schedule must persist the config and schedule a job."""
settings = MagicMock(
fail2ban_socket="/var/run/fail2ban/fail2ban.sock",
database_path=":memory:",
)
http_session = MagicMock()
scheduler = MagicMock()
scheduler.get_job.return_value = None
config = ScheduleConfig(
frequency=ScheduleFrequency.daily,
hour=4,
minute=15,
)
info = await blocklist_service.update_schedule(
db,
scheduler,
http_session,
settings,
config,
)
assert info.config.frequency == ScheduleFrequency.daily
scheduler.add_job.assert_called_once()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Geo prewarm cache filtering # Geo prewarm cache filtering