wfh_screen_sharing/main.py

import os
from typing import Annotated

from dotenv import load_dotenv
from fastapi import Depends, FastAPI, Header, HTTPException, Response, status
from pwdlib import PasswordHash
from sqlalchemy import or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from database import Base, engine, get_db
from models import User
from schemas import UserCreate, UserResponse

load_dotenv()

app = FastAPI(
    title="FastAPI User Admin",
    description="Simple PostgreSQL-backed user administration API",
    version="1.0.0",
)

password_hash = PasswordHash.recommended()

ADMIN_API_KEY = os.getenv("ADMIN_API_KEY")
print(f"ADMIN_API_KEY: {ADMIN_API_KEY}")
if not ADMIN_API_KEY:
    raise RuntimeError("ADMIN_API_KEY is not configured")


# Create tables when the application starts.
# For larger projects, use Alembic migrations instead.
Base.metadata.create_all(bind=engine)


def verify_admin_key(
    x_admin_key: Annotated[str | None, Header()] = None,
) -> None:
    if x_admin_key != ADMIN_API_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing admin API key",
        )


AdminDependency = Annotated[None, Depends(verify_admin_key)]
DatabaseDependency = Annotated[Session, Depends(get_db)]


@app.get("/")
def root():
    return {
        "message": "FastAPI admin service is running",
        "admin_docs": "/docs",
    }


@app.post(
    "/admin/users",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_user(
    user_data: UserCreate,
    db: DatabaseDependency,
    _: AdminDependency,
):
    existing_user = db.scalar(
        select(User).where(
            or_(
                User.username == user_data.username,
                User.email == user_data.email,
            )
        )
    )

    if existing_user:
        if existing_user.username == user_data.username:
            detail = "Username already exists"
        else:
            detail = "Email address already exists"

        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=detail,
        )

    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=password_hash.hash(user_data.password),
        is_admin=user_data.is_admin,
    )

    db.add(new_user)

    try:
        db.commit()
        db.refresh(new_user)
    except IntegrityError:
        db.rollback()

        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Username or email already exists",
        )

    return new_user


@app.get(
    "/admin/users",
    response_model=list[UserResponse],
)
def list_users(
    db: DatabaseDependency,
    _: AdminDependency,
):
    users = db.scalars(
        select(User).order_by(User.id)
    ).all()

    return list(users)


@app.get(
    "/admin/users/{user_id}",
    response_model=UserResponse,
)
def get_user(
    user_id: int,
    db: DatabaseDependency,
    _: AdminDependency,
):
    user = db.get(User, user_id)

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    return user


@app.patch(
    "/admin/users/{user_id}/disable",
    response_model=UserResponse,
)
def disable_user(
    user_id: int,
    db: DatabaseDependency,
    _: AdminDependency,
):
    user = db.get(User, user_id)

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    user.is_active = False
    db.commit()
    db.refresh(user)

    return user


@app.patch(
    "/admin/users/{user_id}/enable",
    response_model=UserResponse,
)
def enable_user(
    user_id: int,
    db: DatabaseDependency,
    _: AdminDependency,
):
    user = db.get(User, user_id)

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    user.is_active = True
    db.commit()
    db.refresh(user)

    return user


@app.delete(
    "/admin/users/{user_id}",
    status_code=status.HTTP_204_NO_CONTENT,
)
def delete_user(
    user_id: int,
    db: DatabaseDependency,
    _: AdminDependency,
):
    user = db.get(User, user_id)

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    db.delete(user)
    db.commit()

    return Response(status_code=status.HTTP_204_NO_CONTENT)