feat: implement auth API (register, login, JWT)

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

imagecreator/backend/app/api/__init__.py

@@ -1 +1,4 @@
 """API module."""
+from app.api import auth, users, images, history
+
+__all__ = ["auth", "users", "images", "history"]

imagecreator/backend/app/api/auth.py

@@ -0,0 +1,44 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.schemas.auth import UserRegister, UserLogin, Token, TokenRefresh
+from app.schemas.user import UserResponse
+from app.services.auth import AuthService
+from app.core.security import create_access_token
+
+router = APIRouter()
+
+@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
+def register(data: UserRegister, db: Session = Depends(get_db)):
+    service = AuthService(db)
+    try:
+        user = service.register(data.username, data.email, data.password)
+        return user
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+
+@router.post("/login", response_model=Token)
+def login(data: UserLogin, db: Session = Depends(get_db)):
+    service = AuthService(db)
+    user = service.authenticate(data.username, data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password"
+        )
+
+    access_token = create_access_token(data={"sub": str(user.id)})
+    # For simplicity, using same token as refresh - in production, use separate refresh token
+    return Token(access_token=access_token, refresh_token=access_token)
+
+@router.post("/refresh", response_model=Token)
+def refresh(data: TokenRefresh, db: Session = Depends(get_db)):
+    # Simplified: in production, validate refresh token properly
+    from app.core.security import decode_access_token
+    payload = decode_access_token(data.refresh_token)
+    if not payload:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
+
+    access_token = create_access_token(data={"sub": payload.get("sub")})
+    return Token(access_token=access_token, refresh_token=data.refresh_token)

imagecreator/backend/app/api/deps.py

@@ -1 +1,39 @@
-"""API dependencies placeholder."""
+from typing import Generator
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.core.security import decode_access_token
+from app.services.auth import AuthService
+
+security = HTTPBearer()
+
+def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+    db: Session = Depends(get_db)
+):
+    token = credentials.credentials
+    payload = decode_access_token(token)
+    if not payload:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token"
+        )
+
+    user_id_str = payload.get("sub")
+    if not user_id_str:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
+
+    from uuid import UUID
+    try:
+        user_id = UUID(user_id_str)
+    except ValueError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid user ID")
+
+    service = AuthService(db)
+    user = service.get_user_by_id(user_id)
+    if not user:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
+
+    return user

imagecreator/backend/app/api/history.py

@@ -0,0 +1,3 @@
+from fastapi import APIRouter
+
+router = APIRouter()

imagecreator/backend/app/api/images.py

@@ -0,0 +1,3 @@
+from fastapi import APIRouter
+
+router = APIRouter()

imagecreator/backend/app/api/users.py

@@ -0,0 +1,3 @@
+from fastapi import APIRouter
+
+router = APIRouter()

imagecreator/backend/app/schemas/auth.py

@@ -1 +1,18 @@
-"""Auth schemas placeholder."""
+from pydantic import BaseModel, EmailStr
+
+class UserRegister(BaseModel):
+    username: str
+    email: EmailStr
+    password: str
+
+class UserLogin(BaseModel):
+    username: str
+    password: str
+
+class Token(BaseModel):
+    access_token: str
+    refresh_token: str
+    token_type: str = "bearer"
+
+class TokenRefresh(BaseModel):
+    refresh_token: str

imagecreator/backend/app/schemas/user.py

@@ -1 +1,21 @@
-"""User schemas placeholder."""
+from pydantic import BaseModel, EmailStr
+from typing import Optional
+from datetime import datetime
+from uuid import UUID
+
+class UserBase(BaseModel):
+    username: str
+    email: EmailStr
+
+class UserCreate(UserBase):
+    password: str
+
+class UserResponse(UserBase):
+    id: UUID
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
+
+class UserUpdate(BaseModel):
+    email: Optional[EmailStr] = None

imagecreator/backend/app/services/auth.py

@@ -1 +1,36 @@
-"""Auth service placeholder."""
+from typing import Optional
+from sqlalchemy.orm import Session
+from uuid import UUID
+
+from app.models.user import User
+from app.core.security import hash_password, verify_password
+
+class AuthService:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def register(self, username: str, email: str, password: str) -> User:
+        # Check if user exists
+        existing = self.db.query(User).filter(
+            (User.username == username) | (User.email == email)
+        ).first()
+        if existing:
+            raise ValueError("Username or email already registered")
+
+        hashed_password = hash_password(password)
+        user = User(username=username, email=email, hashed_password=hashed_password)
+        self.db.add(user)
+        self.db.commit()
+        self.db.refresh(user)
+        return user
+
+    def authenticate(self, username: str, password: str) -> Optional[User]:
+        user = self.db.query(User).filter(User.username == username).first()
+        if not user:
+            return None
+        if not verify_password(password, user.hashed_password):
+            return None
+        return user
+
+    def get_user_by_id(self, user_id: UUID) -> Optional[User]:
+        return self.db.query(User).filter(User.id == user_id).first()

imagecreator/backend/tests/test_auth_api.py

@@ -0,0 +1,84 @@
+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
+
+# Use file-based SQLite for testing to avoid in-memory connection issues
+TEST_DB_PATH = "/tmp/test_auth.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 test_register_success():
+    response = client.post("/api/auth/register", json={
+        "username": "newuser",
+        "email": "new@example.com",
+        "password": "password123"
+    })
+    assert response.status_code == 201
+    data = response.json()
+    assert data["username"] == "newuser"
+    assert "id" in data
+
+def test_register_duplicate_username():
+    db = TestSessionLocal()
+    user = User(username="existing", email="existing@example.com", hashed_password=hash_password("pass"))
+    db.add(user)
+    db.commit()
+    db.close()
+
+    response = client.post("/api/auth/register", json={
+        "username": "existing",
+        "email": "new@example.com",
+        "password": "password123"
+    })
+    assert response.status_code == 400
+
+def test_login_success():
+    db = TestSessionLocal()
+    user = User(username="loginuser", email="login@example.com", hashed_password=hash_password("correctpassword"))
+    db.add(user)
+    db.commit()
+    db.close()
+
+    response = client.post("/api/auth/login", json={
+        "username": "loginuser",
+        "password": "correctpassword"
+    })
+    assert response.status_code == 200
+    data = response.json()
+    assert "access_token" in data
+    assert "token_type" in data
+
+def test_login_wrong_password():
+    db = TestSessionLocal()
+    user = User(username="wrongpass", email="wrongpass@example.com", hashed_password=hash_password("correctpassword"))
+    db.add(user)
+    db.commit()
+    db.close()
+
+    response = client.post("/api/auth/login", json={
+        "username": "wrongpass",
+        "password": "wrongpassword"
+    })
+    assert response.status_code == 401