108 lines
2.9 KiB
Python
108 lines
2.9 KiB
Python
import json
|
|
import os
|
|
import pprint
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import logging
|
|
from helpers import flatten_dict
|
|
|
|
load_dotenv()
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d %(funcName)s %(message)s",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
origins = [
|
|
"http://localhost",
|
|
"http://localhost:3010",
|
|
"http://localhost:4173",
|
|
"http://localhost:5173",
|
|
"http://localhost:8000",
|
|
]
|
|
|
|
vtk = json.loads(os.getenv("VTK_KEYS"))
|
|
vtk_org = {}
|
|
|
|
api_app = FastAPI(title="api app")
|
|
|
|
|
|
@api_app.get("/")
|
|
def test():
|
|
return {"app": "service monitoring api"}
|
|
|
|
|
|
@api_app.get("/vtk/{item_id}")
|
|
def get_vtk(item_id):
|
|
if not item_id in vtk:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
try:
|
|
if not item_id in vtk_org:
|
|
res = requests.get(
|
|
"https://my.vendotek.com/api/v1/org",
|
|
headers={"Authorization": f"Bearer {vtk[item_id]}"},
|
|
)
|
|
data = res.json()
|
|
org_id = data[0]["name"]
|
|
vtk_org[item_id] = org_id
|
|
|
|
res_unit = requests.get(
|
|
f"https://my.vendotek.com/api/v1/org/{vtk_org[item_id]}/unit",
|
|
headers={"Authorization": f"Bearer {vtk[item_id]}"},
|
|
)
|
|
data_unit = res_unit.json()
|
|
result = [
|
|
{
|
|
"sn": d["sn"],
|
|
"last_online": d["last_seen_at"],
|
|
"name": ", ".join(
|
|
list(filter(None, [d["location_name"], d["address"], d["city"]]))
|
|
),
|
|
"ip": list(
|
|
filter(
|
|
lambda item: item["name"] == "IP address",
|
|
list(
|
|
filter(
|
|
lambda item: item["id"] == "LAN",
|
|
list(
|
|
filter(
|
|
lambda item: item["id"] == "Communications",
|
|
d["modules"],
|
|
)
|
|
)[0]["modules"],
|
|
)
|
|
)[0]["details"],
|
|
)
|
|
)[0]["value"],
|
|
}
|
|
for d in data_unit
|
|
]
|
|
return {"status": "success", "data": result}
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
app = FastAPI(title="main app")
|
|
app.exception_handler(404)
|
|
|
|
|
|
async def return_error():
|
|
return {"error": "custom error"}
|
|
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
app.mount("/api", api_app)
|
|
app.mount("/", StaticFiles(directory="front/.output/public", html=True), name="front")
|