Joachim Friberg
142 lines
4.5 KiB
Python
142 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from app.config import get_settings
|
|
from app.docker_api import DockerApiClient, DockerApiError, DockerUsageResolver
|
|
from app.ip_commands import CommandError, IpAddressManager
|
|
from app.service import (
|
|
ConflictError,
|
|
DependencyError,
|
|
EntryService,
|
|
NotFoundError,
|
|
ValidationError,
|
|
)
|
|
from app.storage import EntryStorage
|
|
|
|
|
|
def build_service() -> EntryService:
|
|
settings = get_settings()
|
|
storage = EntryStorage(settings.state_file)
|
|
docker_client = DockerApiClient(settings.docker_api_url, timeout_seconds=settings.docker_timeout_seconds)
|
|
usage_resolver = DockerUsageResolver(docker_client)
|
|
ip_manager = IpAddressManager()
|
|
return EntryService(storage=storage, usage_resolver=usage_resolver, ip_manager=ip_manager)
|
|
|
|
|
|
service = build_service()
|
|
app = FastAPI(title="Docker IP Addr Manager", version="0.1.0")
|
|
|
|
static_dir = Path(__file__).parent / "static"
|
|
app.mount("/static", StaticFiles(directory=static_dir), name="static")
|
|
|
|
|
|
@app.on_event("startup")
|
|
def startup_reconcile() -> None:
|
|
errors = service.reconcile_enabled_entries()
|
|
if errors:
|
|
for error in errors:
|
|
print(f"[startup-reconcile] {error}")
|
|
|
|
|
|
@app.get("/")
|
|
def index() -> FileResponse:
|
|
return FileResponse(static_dir / "index.html")
|
|
|
|
|
|
@app.get("/healthz")
|
|
def healthz() -> dict:
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/api/interfaces")
|
|
def list_interfaces() -> dict:
|
|
try:
|
|
return {"items": service.list_interfaces()}
|
|
except DependencyError as exc:
|
|
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
|
|
|
|
|
@app.get("/api/entries")
|
|
def list_entries() -> dict:
|
|
response = service.list_entries()
|
|
return response.to_dict()
|
|
|
|
|
|
@app.post("/api/refresh")
|
|
def refresh_entries() -> dict:
|
|
response = service.list_entries()
|
|
return response.to_dict()
|
|
|
|
|
|
@app.post("/api/entries")
|
|
def create_entry(payload: dict) -> dict:
|
|
try:
|
|
entry = service.create_entry(payload)
|
|
return entry.to_dict()
|
|
except ValidationError as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
except ConflictError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
|
|
|
|
@app.put("/api/entries/{entry_id}")
|
|
def update_entry(entry_id: str, payload: dict) -> dict:
|
|
try:
|
|
entry = service.update_entry(entry_id, payload)
|
|
return entry.to_dict()
|
|
except ValidationError as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
except ConflictError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
except NotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
|
|
|
|
@app.post("/api/entries/{entry_id}/enable")
|
|
def enable_entry(entry_id: str) -> dict:
|
|
try:
|
|
entry = service.set_enabled(entry_id, enabled=True)
|
|
return entry.to_dict()
|
|
except NotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except (ConflictError, CommandError) as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
except DependencyError as exc:
|
|
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
|
|
|
|
|
@app.post("/api/entries/{entry_id}/disable")
|
|
def disable_entry(entry_id: str) -> dict:
|
|
try:
|
|
entry = service.set_enabled(entry_id, enabled=False)
|
|
return entry.to_dict()
|
|
except NotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except (ConflictError, CommandError) as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
except DependencyError as exc:
|
|
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
|
|
|
|
|
@app.delete("/api/entries/{entry_id}")
|
|
def delete_entry(entry_id: str) -> dict:
|
|
try:
|
|
service.delete_entry(entry_id)
|
|
return {"deleted": True}
|
|
except NotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except ConflictError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
except DependencyError as exc:
|
|
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
|
|
|
|
|
@app.exception_handler(DockerApiError)
|
|
async def docker_error_handler(_, exc: DockerApiError):
|
|
return JSONResponse(status_code=503, content={"detail": str(exc)})
|