fix: seed admin user on startup and run sync as background task
- Seed/update admin user from ADMIN_USERNAME/ADMIN_PASSWORD settings on startup (admin_users table was empty, login was impossible) - POST /api/admin/sync now returns immediately and runs the equipment sync via BackgroundTasks (full sync took minutes and hit nginx 504) - run_sync accepts optional sync_log_id to reuse a pre-created log entry - Adapt admin router test to background sync behavior
This commit is contained in:
+23
-1
@@ -9,6 +9,9 @@ from app.config import get_settings
|
|||||||
from app.database import engine, init_db, async_session
|
from app.database import engine, init_db, async_session
|
||||||
from app.cache import cache
|
from app.cache import cache
|
||||||
from app.routers import equipment, health, admin, rental_requests, contact
|
from app.routers import equipment, health, admin, rental_requests, contact
|
||||||
|
from app.models.admin_user import AdminUser
|
||||||
|
from app.auth import get_password_hash
|
||||||
|
from sqlalchemy import select
|
||||||
from app.services.sync_service import SyncService
|
from app.services.sync_service import SyncService
|
||||||
from app.services.email_service import EmailService
|
from app.services.email_service import EmailService
|
||||||
from app.mcp_server import get_mcp_app, mcp_session_manager
|
from app.mcp_server import get_mcp_app, mcp_session_manager
|
||||||
@@ -36,10 +39,29 @@ async def run_email_retry() -> None:
|
|||||||
logger.info("Email retry complete: %d emails sent", sent)
|
logger.info("Email retry complete: %d emails sent", sent)
|
||||||
|
|
||||||
|
|
||||||
|
async def seed_admin_user() -> None:
|
||||||
|
"""Ensure the admin user from settings exists (create or update password)."""
|
||||||
|
async with async_session() as db:
|
||||||
|
result = await db.execute(
|
||||||
|
select(AdminUser).where(AdminUser.username == settings.admin_username)
|
||||||
|
)
|
||||||
|
user = result.scalar_one_or_none()
|
||||||
|
if not user:
|
||||||
|
db.add(AdminUser(
|
||||||
|
username=settings.admin_username,
|
||||||
|
password_hash=get_password_hash(settings.admin_password),
|
||||||
|
))
|
||||||
|
logger.info("Seeded admin user %s", settings.admin_username)
|
||||||
|
else:
|
||||||
|
user.password_hash = get_password_hash(settings.admin_password)
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(app: FastAPI):
|
async def lifespan(app: FastAPI):
|
||||||
"""Application lifespan: init DB, start scheduler, connect cache."""
|
"""Application lifespan: init DB, seed admin, start scheduler, connect cache."""
|
||||||
await init_db()
|
await init_db()
|
||||||
|
await seed_admin_user()
|
||||||
await cache.connect()
|
await cache.connect()
|
||||||
|
|
||||||
_mcp_cm = mcp_session_manager.run()
|
_mcp_cm = mcp_session_manager.run()
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
"""Admin router: login, sync endpoints, sync log."""
|
"""Admin router: login, sync endpoints, sync log."""
|
||||||
from fastapi import APIRouter, Depends, HTTPException, Response, Query, status
|
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Response, Query, status
|
||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
from typing import Any
|
from typing import Any
|
||||||
@@ -10,7 +10,9 @@ from app.schemas.auth import LoginRequest, TokenResponse, AdminInfo
|
|||||||
from app.schemas.sync import SyncStatus, SyncLogEntry, SyncTriggerResponse
|
from app.schemas.sync import SyncStatus, SyncLogEntry, SyncTriggerResponse
|
||||||
from app.auth import verify_password, create_access_token, get_current_user
|
from app.auth import verify_password, create_access_token, get_current_user
|
||||||
from app.services.sync_service import SyncService
|
from app.services.sync_service import SyncService
|
||||||
|
from app.database import async_session
|
||||||
from app.cache import cache
|
from app.cache import cache
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
||||||
|
|
||||||
@@ -54,13 +56,37 @@ async def get_me(user: AdminUser = Depends(get_current_user)) -> Any:
|
|||||||
|
|
||||||
@router.post("/sync", response_model=SyncTriggerResponse)
|
@router.post("/sync", response_model=SyncTriggerResponse)
|
||||||
async def trigger_sync(
|
async def trigger_sync(
|
||||||
|
background_tasks: BackgroundTasks,
|
||||||
user: AdminUser = Depends(get_current_user),
|
user: AdminUser = Depends(get_current_user),
|
||||||
db: AsyncSession = Depends(get_db),
|
db: AsyncSession = Depends(get_db),
|
||||||
) -> Any:
|
) -> Any:
|
||||||
"""Trigger a manual equipment sync (admin only)."""
|
"""Trigger a manual equipment sync in the background (admin only).
|
||||||
sync_service = SyncService(db)
|
|
||||||
result = await sync_service.run_sync()
|
Runs the sync via BackgroundTasks so the HTTP request returns immediately
|
||||||
return SyncTriggerResponse(sync_id=result["sync_id"], status=result["status"])
|
(a full sync of 700+ items incl. image downloads takes minutes and
|
||||||
|
would otherwise hit proxy timeouts).
|
||||||
|
Poll GET /api/admin/sync-status for progress.
|
||||||
|
"""
|
||||||
|
log_entry = SyncLog(sync_type="equipment", status="running", started_at=datetime.utcnow())
|
||||||
|
db.add(log_entry)
|
||||||
|
await db.commit()
|
||||||
|
await db.refresh(log_entry)
|
||||||
|
|
||||||
|
async def _run_sync(sync_id: int) -> None:
|
||||||
|
async with async_session() as sync_db:
|
||||||
|
sync_service = SyncService(sync_db)
|
||||||
|
try:
|
||||||
|
await sync_service.run_sync(sync_log_id=sync_id)
|
||||||
|
except Exception as exc: # noqa: BLE001
|
||||||
|
log = await sync_db.get(SyncLog, sync_id)
|
||||||
|
if log:
|
||||||
|
log.status = "failed"
|
||||||
|
log.error_message = str(exc)
|
||||||
|
log.completed_at = datetime.utcnow()
|
||||||
|
await sync_db.commit()
|
||||||
|
|
||||||
|
background_tasks.add_task(_run_sync, log_entry.id)
|
||||||
|
return SyncTriggerResponse(sync_id=log_entry.id, status="running")
|
||||||
|
|
||||||
|
|
||||||
@router.get("/sync-status", response_model=SyncStatus)
|
@router.get("/sync-status", response_model=SyncStatus)
|
||||||
|
|||||||
@@ -23,10 +23,10 @@ class SyncService:
|
|||||||
self.db = db
|
self.db = db
|
||||||
self.rentman = rentman or RentmanService()
|
self.rentman = rentman or RentmanService()
|
||||||
|
|
||||||
async def run_sync(self) -> dict[str, Any]:
|
async def run_sync(self, sync_log_id: int | None = None) -> dict[str, Any]:
|
||||||
"""Execute an incremental equipment sync.
|
"""Execute an incremental equipment sync.
|
||||||
|
|
||||||
1. Create sync_log entry (status=running)
|
1. Create sync_log entry (status=running) or reuse sync_log_id
|
||||||
2. Paginate GET /equipment from Rentman
|
2. Paginate GET /equipment from Rentman
|
||||||
3. Compare updateHash with DB, only process changed items
|
3. Compare updateHash with DB, only process changed items
|
||||||
4. Download images only for changed items
|
4. Download images only for changed items
|
||||||
@@ -35,12 +35,20 @@ class SyncService:
|
|||||||
7. Update sync_log (status=completed or failed)
|
7. Update sync_log (status=completed or failed)
|
||||||
Returns dict with sync_id, items_processed, status.
|
Returns dict with sync_id, items_processed, status.
|
||||||
"""
|
"""
|
||||||
log_entry = SyncLog(
|
if sync_log_id is not None:
|
||||||
sync_type="equipment",
|
log_entry = await self.db.get(SyncLog, sync_log_id)
|
||||||
status="running",
|
if not log_entry:
|
||||||
started_at=datetime.utcnow(),
|
log_entry = None
|
||||||
)
|
if sync_log_id is None or log_entry is None:
|
||||||
self.db.add(log_entry)
|
log_entry = SyncLog(
|
||||||
|
sync_type="equipment",
|
||||||
|
status="running",
|
||||||
|
started_at=datetime.utcnow(),
|
||||||
|
)
|
||||||
|
self.db.add(log_entry)
|
||||||
|
else:
|
||||||
|
log_entry.status = "running"
|
||||||
|
log_entry.started_at = datetime.utcnow()
|
||||||
await self.db.commit()
|
await self.db.commit()
|
||||||
await self.db.refresh(log_entry)
|
await self.db.refresh(log_entry)
|
||||||
sync_id = log_entry.id
|
sync_id = log_entry.id
|
||||||
|
|||||||
@@ -35,15 +35,18 @@ async def test_sync_with_valid_token(client, seeded_admin, test_db):
|
|||||||
})
|
})
|
||||||
token = login_resp.json()["access_token"]
|
token = login_resp.json()["access_token"]
|
||||||
|
|
||||||
# Mock sync service
|
# Mock sync service (background task)
|
||||||
with patch.object(SyncService, "run_sync", new_callable=AsyncMock) as mock_sync:
|
with patch.object(SyncService, "run_sync", new_callable=AsyncMock) as mock_sync:
|
||||||
mock_sync.return_value = {"sync_id": 42, "items_processed": 10, "items_failed": 0, "status": "completed"}
|
mock_sync.return_value = {"sync_id": 1, "items_processed": 10, "items_failed": 0, "status": "completed"}
|
||||||
resp = await client.post("/api/admin/sync", cookies={"hms_admin_token": token})
|
resp = await client.post("/api/admin/sync", cookies={"hms_admin_token": token})
|
||||||
|
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
assert data["sync_id"] == 42
|
# Background sync: request returns immediately with a running log entry
|
||||||
assert data["status"] == "completed"
|
assert data["sync_id"] == 1
|
||||||
|
assert data["status"] == "running"
|
||||||
|
# Background task executed by TestClient: run_sync was called once
|
||||||
|
mock_sync.assert_awaited_once()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|||||||
Reference in New Issue
Block a user