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>
45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
import httpx
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.notification_log import NotificationLog
|
|
from app.models.user import User
|
|
|
|
EXPO_PUSH_URL = "https://exp.host/--/api/v2/push/send"
|
|
|
|
|
|
async def send_push_notification(
|
|
db: AsyncSession,
|
|
user: User,
|
|
title: str,
|
|
body: str,
|
|
notif_type: str,
|
|
registration_id: str | None = None,
|
|
) -> None:
|
|
delivery_status = "skipped"
|
|
|
|
if user.expo_push_token:
|
|
payload = {
|
|
"to": user.expo_push_token,
|
|
"title": title,
|
|
"body": body,
|
|
"data": {"type": notif_type, "registration_id": registration_id},
|
|
"sound": "default",
|
|
}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.post(EXPO_PUSH_URL, json=payload)
|
|
delivery_status = "sent" if response.status_code == 200 else "failed"
|
|
except Exception:
|
|
delivery_status = "failed"
|
|
|
|
log = NotificationLog(
|
|
user_id=user.id,
|
|
registration_id=registration_id,
|
|
type=notif_type,
|
|
title=title,
|
|
body=body,
|
|
delivery_status=delivery_status,
|
|
)
|
|
db.add(log)
|
|
await db.commit()
|