167 lines
4.8 KiB
Python
167 lines
4.8 KiB
Python
from fastapi import FastAPI, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
from fastapi import Query
|
|
|
|
from .db import Base, engine
|
|
from . import models, schemas
|
|
from .auth import (
|
|
get_db, get_current_user, authenticate_user,
|
|
create_access_token, get_password_hash,
|
|
)
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
app = FastAPI(title="Server Worklog")
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # oder deine echte Domain
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Nur für lokale Entwicklung, in Produktion via Alembic
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
@app.options("/{path:path}")
|
|
def preflight_handler(path: str):
|
|
return {}
|
|
|
|
@app.post("/auth/register", response_model=schemas.UserRead)
|
|
def register_user(user_in: schemas.UserCreate, db: Session = Depends(get_db)):
|
|
existing = db.query(models.User).filter(models.User.username == user_in.username).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Username already registered")
|
|
user = models.User(
|
|
username=user_in.username,
|
|
password_hash=get_password_hash(user_in.password),
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
@app.post("/auth/token")
|
|
def login_for_access_token(
|
|
form_data: OAuth2PasswordRequestForm = Depends(),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = authenticate_user(db, form_data.username, form_data.password)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect username or password",
|
|
)
|
|
# OIDC-ähnliche Claims
|
|
access_token = create_access_token(
|
|
data={
|
|
"sub": str(user.id),
|
|
"preferred_username": user.username,
|
|
}
|
|
)
|
|
return {"access_token": access_token, "token_type": "bearer"}
|
|
|
|
|
|
|
|
|
|
# Owner CRUD
|
|
@app.post("/owners", response_model=schemas.OwnerRead)
|
|
def create_owner(
|
|
owner_in: schemas.OwnerCreate,
|
|
db: Session = Depends(get_db),
|
|
user=Depends(get_current_user),
|
|
):
|
|
owner = models.Owner(**owner_in.model_dump())
|
|
db.add(owner)
|
|
db.commit()
|
|
db.refresh(owner)
|
|
return owner
|
|
|
|
@app.get("/owners", response_model=List[schemas.OwnerRead])
|
|
def list_owners(
|
|
db: Session = Depends(get_db),
|
|
user=Depends(get_current_user),
|
|
):
|
|
return db.query(models.Owner).all()
|
|
|
|
# Server CRUD
|
|
@app.post("/servers", response_model=schemas.ServerRead)
|
|
def create_server(
|
|
server_in: schemas.ServerCreate,
|
|
db: Session = Depends(get_db),
|
|
user=Depends(get_current_user),
|
|
):
|
|
server = models.Server(**server_in.model_dump())
|
|
db.add(server)
|
|
db.commit()
|
|
db.refresh(server)
|
|
return server
|
|
|
|
@app.get("/servers", response_model=List[schemas.ServerRead])
|
|
def list_servers(
|
|
db: Session = Depends(get_db),
|
|
user=Depends(get_current_user),
|
|
):
|
|
return db.query(models.Server).all()
|
|
|
|
# WorkEntries
|
|
@app.post("/work-entries", response_model=schemas.WorkEntryRead)
|
|
def create_work_entry(
|
|
entry_in: schemas.WorkEntryCreate,
|
|
db: Session = Depends(get_db),
|
|
user=Depends(get_current_user),
|
|
):
|
|
servers = db.query(models.Server).filter(
|
|
models.Server.id.in_(entry_in.server_ids)
|
|
).all()
|
|
if len(servers) != len(entry_in.server_ids):
|
|
raise HTTPException(status_code=400, detail="One or more servers not found")
|
|
|
|
entry = models.WorkEntry(
|
|
title=entry_in.title,
|
|
description=entry_in.description,
|
|
duration_minutes=entry_in.duration_minutes,
|
|
created_by_user_id=user.id,
|
|
servers=servers,
|
|
)
|
|
db.add(entry)
|
|
db.commit()
|
|
db.refresh(entry)
|
|
return entry
|
|
|
|
@app.get("/work-entries", response_model=List[schemas.WorkEntryRead])
|
|
def search_work_entries(
|
|
q: str | None = Query(default=None, description="Freitextsuche in Titel/Beschreibung"),
|
|
server_id: int | None = None,
|
|
db: Session = Depends(get_db),
|
|
user=Depends(get_current_user),
|
|
):
|
|
query = db.query(models.WorkEntry)
|
|
if q:
|
|
like = f"%{q}%"
|
|
query = query.filter(
|
|
(models.WorkEntry.title.ilike(like)) |
|
|
(models.WorkEntry.description.ilike(like))
|
|
)
|
|
if server_id:
|
|
query = query.join(models.WorkEntry.servers).filter(models.Server.id == server_id)
|
|
query = query.order_by(models.WorkEntry.created_at.desc())
|
|
return query.all()
|
|
|
|
@app.get("/servers/{server_id}/work-entries", response_model=List[schemas.WorkEntryRead])
|
|
def list_work_entries_for_server(
|
|
server_id: int,
|
|
db: Session = Depends(get_db),
|
|
user=Depends(get_current_user),
|
|
):
|
|
entries = (
|
|
db.query(models.WorkEntry)
|
|
.join(models.WorkEntry.servers)
|
|
.filter(models.Server.id == server_id)
|
|
.order_by(models.WorkEntry.created_at.asc())
|
|
.all()
|
|
)
|
|
return entries
|
|
|
|
|