from datetime import datetime, UTC from typing import List from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from .database import SessionLocal, engine from . import models, schemas, crud models.Base.metadata.create_all(bind=engine) app = FastAPI() # Dependency def get_db(): db = SessionLocal() try: yield db finally: db.close() @app.get("/hosts/", response_model=List[schemas.Host]) def get_hosts(db: Session = Depends(get_db)): hosts = crud.get_hosts(db) return hosts @app.get("/get_host_by_hostname/{hostname}") def get_host_by_hostname(hostname: str, db: Session = Depends(get_db)): return crud.get_host_by_hostname(db, hostname=hostname) @app.get("/get_host_id_for_hostname/{hostname}") def get_host_id_for_hostname(hostname: str, db: Session = Depends(get_db)): return crud.get_host_id_for_hostname(db, hostname=hostname) @app.post("/add_host/", response_model=schemas.HostCreate, status_code=201) def create_host(host: schemas.HostCreate, db: Session = Depends(get_db)): db_host = crud.get_host_by_hostname(db, hostname=host.hostname) if db_host: raise HTTPException(status_code=400, detail="Host already exists") return crud.create_host(db=db, host=host) @app.delete("/delete_host/{host_id}") def delete_host(host_id: int, db: Session = Depends(get_db)): return crud.delete_host(db, host_id=host_id) @app.delete("/delete_host_by_hostname/{hostname}") def delete_host_by_hostname(hostname: str, db: Session = Depends(get_db)): return crud.delete_host_by_hostname(db, hostname=hostname) @app.get("/get_cpus_by_host_id/{host_id}", response_model=List[schemas.CPU]) def get_cpus_by_host_id(host_id: int, db: Session = Depends(get_db)) -> List[schemas.CPU]: return crud.get_cpus_by_host_id(db, host_id=host_id) @app.post("/add_cpu_to_host/", response_model=schemas.CPUCreate, status_code=201) def create_cpu(cpu: schemas.CPUCreate, db: Session = Depends(get_db)): if cpu.time == datetime.fromtimestamp(0, UTC): cpu.time = datetime.now() return crud.create_cpu(db, cpu=cpu)