gauvainboiche
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
import json
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from urllib.parse import urlparse
|
|
|
|
_lock = threading.Lock()
|
|
_equipments = []
|
|
_next_id = 1
|
|
MAX_BODY = 1_048_576 # 1 Mo
|
|
|
|
|
|
class InventoryHandler(BaseHTTPRequestHandler):
|
|
def _send_json(self, code, payload):
|
|
body = json.dumps(payload).encode("utf-8")
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _read_json(self):
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
|
|
if not length or length > MAX_BODY:
|
|
raise ValueError("Corps absent ou trop volumineux.")
|
|
|
|
if "application/json" not in self.headers.get("Content-Type", ""):
|
|
raise ValueError("Content-Type invalide.")
|
|
|
|
return json.loads(self.rfile.read(length))
|
|
|
|
def do_GET(self):
|
|
path = urlparse(self.path).path
|
|
parts = path.strip("/").split("/")
|
|
|
|
if parts[0] != "equipments":
|
|
return self._send_json(404, {"erreur": "route inconnue"})
|
|
|
|
with _lock:
|
|
if len(parts) == 1:
|
|
return self._send_json(200, _equipments)
|
|
if len(parts) == 2:
|
|
try:
|
|
eid = int(parts[1])
|
|
except ValueError:
|
|
return self._send_json(400, {"erreur": "id non entier"})
|
|
for e in _equipments:
|
|
if e["id"] == eid:
|
|
return self._send_json(200, e)
|
|
return self._send_json(404, {"erreur": "équipement introuvable"})
|
|
|
|
return self._send_json(405, {"erreur": "méthode non autorisée"})
|
|
|
|
def do_POST(self):
|
|
global _next_id
|
|
|
|
if urlparse(self.path).path.rstrip("/") != "/equipments":
|
|
return self._send_json(404, {"erreur": "route inconnue"})
|
|
|
|
try:
|
|
data = self._read_json()
|
|
except (ValueError, json.JSONDecodeError) as exc:
|
|
return self._send_json(400, {"erreur": f"JSON invalide : {exc}"})
|
|
if not isinstance(data, dict) or "hostname" not in data or "ip" not in data:
|
|
return self._send_json(400, {"erreur": "hostname et ip requis"})
|
|
|
|
with _lock:
|
|
equipement = {"id": _next_id, "hostname": data["hostname"], "ip": data["ip"]}
|
|
_equipments.append(equipement)
|
|
_next_id += 1
|
|
|
|
self._send_json(201, equipement)
|
|
|
|
if __name__ == "__main__":
|
|
HTTPServer(("127.0.0.1", 8000), InventoryHandler).serve_forever()
|