1c5719ac85
Full-stack mobile app for pole dance championship management. Backend: FastAPI + SQLAlchemy 2 (async) + SQLite (dev) / PostgreSQL (prod) - JWT auth with refresh token rotation - Championship CRUD with Instagram Graph API sync (APScheduler) - Registration flow with status management - Participant list publish with Expo push notifications - Alembic migrations, pytest test suite Mobile: React Native + Expo (TypeScript) - Auth gate: pending approval screen for new members - Championships list & detail screens - Registration form with status tracking - React Query + Zustand + React Navigation v6 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
import jwt
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from app.crud import crud_user
|
|
from app.database import AsyncSession, get_db
|
|
from app.models.user import User
|
|
from app.services.auth_service import decode_access_token
|
|
|
|
bearer_scheme = HTTPBearer()
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> User:
|
|
try:
|
|
payload = decode_access_token(credentials.credentials)
|
|
except jwt.InvalidTokenError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token"
|
|
)
|
|
user = await crud_user.get(db, payload["sub"])
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
return user
|
|
|
|
|
|
async def get_approved_user(user: User = Depends(get_current_user)) -> User:
|
|
if user.status != "approved":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Account pending approval",
|
|
)
|
|
return user
|
|
|
|
|
|
async def get_organizer(user: User = Depends(get_approved_user)) -> User:
|
|
if user.role not in ("organizer", "admin"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Organizer access required",
|
|
)
|
|
return user
|
|
|
|
|
|
async def get_admin(user: User = Depends(get_approved_user)) -> User:
|
|
if user.role != "admin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin access required",
|
|
)
|
|
return user
|