441 lines
14 KiB
Python
441 lines
14 KiB
Python
from fastapi import FastAPI, APIRouter, Depends, HTTPException, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
import uvicorn
|
|
import os
|
|
from dotenv import load_dotenv
|
|
import yaml
|
|
from pathlib import Path
|
|
from minio import Minio
|
|
from datetime import timedelta
|
|
|
|
from database import get_db_cursor, init_db
|
|
from auth import (
|
|
get_password_hash,
|
|
verify_password,
|
|
create_access_token,
|
|
get_current_user,
|
|
ACCESS_TOKEN_EXPIRE_MINUTES
|
|
)
|
|
from models import (
|
|
UserRegister,
|
|
UserLogin,
|
|
Token,
|
|
UserResponse,
|
|
SectionCreate,
|
|
SectionResponse,
|
|
AppCreate,
|
|
AppUpdate,
|
|
AppResponse,
|
|
AppEntry,
|
|
AppData
|
|
)
|
|
|
|
app = FastAPI(title="Navix API", version="2.0.0")
|
|
router = APIRouter()
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
load_dotenv()
|
|
|
|
MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT")
|
|
MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY")
|
|
MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY")
|
|
MINIO_BUCKET = os.getenv("MINIO_BUCKET")
|
|
|
|
minio_client = Minio(
|
|
MINIO_ENDPOINT,
|
|
access_key=MINIO_ACCESS_KEY,
|
|
secret_key=MINIO_SECRET_KEY,
|
|
secure=True
|
|
)
|
|
|
|
BUCKET = MINIO_BUCKET or "navix-icons"
|
|
APPS_FILE = Path(__file__).parent / "apps.yaml"
|
|
|
|
# Initialize database connection
|
|
@app.on_event("startup")
|
|
async def startup_event():
|
|
if init_db():
|
|
print("✅ Database connected successfully")
|
|
else:
|
|
print("⚠️ Database connection failed")
|
|
|
|
|
|
# ============================================================================
|
|
# Authentication Endpoints
|
|
# ============================================================================
|
|
|
|
@router.post("/auth/register", response_model=Token, status_code=status.HTTP_201_CREATED)
|
|
def register(user_data: UserRegister):
|
|
"""Register a new user"""
|
|
with get_db_cursor(commit=True) as cursor:
|
|
# Check if username exists
|
|
cursor.execute("SELECT id FROM users WHERE username = %s", (user_data.username,))
|
|
if cursor.fetchone():
|
|
raise HTTPException(status_code=400, detail="Username already exists")
|
|
|
|
# Check if email exists
|
|
cursor.execute("SELECT id FROM users WHERE email = %s", (user_data.email,))
|
|
if cursor.fetchone():
|
|
raise HTTPException(status_code=400, detail="Email already exists")
|
|
|
|
# Create user
|
|
hashed_password = get_password_hash(user_data.password)
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO users (username, email, password_hash)
|
|
VALUES (%s, %s, %s)
|
|
RETURNING id, username, email, created_at
|
|
""",
|
|
(user_data.username, user_data.email, hashed_password)
|
|
)
|
|
user = cursor.fetchone()
|
|
|
|
# Create access token
|
|
access_token = create_access_token(
|
|
data={"sub": user["id"], "username": user["username"]},
|
|
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
)
|
|
|
|
return Token(
|
|
access_token=access_token,
|
|
user=UserResponse(**user)
|
|
)
|
|
|
|
|
|
@router.post("/auth/login", response_model=Token)
|
|
def login(credentials: UserLogin):
|
|
"""Login user and return JWT token"""
|
|
with get_db_cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id, username, email, password_hash, created_at FROM users WHERE username = %s",
|
|
(credentials.username,)
|
|
)
|
|
user = cursor.fetchone()
|
|
|
|
if not user or not verify_password(credentials.password, user["password_hash"]):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect username or password"
|
|
)
|
|
|
|
# Create access token
|
|
access_token = create_access_token(
|
|
data={"sub": user["id"], "username": user["username"]},
|
|
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
)
|
|
|
|
return Token(
|
|
access_token=access_token,
|
|
user=UserResponse(
|
|
id=user["id"],
|
|
username=user["username"],
|
|
email=user["email"],
|
|
created_at=user["created_at"]
|
|
)
|
|
)
|
|
|
|
|
|
@router.get("/auth/me", response_model=UserResponse)
|
|
def get_me(current_user: dict = Depends(get_current_user)):
|
|
"""Get current user information"""
|
|
with get_db_cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id, username, email, created_at FROM users WHERE id = %s",
|
|
(current_user["user_id"],)
|
|
)
|
|
user = cursor.fetchone()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return UserResponse(**user)
|
|
|
|
|
|
# ============================================================================
|
|
# Section Endpoints
|
|
# ============================================================================
|
|
|
|
@router.get("/sections", response_model=list[SectionResponse])
|
|
def get_sections(current_user: dict = Depends(get_current_user)):
|
|
"""Get all sections with apps for the current user"""
|
|
with get_db_cursor() as cursor:
|
|
# Get sections for user
|
|
cursor.execute(
|
|
"""
|
|
SELECT id, name, display_order
|
|
FROM sections
|
|
WHERE user_id = %s
|
|
ORDER BY display_order, name
|
|
""",
|
|
(current_user["user_id"],)
|
|
)
|
|
sections = cursor.fetchall()
|
|
|
|
result = []
|
|
for section in sections:
|
|
# Get apps for each section
|
|
cursor.execute(
|
|
"""
|
|
SELECT id, section_id, name, url, icon, description, display_order
|
|
FROM apps
|
|
WHERE section_id = %s
|
|
ORDER BY display_order, name
|
|
""",
|
|
(section["id"],)
|
|
)
|
|
apps = cursor.fetchall()
|
|
|
|
result.append({
|
|
"id": section["id"],
|
|
"name": section["name"],
|
|
"display_order": section["display_order"],
|
|
"apps": [dict(app) for app in apps]
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
@router.post("/sections", response_model=SectionResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_section(section: SectionCreate, current_user: dict = Depends(get_current_user)):
|
|
"""Create a new section for the current user"""
|
|
with get_db_cursor(commit=True) as cursor:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO sections (user_id, name, display_order)
|
|
VALUES (%s, %s, COALESCE((SELECT MAX(display_order) + 1 FROM sections WHERE user_id = %s), 0))
|
|
RETURNING id, name, display_order
|
|
""",
|
|
(current_user["user_id"], section.name, current_user["user_id"])
|
|
)
|
|
new_section = cursor.fetchone()
|
|
return SectionResponse(**new_section, apps=[])
|
|
|
|
|
|
# ============================================================================
|
|
# App Endpoints
|
|
# ============================================================================
|
|
|
|
@router.post("/apps", response_model=AppResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_app(app: AppCreate, current_user: dict = Depends(get_current_user)):
|
|
"""Create a new app in a section"""
|
|
with get_db_cursor(commit=True) as cursor:
|
|
# Verify section belongs to user
|
|
cursor.execute(
|
|
"SELECT id FROM sections WHERE id = %s AND user_id = %s",
|
|
(app.section_id, current_user["user_id"])
|
|
)
|
|
if not cursor.fetchone():
|
|
raise HTTPException(status_code=404, detail="Section not found")
|
|
|
|
# Create app
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO apps (section_id, name, url, icon, description, display_order)
|
|
VALUES (%s, %s, %s, %s, %s, COALESCE((SELECT MAX(display_order) + 1 FROM apps WHERE section_id = %s), 0))
|
|
RETURNING id, section_id, name, url, icon, description, display_order
|
|
""",
|
|
(app.section_id, app.name, app.url, app.icon, app.description, app.section_id)
|
|
)
|
|
new_app = cursor.fetchone()
|
|
return AppResponse(**new_app)
|
|
|
|
|
|
@router.put("/apps/{app_id}", response_model=AppResponse)
|
|
def update_app(app_id: int, app_update: AppUpdate, current_user: dict = Depends(get_current_user)):
|
|
"""Update an existing app"""
|
|
with get_db_cursor(commit=True) as cursor:
|
|
# Verify app belongs to user
|
|
cursor.execute(
|
|
"""
|
|
SELECT a.id FROM apps a
|
|
JOIN sections s ON a.section_id = s.id
|
|
WHERE a.id = %s AND s.user_id = %s
|
|
""",
|
|
(app_id, current_user["user_id"])
|
|
)
|
|
if not cursor.fetchone():
|
|
raise HTTPException(status_code=404, detail="App not found")
|
|
|
|
# Build update query dynamically
|
|
update_fields = []
|
|
values = []
|
|
|
|
if app_update.name is not None:
|
|
update_fields.append("name = %s")
|
|
values.append(app_update.name)
|
|
if app_update.url is not None:
|
|
update_fields.append("url = %s")
|
|
values.append(app_update.url)
|
|
if app_update.icon is not None:
|
|
update_fields.append("icon = %s")
|
|
values.append(app_update.icon)
|
|
if app_update.description is not None:
|
|
update_fields.append("description = %s")
|
|
values.append(app_update.description)
|
|
if app_update.section_id is not None:
|
|
# Verify new section belongs to user
|
|
cursor.execute(
|
|
"SELECT id FROM sections WHERE id = %s AND user_id = %s",
|
|
(app_update.section_id, current_user["user_id"])
|
|
)
|
|
if not cursor.fetchone():
|
|
raise HTTPException(status_code=404, detail="Target section not found")
|
|
update_fields.append("section_id = %s")
|
|
values.append(app_update.section_id)
|
|
|
|
if not update_fields:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
|
|
values.append(app_id)
|
|
query = f"UPDATE apps SET {', '.join(update_fields)} WHERE id = %s RETURNING id, section_id, name, url, icon, description, display_order"
|
|
|
|
cursor.execute(query, values)
|
|
updated_app = cursor.fetchone()
|
|
return AppResponse(**updated_app)
|
|
|
|
|
|
@router.delete("/apps/{app_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_app_by_id(app_id: int, current_user: dict = Depends(get_current_user)):
|
|
"""Delete an app"""
|
|
with get_db_cursor(commit=True) as cursor:
|
|
# Verify app belongs to user
|
|
cursor.execute(
|
|
"""
|
|
DELETE FROM apps
|
|
WHERE id = %s AND section_id IN (
|
|
SELECT id FROM sections WHERE user_id = %s
|
|
)
|
|
RETURNING id
|
|
""",
|
|
(app_id, current_user["user_id"])
|
|
)
|
|
if not cursor.fetchone():
|
|
raise HTTPException(status_code=404, detail="App not found")
|
|
|
|
|
|
# ============================================================================
|
|
# Legacy YAML Endpoints (for backward compatibility)
|
|
# ============================================================================
|
|
|
|
|
|
# ============================================================================
|
|
# Legacy YAML Endpoints (for backward compatibility)
|
|
# ============================================================================
|
|
|
|
@router.get("/")
|
|
def root():
|
|
return {"message": "Welcome to Navix API v2.0!"}
|
|
|
|
|
|
@router.get("/apps")
|
|
def get_apps():
|
|
"""Legacy endpoint - returns apps from YAML file"""
|
|
if not APPS_FILE.exists():
|
|
return {"error": "apps.yaml not found"}
|
|
with open(APPS_FILE, "r") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
|
|
@router.post("/add_app")
|
|
def add_app(entry: AppEntry):
|
|
"""Legacy endpoint - adds app to YAML file"""
|
|
if not APPS_FILE.exists():
|
|
current = {"sections": []}
|
|
else:
|
|
with open(APPS_FILE, "r") as f:
|
|
current = yaml.safe_load(f) or {"sections": []}
|
|
|
|
for section in current["sections"]:
|
|
if section["name"] == entry.section:
|
|
section["apps"].append(entry.app.dict())
|
|
break
|
|
else:
|
|
current["sections"].append({
|
|
"name": entry.section,
|
|
"apps": [entry.app.dict()]
|
|
})
|
|
|
|
with open(APPS_FILE, "w") as f:
|
|
yaml.safe_dump(current, f)
|
|
|
|
return {"status": "added"}
|
|
|
|
|
|
@router.post("/edit_app")
|
|
def edit_app(entry: AppEntry):
|
|
"""Legacy endpoint - edits app in YAML file"""
|
|
if not APPS_FILE.exists():
|
|
return {"error": "apps.yaml not found"}
|
|
|
|
with open(APPS_FILE, "r") as f:
|
|
current = yaml.safe_load(f) or {"sections": []}
|
|
|
|
updated = False
|
|
for section in current["sections"]:
|
|
if section["name"] == entry.section:
|
|
for i, app in enumerate(section["apps"]):
|
|
if app["name"] == (entry.original_name or entry.app.name):
|
|
section["apps"][i] = entry.app.dict()
|
|
updated = True
|
|
break
|
|
break
|
|
|
|
if not updated:
|
|
return JSONResponse(status_code=404, content={"error": "App not found to edit"})
|
|
|
|
with open(APPS_FILE, "w") as f:
|
|
yaml.safe_dump(current, f)
|
|
|
|
return {"status": "updated"}
|
|
|
|
|
|
@router.post("/delete_app")
|
|
def delete_app(entry: AppEntry):
|
|
"""Legacy endpoint - deletes app from YAML file"""
|
|
if not APPS_FILE.exists():
|
|
return {"error": "apps.yaml not found"}
|
|
|
|
with open(APPS_FILE, "r") as f:
|
|
current = yaml.safe_load(f) or {"sections": []}
|
|
|
|
deleted = False
|
|
for section in current["sections"]:
|
|
if section["name"] == entry.section:
|
|
original_len = len(section["apps"])
|
|
section["apps"] = [a for a in section["apps"] if a["name"] != entry.app.name]
|
|
if len(section["apps"]) != original_len:
|
|
deleted = True
|
|
break
|
|
|
|
if not deleted:
|
|
return JSONResponse(status_code=404, content={"error": "App not found to delete"})
|
|
|
|
with open(APPS_FILE, "w") as f:
|
|
yaml.safe_dump(current, f)
|
|
|
|
return {"status": "deleted"}
|
|
|
|
|
|
@router.get("/icon/{filename}")
|
|
def get_public_icon_url(filename: str):
|
|
"""Get public URL for an icon from MinIO"""
|
|
url = f"https://{MINIO_ENDPOINT}/{BUCKET}/{filename}"
|
|
return JSONResponse(content={"url": url})
|
|
|
|
|
|
# ============================================================================
|
|
# App Registration
|
|
# ============================================================================
|
|
|
|
app.include_router(router, prefix="/api")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
|
|
|