from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
import requests
import os
import subprocess
import sqlite3
from wakeonlan import send_magic_packet
from pydantic import BaseModel
from fastapi.openapi.utils import get_openapi
# Database setup
def init_db():
conn = sqlite3.connect("servers.db")
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS servers (
name TEXT PRIMARY KEY,
ip TEXT NOT NULL,
mac TEXT NOT NULL)''')
conn.commit()
conn.close()
init_db()
def get_server_info(server_name):
conn = sqlite3.connect("servers.db")
cursor = conn.cursor()
cursor.execute("SELECT ip, mac FROM servers WHERE name = ?", (server_name,))
server = cursor.fetchone()
conn.close()
if server:
return {"ip": server[0], "mac": server[1]}
return None
def list_servers():
conn = sqlite3.connect("servers.db")
cursor = conn.cursor()
cursor.execute("SELECT name, ip, mac FROM servers")
servers = cursor.fetchall()
conn.close()
return [{"name": s[0], "ip": s[1], "mac": s[2]} for s in servers]
def add_or_update_server(name: str, ip: str, mac: str):
conn = sqlite3.connect("servers.db")
cursor = conn.cursor()
cursor.execute("REPLACE INTO servers (name, ip, mac) VALUES (?, ?, ?)", (name, ip, mac))
conn.commit()
conn.close()
app = FastAPI()
class ServerModel(BaseModel):
name: str
ip: str
mac: str
def get_proxmox_status(server_ip):
"""Fetch Proxmox server status."""
try:
response = requests.get(f"https://{server_ip}:8006/api2/json/nodes", verify=False) # Assuming no authentication
response.raise_for_status()
data = response.json()
return {"status": data["data"][0]["status"]} # Extract node status
except requests.RequestException as e:
return {"status": "unknown", "error": str(e)}
@app.get("/openapi.json")
def get_openapi_spec():
"""Returns the OpenAPI specification."""
return get_openapi(title=app.title, version="1.0.0", routes=app.routes)
@app.get("/apidoc", response_class=HTMLResponse)
def api_docs():
"""Returns the API documentation using RapiDoc."""
return HTMLResponse(content="""
API Documentation
""", status_code=200)
@app.get("/statuses/{server_name}")
def status(server_name: str):
"""Returns the Proxmox server status for a specific server."""
server = get_server_info(server_name)
if not server:
raise HTTPException(status_code=404, detail="Server not found")
return get_proxmox_status(server["ip"])
@app.get("/statuses")
def list_all_statuses():
"""Returns the statuses of all servers."""
servers = list_servers()
return {server["name"]: get_proxmox_status(server["ip"]) for server in servers}
def check_power_state(server_ip):
"""Check if the server is online by pinging it."""
response = subprocess.run(["ping", "-c", "1", server_ip], stdout=subprocess.DEVNULL)
return "on" if response.returncode == 0 else "off"
@app.get("/states/{server_name}")
def get_power_status(server_name: str):
"""Returns the power status of a specific server."""
server = get_server_info(server_name)
if not server:
raise HTTPException(status_code=404, detail="Server not found")
return {"power": check_power_state(server["ip"])}
@app.get("/states")
def list_all_states():
"""Returns the power states of all servers."""
servers = list_servers()
return {server["name"]: check_power_state(server["ip"]) for server in servers}
@app.put("/states/{server_name}")
@app.patch("/states/{server_name}")
def control_power(server_name: str, state: str):
"""Controls the server power state (wake or shutdown)."""
server = get_server_info(server_name)
if not server:
raise HTTPException(status_code=404, detail="Server not found")
if state == "on":
if check_power_state(server["ip"]) == "off":
send_magic_packet(server["mac"])
return {"message": "Wake-on-LAN signal sent"}
return {"message": "Server is already on"}
elif state == "off":
try:
requests.post(f"https://{server["ip"]}:8006/api2/json/nodes/shutdown", verify=False)
return {"message": "Shutdown command sent"}
except requests.RequestException as e:
raise HTTPException(status_code=500, detail=str(e))
else:
raise HTTPException(status_code=400, detail="Invalid state. Use 'on' or 'off'")
@app.get("/servers")
def get_servers():
"""Returns the list of registered servers."""
return list_servers()
@app.post("/servers")
def add_server(server: ServerModel):
"""Adds or updates a server."""
add_or_update_server(server.name, server.ip, server.mac)
return {"message": "Server added/updated successfully"}
# Run the server with: uvicorn script_name:app --host 0.0.0.0 --port 8000