krz/hutch-notify

 1from fastapi import APIRouter, Depends, HTTPException
 2from sqlalchemy import select
 3from sqlalchemy.ext.asyncio import AsyncSession
 4
 5from app.apns import APNSClient
 6from app.db import get_db
 7from app.models import Device
 8from app.security import require_api_key
 9
10router = APIRouter(prefix="/v1/test", tags=["test"], dependencies=[Depends(require_api_key)])
11
12
13@router.post("/push/{device_id}")
14async def test_push(device_id: int, db: AsyncSession = Depends(get_db)) -> dict:
15    result = await db.execute(select(Device).where(Device.id == device_id))
16    device = result.scalar_one_or_none()
17    if device is None:
18        raise HTTPException(status_code=404, detail="device not found")
19
20    client = APNSClient()
21    try:
22        ok, apns_id, error = await client.send_alert(
23            device_token=device.apns_token,
24            apns_env=device.apns_env,
25            topic=device.bundle_id,
26            title="Hutch test",
27            body="Test push from hutch-notify",
28            payload={"event_type": "test", "deep_link": "hutch://home"},
29        )
30    finally:
31        await client.aclose()
32
33    return {"ok": ok, "apns_id": apns_id, "error": error}