Python-code moved to backend/ directory.

backend/app/auth.py

@@ -0,0 +1,66 @@
+from datetime import datetime, timedelta
+from typing import Optional
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from jose import jwt, JWTError
+from passlib.context import CryptContext
+from sqlalchemy.orm import Session
+
+from .db import SessionLocal
+from . import models
+from .config import settings
+
+SECRET_KEY = settings.secret_key
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 60
+
+pwd_context = CryptContext(schemes=["bcrypt_sha256"], deprecated="auto")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+def authenticate_user(db: Session, username: str, password: str) -> Optional[models.User]:
+    user = db.query(models.User).filter(models.User.username == username).first()
+    if not user or not verify_password(password, user.password_hash):
+        return None
+    return user
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
+    to_encode.update({"exp": expire})
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+async def get_current_user(
+    token: str = Depends(oauth2_scheme),
+    db: Session = Depends(get_db),
+) -> models.User:
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str | None = payload.get("preferred_username") or payload.get("sub")
+        if username is None:
+            raise credentials_exception
+    except JWTError:
+        raise credentials_exception
+    user = db.query(models.User).filter(models.User.username == username).first()
+    if user is None:
+        raise credentials_exception
+    return user
+

backend/app/config.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+from dotenv import load_dotenv
+import os
+
+# .env laden
+load_dotenv()
+
+class Settings(BaseModel):
+    db_host: str = os.getenv("DB_HOST", "localhost")
+    db_port: int = int(os.getenv("DB_PORT", "5432"))
+    db_name: str = os.getenv("DB_NAME", "server_worklog")
+    db_user: str = os.getenv("DB_USER", "postgres")
+    db_password: str = os.getenv("DB_PASSWORD", "postgres")
+
+    secret_key: str = os.getenv("SECRET_KEY", "CHANGE_ME")
+
+    @property
+    def database_url(self):
+        return (
+            f"postgresql+psycopg://{self.db_user}:{self.db_password}"
+            f"@{self.db_host}:{self.db_port}/{self.db_name}"
+        )
+
+settings = Settings()
+

backend/app/db.py

@@ -0,0 +1,9 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, DeclarativeBase
+from app.config import settings
+
+engine = create_engine(settings.database_url, future=True)
+SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+class Base(DeclarativeBase):
+    pass

backend/app/main.py

@@ -0,0 +1,155 @@
+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
+
+app = FastAPI(title="Server Worklog")
+
+# Nur für lokale Entwicklung, in Produktion via Alembic
+Base.metadata.create_all(bind=engine)
+
+@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
+
+

backend/app/models.py

@@ -0,0 +1,63 @@
+from sqlalchemy import (
+    Column, Integer, String, Text, ForeignKey, DateTime, Table
+)
+from sqlalchemy.orm import relationship, Mapped, mapped_column
+from datetime import datetime
+
+from .db import Base
+
+work_entry_servers = Table(
+    "work_entry_servers",
+    Base.metadata,
+    Column("work_entry_id", ForeignKey("work_entries.id"), primary_key=True),
+    Column("server_id", ForeignKey("servers.id"), primary_key=True),
+)
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
+    password_hash: Mapped[str] = mapped_column(String(255))
+    is_active: Mapped[bool] = mapped_column(default=True)
+
+class Owner(Base):
+    __tablename__ = "owners"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String(100), unique=True)
+    contact: Mapped[str | None] = mapped_column(String(255), nullable=True)
+
+    servers: Mapped[list["Server"]] = relationship("Server", back_populates="owner")
+
+class Server(Base):
+    __tablename__ = "servers"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    hostname: Mapped[str] = mapped_column(String(100), unique=True, index=True)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    owner_id: Mapped[int | None] = mapped_column(ForeignKey("owners.id"))
+
+    owner: Mapped[Owner | None] = relationship("Owner", back_populates="servers")
+    work_entries: Mapped[list["WorkEntry"]] = relationship(
+        "WorkEntry",
+        secondary=work_entry_servers,
+        back_populates="servers",
+    )
+
+class WorkEntry(Base):
+    __tablename__ = "work_entries"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    title: Mapped[str] = mapped_column(String(200))
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
+    duration_minutes: Mapped[int | None] = mapped_column(nullable=True)
+    created_by_user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
+
+    servers: Mapped[list[Server]] = relationship(
+        "Server",
+        secondary=work_entry_servers,
+        back_populates="work_entries",
+    )
+

backend/app/schemas.py

@@ -0,0 +1,56 @@
+from datetime import datetime
+from pydantic import BaseModel
+
+class OwnerBase(BaseModel):
+    name: str
+    contact: str | None = None
+
+class OwnerCreate(OwnerBase):
+    pass
+
+class OwnerRead(OwnerBase):
+    id: int
+    class Config:
+        from_attributes = True
+
+class ServerBase(BaseModel):
+    hostname: str
+    description: str | None = None
+    owner_id: int | None = None
+
+class ServerCreate(ServerBase):
+    pass
+
+class ServerRead(ServerBase):
+    id: int
+    owner: OwnerRead | None = None
+    class Config:
+        from_attributes = True
+
+class WorkEntryBase(BaseModel):
+    title: str
+    description: str | None = None
+    duration_minutes: int | None = None
+
+class WorkEntryCreate(WorkEntryBase):
+    server_ids: list[int] | None = None
+    pass
+
+class WorkEntryRead(WorkEntryBase):
+    id: int
+    created_at: datetime
+    created_by_user_id: int
+    servers: list[ServerRead]
+    class Config:
+        from_attributes = True
+
+class UserCreate(BaseModel):
+    username: str
+    password: str
+
+class UserRead(BaseModel):
+    id: int
+    username: str
+    class Config:
+        from_attributes = True
+