feat(api): Implement authentication router with register and login

This commit is contained in:
Yunxiao Xu
2026-02-10 12:25:53 -08:00
parent ab117296e4
commit 979e1ad2d6
3 changed files with 102 additions and 0 deletions

View File

@@ -22,6 +22,7 @@ dependencies = [
"authlib>=1.3.0",
"httpx>=0.27.0",
"argon2-cffi>=23.1.0",
"email-validator>=2.1.0",
"fastapi>=0.109.0",
"uvicorn>=0.27.0",
"python-jose[cryptography]>=3.3.0",

View File

@@ -1,5 +1,6 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from ea_chatbot.api.routers import auth
app = FastAPI(
title="Election Analytics Chatbot API",
@@ -16,6 +17,8 @@ app.add_middleware(
allow_headers=["*"],
)
app.include_router(auth.router)
@app.get("/health")
async def health_check():
return {"status": "ok"}

View File

@@ -0,0 +1,98 @@
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, EmailStr
from typing import Optional
from ea_chatbot.config import Settings
from ea_chatbot.history.manager import HistoryManager
from ea_chatbot.auth import OIDCClient
from ea_chatbot.api.utils import create_access_token
settings = Settings()
history_manager = HistoryManager(settings.history_db_url)
# Initialize OIDC Client if configured
oidc_client = None
if settings.oidc_client_id and settings.oidc_client_secret and settings.oidc_server_metadata_url:
oidc_client = OIDCClient(
client_id=settings.oidc_client_id,
client_secret=settings.oidc_client_secret,
server_metadata_url=settings.oidc_server_metadata_url,
# This will be updated to the frontend URL later
redirect_uri="http://localhost:3000/auth/callback"
)
router = APIRouter(prefix="/auth", tags=["auth"])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
class Token(BaseModel):
access_token: str
token_type: str
class UserCreate(BaseModel):
email: EmailStr
password: str
display_name: Optional[str] = None
class UserResponse(BaseModel):
id: str
email: str
display_name: str
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user_in: UserCreate):
"""Register a new user."""
user = history_manager.get_user(user_in.email)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="User already exists"
)
user = history_manager.create_user(
email=user_in.email,
password=user_in.password,
display_name=user_in.display_name
)
return {
"id": str(user.id),
"email": user.username,
"display_name": user.display_name
}
@router.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
"""Login with email and password to get a JWT."""
user = history_manager.authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(data={"sub": user.username, "user_id": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
@router.get("/oidc/login")
async def oidc_login():
"""Get the OIDC authorization URL."""
if not oidc_client:
raise HTTPException(
status_code=status.HTTP_501_NOT_IMPLEMENTED,
detail="OIDC is not configured"
)
url = oidc_client.get_login_url()
return {"url": url}
@router.get("/me", response_model=UserResponse)
async def get_me(token: str = Depends(oauth2_scheme)):
"""Get the current authenticated user's profile."""
# This currently only validates the token exists via oauth2_scheme
# The next task will implement the full dependency to decode JWT and fetch user
return {
"id": "unknown",
"email": "unknown",
"display_name": "unknown"
}