life-noc/scripts/life_noc_input_api.py

61 lines
1.7 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
from pathlib import Path
import sys
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from lib.life_noc_inputs import list_items, get_item, set_item, complete_item # noqa: E402
class SetInputRequest(BaseModel):
value: str
origin: str = "manual"
class CompleteInputRequest(BaseModel):
origin: str = "manual"
app = FastAPI(title="Life-NOC Input API", version="1.0.0")
@app.get("/health")
def health() -> dict:
return {"status": "ok"}
@app.get("/inputs/{domain}")
def api_list_inputs(domain: str) -> dict:
return {"domain": domain, "items": list_items(domain)}
@app.get("/inputs/{domain}/{item_key}")
def api_get_input(domain: str, item_key: str) -> dict:
try:
item = get_item(domain, item_key)
except KeyError:
raise HTTPException(status_code=404, detail="item introuvable")
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {"domain": domain, "item_key": item_key, **item}
@app.post("/inputs/{domain}/{item_key}")
def api_set_input(domain: str, item_key: str, req: SetInputRequest) -> dict:
item = set_item(domain, item_key, req.value, origin=req.origin)
return {"domain": domain, "item_key": item_key, **item}
@app.post("/inputs/{domain}/{item_key}/complete")
def api_complete_input(domain: str, item_key: str, req: CompleteInputRequest | None = None) -> dict:
origin = "manual" if req is None else req.origin
item = complete_item(domain, item_key, origin=origin)
return {"domain": domain, "item_key": item_key, **item}