feat: implement users API

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

imagecreator/backend/app/api/deps.py

@@ -21,7 +21,7 @@ def get_current_user(
             detail="Invalid or expired token"
         )
 
-    user_id_str = payload.get("sub")
+    user_id_str = payload.sub
     if not user_id_str:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
 

imagecreator/backend/app/api/users.py

@@ -1,3 +1,32 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
 
-router = APIRouter()
+from app.database import get_db
+from app.api.deps import get_current_user
+from app.models.user import User
+from app.schemas.user import UserResponse, UserUpdate
+from app.services.auth import AuthService
+
+router = APIRouter()
+
+@router.get("/me", response_model=UserResponse)
+def get_current_user_info(current_user: User = Depends(get_current_user)):
+    return current_user
+
+@router.put("/me", response_model=UserResponse)
+def update_current_user(
+    data: UserUpdate,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    service = AuthService(db)
+    if data.email and data.email != current_user.email:
+        # Check if email is taken
+        existing = db.query(User).filter(User.email == data.email, User.id != current_user.id).first()
+        if existing:
+            raise HTTPException(status_code=400, detail="Email already registered")
+        current_user.email = data.email
+
+    db.commit()
+    db.refresh(current_user)
+    return current_user

imagecreator/backend/tests/test_users_api.py

@@ -0,0 +1,49 @@
+import pytest
+import os
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from app.database import Base, get_db
+from app.main import app
+from app.models.user import User
+from app.core.security import hash_password, create_access_token
+
+# Use file-based SQLite for testing to avoid in-memory connection issues
+TEST_DB_PATH = "/tmp/test_users.db"
+if os.path.exists(TEST_DB_PATH):
+    os.remove(TEST_DB_PATH)
+
+engine = create_engine(f"sqlite:///{TEST_DB_PATH}", connect_args={"check_same_thread": False})
+TestSessionLocal = sessionmaker(bind=engine)
+
+def override_get_db():
+    db = TestSessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+app.dependency_overrides[get_db] = override_get_db
+client = TestClient(app)
+
+# Create tables once at module level
+Base.metadata.create_all(bind=engine)
+
+def get_auth_header(user_id):
+    token = create_access_token({"sub": str(user_id)})
+    return {"Authorization": f"Bearer {token}"}
+
+def test_get_current_user():
+    db = TestSessionLocal()
+    user = User(username="testuser", email="test@example.com", hashed_password=hash_password("pass"))
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    db.close()
+
+    response = client.get("/api/users/me", headers=get_auth_header(user.id))
+    assert response.status_code == 200
+    data = response.json()
+    assert data["username"] == "testuser"
+    assert data["email"] == "test@example.com"
+    assert "password" not in data