69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
"""FastAPI App: phpipam -> Outline Sync."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
|
|
from .config import settings
|
|
from .phpipam_client import PhpIpamClient
|
|
from .scheduler import scheduler, start_scheduler, stop_scheduler
|
|
from .sync import run_sync
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
|
)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
start_scheduler()
|
|
yield
|
|
stop_scheduler()
|
|
|
|
|
|
app = FastAPI(title="phpipam-outline-sync", version="0.1.0", lifespan=lifespan)
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/subnets")
|
|
def subnets() -> list[dict]:
|
|
"""Nur zum Testen der phpipam-Verbindung, ohne Outline anzufassen."""
|
|
try:
|
|
with PhpIpamClient() as ipam:
|
|
return ipam.list_subnets()
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(status_code=502, detail=f"phpipam-Fehler: {exc}") from exc
|
|
|
|
|
|
@app.post("/sync")
|
|
def sync(dry_run: bool | None = None) -> dict:
|
|
"""Synchronisiert alle Subnetze + Adressen aus phpipam nach Outline.
|
|
|
|
?dry_run=true -> nichts wird in Outline geschrieben, nur eine Vorschau.
|
|
"""
|
|
try:
|
|
results = run_sync(dry_run=dry_run)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(status_code=502, detail=str(exc)) from exc
|
|
return {"count": len(results), "results": results}
|
|
|
|
|
|
@app.get("/schedule")
|
|
def schedule_info() -> dict:
|
|
"""Zeigt, ob der automatische Zeitplan aktiv ist und wann er als nächstes läuft."""
|
|
if not settings.sync_schedule_enabled or not scheduler.running:
|
|
return {"enabled": False, "cron": settings.sync_schedule_cron}
|
|
job = scheduler.get_job("phpipam-outline-sync")
|
|
return {
|
|
"enabled": True,
|
|
"cron": settings.sync_schedule_cron,
|
|
"next_run": job.next_run_time.isoformat() if job and job.next_run_time else None,
|
|
}
|