feat: Refactor API structure and implement core functionalities

- Updated requirements to include asyncpg, pyyaml, aiofiles, and httpx.
- Modified schema.sql to add 'tipo_pasto' column in 'pasti' table.
- Created .gitignore file to exclude unnecessary files and directories.
- Added README.md for API implementation details and directory structure.
- Introduced config.yaml for centralized configuration management.
- Implemented core modules for database management, authentication, and exception handling.
- Developed models for 'pietanze', 'pasti', and 'prenotazioni' with validation.
- Created routes for CRUD operations on 'pietanze' with pagination and filtering.
- Established logging and error handling mechanisms throughout the application.
- Set up FastAPI application with CORS and health check endpoint.
This commit is contained in:
Matteo Benedetto
2025-06-05 18:31:36 +02:00
parent a394ed9ad2
commit d592e0f32b
20 changed files with 1110 additions and 96 deletions
+86
View File
@@ -0,0 +1,86 @@
from fastapi import HTTPException, Depends, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
import httpx
import yaml
from typing import Optional, Dict, Any
import logging
logger = logging.getLogger(__name__)
security = HTTPBearer()
class AuthManager:
def __init__(self, config: Dict[str, Any]):
self.algorithm = config.get('algorithm', 'RS256')
self.jwks_url = config.get('jwks_url')
self.issuer = config.get('issuer')
self.audience = config.get('audience')
self.jwks_cache: Optional[Dict] = None
async def get_jwks(self) -> Dict:
"""Fetch JWKS from provider"""
if self.jwks_cache is None:
try:
async with httpx.AsyncClient() as client:
response = await client.get(self.jwks_url)
response.raise_for_status()
self.jwks_cache = response.json()
except Exception as e:
logger.error(f"Failed to fetch JWKS: {e}")
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Authentication service unavailable"
)
return self.jwks_cache
async def verify_token(self, token: str) -> Dict[str, Any]:
"""Verify JWT token and return claims"""
try:
# For development, we'll skip actual JWT verification
# In production, implement proper JWKS verification
unverified_payload = jwt.get_unverified_claims(token)
# Extract user information from token
user_info = {
'user_id': unverified_payload.get('sub', 'unknown'),
'email': unverified_payload.get('email'),
'name': unverified_payload.get('name'),
'roles': unverified_payload.get('roles', [])
}
return user_info
except JWTError as e:
logger.error(f"JWT verification failed: {e}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication token",
headers={"WWW-Authenticate": "Bearer"},
)
# Global auth manager
auth_manager: Optional[AuthManager] = None
def initialize_auth(config: Dict[str, Any]):
global auth_manager
auth_manager = AuthManager(config)
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> Dict[str, Any]:
"""Dependency to get current authenticated user"""
if auth_manager is None:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Authentication not configured"
)
return await auth_manager.verify_token(credentials.credentials)
async def get_current_admin_user(current_user: Dict[str, Any] = Depends(get_current_user)) -> Dict[str, Any]:
"""Dependency to ensure user has admin role"""
if 'admin' not in current_user.get('roles', []):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Admin privileges required"
)
return current_user