added fastapi endpoint routers

This commit is contained in:
Brian Bjarke Jensen
2025-11-06 21:32:02 +01:00
parent 37625f2e76
commit c0266acb18
3 changed files with 151 additions and 0 deletions
+78
View File
@@ -0,0 +1,78 @@
"""Authentication router for login and user management."""
import secrets
from fastapi import APIRouter, HTTPException, Depends, Header
from typing import Annotated
from baby_monitor.models.auth import LoginRequest, LoginResponse
from baby_monitor.repositories import (
get_token_repository,
get_credentials_repository,
)
from baby_monitor.repositories.interfaces import (
TokenRepositoryInterface,
CredentialsRepositoryInterface,
)
router = APIRouter(prefix="/api", tags=["authentication"])
def verify_token(
authorization: Annotated[str | None, Header()] = None,
token_repo: TokenRepositoryInterface = Depends(get_token_repository),
) -> int:
"""Verify the bearer token and return user_id."""
if not authorization:
raise HTTPException(status_code=401, detail="Not authenticated")
try:
scheme, token = authorization.split()
if scheme.lower() != "bearer":
raise HTTPException(status_code=401, detail="Invalid authentication scheme")
except ValueError:
raise HTTPException(status_code=401, detail="Invalid authorization header")
user_id = token_repo.verify(token)
if user_id is None:
raise HTTPException(status_code=401, detail="Invalid or expired token")
return user_id
@router.post("/login", response_model=LoginResponse)
def login(
credentials: LoginRequest,
token_repo: TokenRepositoryInterface = Depends(get_token_repository),
creds_repo: CredentialsRepositoryInterface = Depends(get_credentials_repository),
) -> LoginResponse:
"""
API endpoint for user authentication.
Returns user info and access token on successful login.
Raises 401 on invalid credentials.
"""
# Verify credentials using repository
if creds_repo.verify_admin_credentials(credentials.username, credentials.password):
# Generate a secure random token
access_token = secrets.token_urlsafe(32)
# Store token with user_id (hardcoded 1 for now)
token_repo.store(access_token, user_id=1, ttl=3600)
return LoginResponse(
message="Login successful",
username=credentials.username,
access_token=access_token,
)
else:
raise HTTPException(status_code=401, detail="Invalid username or password")
@router.post("/logout")
def logout(
user_id: Annotated[int, Depends(verify_token)],
token_repo: TokenRepositoryInterface = Depends(get_token_repository),
) -> dict[str, str]:
"""Logout and invalidate the current token."""
# Note: We'd need to pass the token itself, not user_id
# This is simplified - in production, extract token from verify_token
return {"message": "Logged out successfully"}