145 lines
4.5 KiB
Python
145 lines
4.5 KiB
Python
from fastapi import FastAPI, Request, HTTPException, Form, responses
|
|
from fastapi.responses import RedirectResponse, HTMLResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from typing import Annotated
|
|
from pydantic import BaseModel
|
|
|
|
from sqlmodel import SQLModel, Relationship, Session, create_engine, select
|
|
|
|
import json
|
|
from models import ordinateur, employee
|
|
|
|
app = FastAPI()
|
|
#ordi1 = ordinateur()
|
|
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
#sqlitemodel
|
|
sqlite_file_name = "mydb.db"
|
|
sqlite_url = f"sqlite:///mydb.db"
|
|
|
|
engine = create_engine(sqlite_url)
|
|
|
|
def create_db_and_tables():
|
|
SQLModel.metadata.create_all(engine)
|
|
|
|
SQLModel.metadata.create_all(engine)
|
|
|
|
@app.get("/ordi/{ordi_id}", response_class=HTMLResponse)
|
|
async def get_ordi1_info(request: Request, ordi_id: int):
|
|
with Session(engine) as session:
|
|
statement= select(ordinateur).where(ordinateur.id == ordi_id)
|
|
this_ordi = session.exec(statement).first()
|
|
return templates.TemplateResponse(
|
|
request=request, name="item.html", context={"ordi": this_ordi}
|
|
)
|
|
|
|
@app.get("/employee/create", response_class=HTMLResponse)
|
|
async def get_employee_form(request: Request):
|
|
return templates.TemplateResponse(
|
|
"employee_form.html", {"request" : request})
|
|
|
|
@app.post("/employee/add", response_class=HTMLResponse)
|
|
async def submit_employee_form(
|
|
first_name: str = Form(...),
|
|
family_name: str = Form(...),
|
|
badge_number: str = Form(...)
|
|
):
|
|
with Session(engine) as session:
|
|
new_employee = employee(
|
|
first_name=first_name,
|
|
family_name=family_name,
|
|
badge_number=badge_number
|
|
)
|
|
session.add(new_employee)
|
|
session.commit()
|
|
|
|
return responses.RedirectResponse(
|
|
url="/employees", status_code=303)
|
|
|
|
@app.get("/employees", response_class=HTMLResponse)
|
|
async def get_employees(request: Request):
|
|
with Session(engine) as session:
|
|
statement= select(employee)
|
|
employees= session.exec(statement).all()
|
|
return templates.TemplateResponse(
|
|
"employees.html", {"request" : request, "employees" : employees}
|
|
)
|
|
|
|
|
|
@app.post("/employee/{e_id}/delete")
|
|
async def delete_employee(e_id: int):
|
|
with Session(engine) as session:
|
|
db_e = session.get(employee, e_id)
|
|
if not db_e:
|
|
raise HTTPException(status_code=404, detail="Employé non trouvé")
|
|
session.delete(db_e)
|
|
session.commit()
|
|
return responses.RedirectResponse(url="/employees", status_code=303)
|
|
|
|
@app.post("/endpoint")
|
|
async def receive_info(request: Request):
|
|
# Lire le body brut
|
|
body = await request.body()
|
|
print(body)
|
|
|
|
|
|
# Parser le JSON
|
|
try:
|
|
data = json.loads(body)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(status_code=400, detail="Invalid JSON")
|
|
|
|
|
|
# Debug
|
|
hardware = data.get('HARDWARE',{})
|
|
software = data.get('SOFTWARE',{})
|
|
|
|
with Session(engine) as session:
|
|
statement = select(ordinateur).where(ordinateur.mac_address == hardware.get("mac_address", ""))
|
|
ordi1 = session.exec(statement).first()
|
|
|
|
|
|
is_new = False
|
|
if not ordi1:
|
|
ordi1 = ordinateur()
|
|
is_new = True
|
|
|
|
print("Infos reçues :", data)
|
|
ordi1.mb_serial = hardware.get("mb_serial", "")
|
|
ordi1.hostname = hardware.get("hostname", "")
|
|
ordi1.cpu = hardware.get("cpu", "")
|
|
ordi1.cpu_id = hardware.get("cpu_id", "")
|
|
ordi1.memory_mb = hardware.get("memory_mb", "")
|
|
ordi1.ram_size = hardware.get("ram_size", "")
|
|
ordi1.ram_gen = hardware.get("ram_gen", "")
|
|
ordi1.sizes = hardware.get("sizes", "")
|
|
ordi1.cpu_freq_min = hardware.get("cpu_freq_min", "")
|
|
ordi1.cpu_freq_cur = hardware.get("cpu_freq_cur", "")
|
|
ordi1.cpu_freq_max = hardware.get("cpu_freq_max", "")
|
|
ordi1.gpu_model = hardware.get("gpu_model", "")
|
|
ordi1.chassis_serial = hardware.get("chassis_serial", "")
|
|
ordi1.cpu_cores_nb = hardware.get("cpu_cores_nb", "")
|
|
ordi1.cpu_threads_nb = hardware.get("cpu_threads_nb", "")
|
|
ordi1.ram_number = hardware.get("ram_number", "")
|
|
ordi1.ram_slots_nb = hardware.get("ram_slots_nb", "")
|
|
ordi1.mac_address = hardware.get("mac_address", "")
|
|
|
|
ordi1.os = software.get("os", "")
|
|
ordi1.arch = software.get("arch", "")
|
|
ordi1.desktop = software.get("desktop", "")
|
|
ordi1.wm = software.get("wm", "")
|
|
ordi1.kernel = software.get("kernel", "")
|
|
|
|
if is_new:
|
|
session.add(ordi1)
|
|
|
|
|
|
session.commit()
|
|
#session.close()
|
|
|
|
return ({"status": "ok"})
|
