feat:

--FastAPI added

.gitignore

@@ -1,6 +1,6 @@
 
 venv/
-
+.env
 
 
 # Ignore VSCode settings (optional)

database.py

@@ -0,0 +1,36 @@
+import os
+
+from dotenv import load_dotenv
+from sqlalchemy import create_engine
+from sqlalchemy.orm import DeclarativeBase, sessionmaker
+
+load_dotenv()
+
+DATABASE_URL = os.getenv("DATABASE_URL")
+
+if not DATABASE_URL:
+    raise RuntimeError("DATABASE_URL is not configured")
+
+engine = create_engine(
+    DATABASE_URL,
+    pool_pre_ping=True,
+)
+
+SessionLocal = sessionmaker(
+    bind=engine,
+    autoflush=False,
+    autocommit=False,
+)
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+def get_db():
+    db = SessionLocal()
+
+    try:
+        yield db
+    finally:
+        db.close()

experimental/main.py

@@ -0,0 +1,6 @@
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+ADMIN_API_KEY = os.getenv("ADMIN_API_KEY")

main.py

@@ -0,0 +1,214 @@
+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)

models.py

@@ -0,0 +1,52 @@
+from datetime import datetime
+
+from sqlalchemy import Boolean, DateTime, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from database import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(
+        primary_key=True,
+        index=True,
+    )
+
+    username: Mapped[str] = mapped_column(
+        String(50),
+        unique=True,
+        nullable=False,
+        index=True,
+    )
+
+    email: Mapped[str] = mapped_column(
+        String(255),
+        unique=True,
+        nullable=False,
+        index=True,
+    )
+
+    hashed_password: Mapped[str] = mapped_column(
+        String(255),
+        nullable=False,
+    )
+
+    is_active: Mapped[bool] = mapped_column(
+        Boolean,
+        default=True,
+        nullable=False,
+    )
+
+    is_admin: Mapped[bool] = mapped_column(
+        Boolean,
+        default=False,
+        nullable=False,
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )

requirements.txt

@@ -0,0 +1,7 @@
+fastapi
+uvicorn[standard]
+sqlalchemy
+psycopg[binary]
+python-dotenv
+pwdlib[argon2]
+email-validator

schemas.py

@@ -0,0 +1,31 @@
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, EmailStr, Field
+
+
+class UserCreate(BaseModel):
+    username: str = Field(
+        min_length=3,
+        max_length=50,
+        pattern=r"^[a-zA-Z0-9_.-]+$",
+    )
+
+    email: EmailStr
+
+    password: str = Field(
+        min_length=8,
+        max_length=128,
+    )
+
+    is_admin: bool = False
+
+
+class UserResponse(BaseModel):
+    id: int
+    username: str
+    email: EmailStr
+    is_active: bool
+    is_admin: bool
+    created_at: datetime
+
+    model_config = ConfigDict(from_attributes=True)