feat:added leave function

alembic/versions/1eacc17f4c52_add_leaves_table_and_device_tokens_array.py

@@ -0,0 +1,33 @@
+"""add leaves table and device_tokens array
+
+Revision ID: 1eacc17f4c52
+Revises: dd61202db14f
+Create Date: 2025-11-18 22:14:31.077909
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = '1eacc17f4c52'
+down_revision: Union[str, Sequence[str], None] = 'dd61202db14f'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('users', sa.Column('device_tokens', postgresql.ARRAY(sa.String()), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('users', 'device_tokens')
+    # ### end Alembic commands ###

alembic/versions/a3c79664f866_sync_models.py

@@ -0,0 +1,33 @@
+"""sync models
+
+Revision ID: a3c79664f866
+Revises: 1eacc17f4c52
+Create Date: 2025-11-18 22:23:01.757260
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = 'a3c79664f866'
+down_revision: Union[str, Sequence[str], None] = '1eacc17f4c52'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('users', sa.Column('device_tokens', postgresql.ARRAY(sa.String()), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('users', 'device_tokens')
+    # ### end Alembic commands ###

src/core/config.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from pydantic import PostgresDsn, computed_field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -37,6 +38,10 @@ class Settings(BaseSettings):
     GOOGLE_CLIENT_SECRET: str
     GOOGLE_REDIRECT_URI: str
 
+    FCM_SERVER_KEY: Optional[str] = None
+    SICK_LEAVE_LIMIT: int = 10
+    CASUAL_LEAVE_LIMIT: int = 10
+
     AUTH_BASE: str = "https://accounts.google.com/o/oauth2/v2/auth"
     TOKEN_URL: str = "https://oauth2.googleapis.com/token"
     GMAIL_SEND_SCOPE: str = "https://www.googleapis.com/auth/gmail.send"
@@ -57,4 +62,5 @@ class Settings(BaseSettings):
         env_file=".env", case_sensitive=False, env_file_encoding="utf-8"
     )
 
+
 settings = Settings()

src/core/models.py

@@ -1,7 +1,11 @@
+from sqlalchemy.dialects.postgresql import ARRAY
+from sqlalchemy import Column, String
 import uuid
 from datetime import date, datetime
 from enum import Enum
 from typing import List, Optional
+from src.profile.models import Leaves
+
 
 from sqlalchemy import CheckConstraint, UniqueConstraint
 from sqlmodel import Field, Relationship, SQLModel

src/main.py

@@ -13,7 +13,7 @@ app = FastAPI(title="Yuvabe App API")
 
 app.include_router(home_router, prefix="/home", tags=["Home"])
 
-# init_db()
+init_db()
 
 app.include_router(auth_router)
 

src/profile/models.py

@@ -1,2 +1,39 @@
 import uuid
-import sqlmodel
+from datetime import date, datetime
+from typing import Optional
+from sqlmodel import SQLModel, Field
+from enum import Enum
+
+
+class LeaveStatus(str, Enum):
+    PENDING = "PENDING"
+    APPROVED = "APPROVED"
+    REJECTED = "REJECTED"
+
+
+class Leaves(SQLModel, table=True):
+    __tablename__ = "leaves"
+
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+
+    # Foreign keys (users table)
+    user_id: uuid.UUID = Field(foreign_key="users.id", nullable=False)
+    mentor_id: uuid.UUID = Field(foreign_key="users.id", nullable=False)
+    lead_id: uuid.UUID = Field(foreign_key="users.id", nullable=False)
+
+    leave_type: str = Field(nullable=False)
+    from_date: date = Field(nullable=False)
+    to_date: date = Field(nullable=False)
+    days: int = Field(nullable=False)
+    reason: Optional[str] = None
+
+    status: LeaveStatus = Field(default=LeaveStatus.PENDING)
+
+    approved_by: Optional[uuid.UUID] = Field(foreign_key="users.id", default=None)
+    approved_at: Optional[datetime] = None
+    reject_reason: Optional[str] = None
+
+    comment: Optional[str] = None
+
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)

src/profile/router.py

@@ -16,17 +16,113 @@ from src.profile.service import update_user_profile
 from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
 from sqlmodel import select
 from src.core.models import Users, Teams, Roles, UserTeamsRole
+from fastapi import APIRouter, Query, HTTPException
+from fastapi.responses import RedirectResponse, JSONResponse
+import httpx
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlmodel.ext.asyncio.session import AsyncSession
+from src.auth.utils import get_current_user  # adjust path
+from src.profile.schemas import (
+    ApplyLeaveRequest,
+    ApproveRejectRequest,
+    LeaveResponse,
+    BalanceResponse,
+    DeviceTokenIn,
+)
+from src.profile import service
+from typing import List
 
 
 router = APIRouter(prefix="/profile", tags=["Profile"])
 
-# src/routers/gmail_oauth_router.py
-from fastapi import APIRouter, Query, HTTPException
-from fastapi.responses import RedirectResponse, JSONResponse
-import httpx
+
+@router.post("/apply", response_model=LeaveResponse)
+async def apply_leave_endpoint(
+    payload: ApplyLeaveRequest,
+    session: AsyncSession = Depends(get_async_session),
+    current_user=Depends(get_current_user),
+):
+    user_id = current_user
+    leave = await service.apply_leave(session, user_id, payload)
+    return leave
+
+
+@router.get("/pending", response_model=List[LeaveResponse])
+async def pending_leaves(
+    session: AsyncSession = Depends(get_async_session),
+    current_user=Depends(get_current_user),
+):
+    user_id = current_user
+    leaves = await service.get_pending_leaves_for_approver(session, user_id)
+    return leaves
+
+
+@router.get("/my", response_model=List[LeaveResponse])
+async def my_leaves(
+    session: AsyncSession = Depends(get_async_session),
+    current_user=Depends(get_current_user),
+):
+    leaves = await service.get_my_leaves(session, current_user)
+    return leaves
+
+
+@router.get("/team", response_model=List[LeaveResponse])
+async def team_leaves(
+    session: AsyncSession = Depends(get_async_session),
+    current_user=Depends(get_current_user),
+):
+    leaves = await service.get_team_leaves(session, current_user)
+    return leaves
+
+
+@router.post("/{leave_id}/approve", response_model=LeaveResponse)
+async def approve_leave_endpoint(
+    leave_id: str,
+    payload: ApproveRejectRequest,
+    session: AsyncSession = Depends(get_async_session),
+    current_user=Depends(get_current_user),
+):
+    leave = await service.approve_leave(
+        session, current_user, leave_id, comment=payload.comment
+    )
+    return leave
+
+
+@router.post("/{leave_id}/reject", response_model=LeaveResponse)
+async def reject_leave_endpoint(
+    leave_id: str,
+    payload: ApproveRejectRequest,
+    session: AsyncSession = Depends(get_async_session),
+    current_user=Depends(get_current_user),
+):
+    leave = await service.reject_leave(
+        session,
+        current_user,
+        leave_id,
+        reject_reason=payload.reject_reason,
+        comment=payload.comment,
+    )
+    return leave
+
+
+@router.get("/balance", response_model=List[BalanceResponse])
+async def get_balance(
+    session: AsyncSession = Depends(get_async_session),
+    current_user=Depends(get_current_user),
+):
+    return await service.get_leave_balance(session, current_user)
+
+
+@router.post("/device-token")
+async def save_device_token(
+    payload: DeviceTokenIn,
+    session: AsyncSession = Depends(get_async_session),
+    current_user=Depends(get_current_user),
+):
+    tokens = await service.add_device_token(session, current_user, payload.device_token)
+    return {"status": "ok", "tokens": tokens}
 
 
-router = APIRouter(prefix="/gmail", tags=["Gmail OAuth"])
 
 
 @router.get("/login")
@@ -47,7 +143,7 @@ async def google_callback(code: str | None = None, state: str | None = None):
     async with httpx.AsyncClient() as client:
         r = await client.get(
             "https://www.googleapis.com/oauth2/v3/userinfo",
-            headers={"Authorization": f"Bearer {access_token}"}
+            headers={"Authorization": f"Bearer {access_token}"},
         )
     userinfo = r.json()
 
@@ -57,15 +153,17 @@ async def google_callback(code: str | None = None, state: str | None = None):
     USER_TOKEN_STORE[google_user_id] = {
         "access_token": access_token,
         "refresh_token": refresh_token,
-        "email": user_email
+        "email": user_email,
     }
 
-    return JSONResponse({
-        "status": "ok",
-        "user_id": google_user_id,
-        "email": user_email,
-        "state": state,
-    })
+    return JSONResponse(
+        {
+            "status": "ok",
+            "user_id": google_user_id,
+            "email": user_email,
+            "state": state,
+        }
+    )
 
 
 @router.post("/send-mail")
@@ -73,7 +171,6 @@ async def send_mail(req: SendMailRequest):
     return await send_email_service(req)
 
 
-
 @router.get("/", response_model=BaseResponse)
 async def get_assets(
     user_id: str = Depends(get_current_user),

src/profile/schemas.py

@@ -1,7 +1,46 @@
-from pydantic import BaseModel, EmailStr
-from typing import Optional
+from pydantic import BaseModel, EmailStr ,Field
+from typing import Optional,List
 import uuid
 from enum import Enum
+from datetime import date
+
+
+class ApplyLeaveRequest(BaseModel):
+    leave_type: str
+    from_date: date
+    to_date: date
+    reason: Optional[str] = None
+
+class ApproveRejectRequest(BaseModel):
+    comment: Optional[str] = None
+    reject_reason: Optional[str] = None  # used for reject endpoint
+
+class LeaveResponse(BaseModel):
+    id: uuid.UUID
+    user_id: uuid.UUID
+    mentor_id: uuid.UUID
+    lead_id: uuid.UUID
+    leave_type: str
+    from_date: date
+    to_date: date
+    days: int
+    reason: Optional[str]
+    status: str
+    approved_by: Optional[uuid.UUID]
+    approved_at: Optional[str]
+    reject_reason: Optional[str]
+    comment: Optional[str]
+
+class BalanceResponse(BaseModel):
+    leave_type: str
+    limit: int
+    used: int
+    remaining: int
+
+class DeviceTokenIn(BaseModel):
+    device_token: str
+    device_type: Optional[str] = None
+
 
 
 class AssetStatus(str, Enum):

src/profile/service.py

@@ -1,8 +1,6 @@
-from src.profile.utils import build_raw_message
-from src.profile.utils import refresh_access_token
+from src.profile.utils import build_raw_message, refresh_access_token
 from src.profile.schemas import SendMailRequest
-from src.core.models import Assets
-from ast import List
+from src.core.models import Assets, Users
 from datetime import datetime
 import uuid
 from fastapi import HTTPException
@@ -12,13 +10,274 @@ import uuid
 from typing import List
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
-from src.core.models import Assets
 import httpx
 from src.core.config import settings
+from typing import List, Optional
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+from fastapi import HTTPException
+from datetime import datetime
+from src.profile.models import Leaves
+from src.profile.schemas import ApplyLeaveRequest, ApproveRejectRequest
+from src.profile.utils import calculate_days, find_mentor_and_lead
+from src.profile.utils import get_tokens_for_user, send_push_to_tokens
+from src.core.config import settings
 
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
+# src/profile/service.py
+
+# Leave limits (you can move to config)
+SICK_LIMIT = getattr(settings, "SICK_LEAVE_LIMIT", 10)
+CASUAL_LIMIT = getattr(settings, "CASUAL_LEAVE_LIMIT", 10)
+
+
+async def apply_leave(session: AsyncSession, user_id, payload: ApplyLeaveRequest):
+    # compute days
+    days = calculate_days(payload.from_date, payload.to_date)
+    if days <= 0:
+        raise HTTPException(status_code=400, detail="Invalid date range")
+
+    # find mentor and lead
+    mentor, lead = await find_mentor_and_lead(session, user_id)
+    if not mentor or not lead:
+        raise HTTPException(status_code=400, detail="Mentor or Lead not found for user")
+
+    # check remaining balance
+    limit = SICK_LIMIT if payload.leave_type.lower() == "sick" else CASUAL_LIMIT
+    # sum used days for this leave_type
+    q = select(Leaves).where(
+        Leaves.user_id == user_id,
+        Leaves.leave_type.ilike(payload.leave_type),
+        Leaves.status == "APPROVED",
+    )
+    rows = (await session.exec(q)).all()
+    used = sum(r.days for r in rows) if rows else 0
+    remaining = limit - used
+    if days > remaining:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Insufficient {payload.leave_type} balance. Remaining {remaining}",
+        )
+
+    leave = Leaves(
+        user_id=user_id,
+        mentor_id=mentor.id,
+        lead_id=lead.id,
+        leave_type=payload.leave_type,
+        from_date=payload.from_date,
+        to_date=payload.to_date,
+        days=days,
+        reason=payload.reason,
+        status="PENDING",
+    )
+    session.add(leave)
+    await session.commit()
+    await session.refresh(leave)
+
+    # push notifications to mentor & lead
+    title = "New Leave Request"
+    body = f"{user_id} applied {payload.leave_type} leave ({days} days)."
+    mentor_tokens = await get_tokens_for_user(session, mentor.id)
+    lead_tokens = await get_tokens_for_user(session, lead.id)
+    await send_push_to_tokens(
+        mentor_tokens,
+        title,
+        body,
+        data={"leave_id": str(leave.id), "action": "leave_request"},
+    )
+    await send_push_to_tokens(
+        lead_tokens,
+        title,
+        body,
+        data={"leave_id": str(leave.id), "action": "leave_request"},
+    )
+
+    return leave
+
+
+async def get_pending_leaves_for_approver(
+    session: AsyncSession, approver_user_id
+) -> List[Leaves]:
+    # returns pending leaves where mentor_id == approver OR lead_id == approver
+    stmt = select(Leaves).where(
+        (Leaves.mentor_id == approver_user_id) | (Leaves.lead_id == approver_user_id),
+        Leaves.status == "PENDING",
+    )
+    return (await session.exec(stmt)).all()
+
+
+async def get_my_leaves(session: AsyncSession, user_id) -> List[Leaves]:
+    stmt = (
+        select(Leaves)
+        .where(Leaves.user_id == user_id)
+        .order_by(Leaves.created_at.desc())
+    )
+    return (await session.exec(stmt)).all()
+
+
+async def get_team_leaves(session: AsyncSession, lead_user_id) -> List[Leaves]:
+    # lead can view leaves where lead_id == lead_user_id
+    stmt = (
+        select(Leaves)
+        .where(Leaves.lead_id == lead_user_id)
+        .order_by(Leaves.created_at.desc())
+    )
+    return (await session.exec(stmt)).all()
+
+
+async def approve_leave(
+    session: AsyncSession, approver_id, leave_id: str, comment: Optional[str] = None
+):
+    # transaction-safe update
+    async with session.begin():
+        stmt = select(Leaves).where(Leaves.id == leave_id).with_for_update()
+        leave = (await session.exec(stmt)).one_or_none()
+        if not leave:
+            raise HTTPException(404, "Leave not found")
+        if leave.status != "PENDING":
+            raise HTTPException(400, "Leave is not pending")
+
+        # optional: verify approver is mentor or lead for this leave
+        if str(approver_id) not in (str(leave.mentor_id), str(leave.lead_id)):
+            # you might want to check roles more thoroughly
+            raise HTTPException(403, "Not authorized to approve this leave")
+
+        # check balance again before approving
+        # compute limit and used
+        limit = SICK_LIMIT if leave.leave_type.lower() == "sick" else CASUAL_LIMIT
+        q = select(Leaves).where(
+            Leaves.user_id == leave.user_id,
+            Leaves.leave_type.ilike(leave.leave_type),
+            Leaves.status == "APPROVED",
+        )
+        approved_rows = (await session.exec(q)).all()
+        used = sum(r.days for r in approved_rows) if approved_rows else 0
+        if used + leave.days > limit:
+            raise HTTPException(400, "Insufficient balance at approval time")
+
+        # update
+        leave.status = "APPROVED"
+        leave.approved_by = approver_id
+        leave.approved_at = datetime.utcnow()
+        if comment:
+            leave.comment = comment
+
+        session.add(leave)
+    # commit done by context manager
+
+    # send push notification to member and lead
+    title = "Leave Approved"
+    body = f"Your leave ({leave.leave_type}) has been approved."
+    member_tokens = await get_tokens_for_user(session, leave.user_id)
+    lead_tokens = await get_tokens_for_user(session, leave.lead_id)
+    await send_push_to_tokens(
+        member_tokens,
+        title,
+        body,
+        data={"leave_id": str(leave.id), "action": "leave_approved"},
+    )
+    await send_push_to_tokens(
+        lead_tokens,
+        title,
+        body,
+        data={"leave_id": str(leave.id), "action": "leave_approved"},
+    )
+
+    return leave
+
+
+async def reject_leave(
+    session: AsyncSession,
+    approver_id,
+    leave_id: str,
+    reject_reason: Optional[str] = None,
+    comment: Optional[str] = None,
+):
+    async with session.begin():
+        stmt = select(Leaves).where(Leaves.id == leave_id).with_for_update()
+        leave = (await session.exec(stmt)).one_or_none()
+        if not leave:
+            raise HTTPException(404, "Leave not found")
+        if leave.status != "PENDING":
+            raise HTTPException(400, "Leave is not pending")
+
+        if str(approver_id) not in (str(leave.mentor_id), str(leave.lead_id)):
+            raise HTTPException(403, "Not authorized to reject this leave")
+
+        leave.status = "REJECTED"
+        leave.approved_by = approver_id
+        leave.approved_at = datetime.utcnow()
+        leave.reject_reason = reject_reason
+        if comment:
+            leave.comment = comment
+        session.add(leave)
+
+    # push to member + lead
+    title = "Leave Rejected"
+    body = f"Your leave ({leave.leave_type}) has been rejected. Reason: {leave.reject_reason or 'N/A'}"
+    member_tokens = await get_tokens_for_user(session, leave.user_id)
+    lead_tokens = await get_tokens_for_user(session, leave.lead_id)
+    await send_push_to_tokens(
+        member_tokens,
+        title,
+        body,
+        data={"leave_id": str(leave.id), "action": "leave_rejected"},
+    )
+    await send_push_to_tokens(
+        lead_tokens,
+        title,
+        body,
+        data={"leave_id": str(leave.id), "action": "leave_rejected"},
+    )
+
+    return leave
+
+
+async def add_device_token(session: AsyncSession, user_id, device_token: str):
+    """
+    Add FCM token to Users.device_tokens ARRAY.
+    Avoid duplicates.
+    """
+
+    # 1) Fetch user
+    user = await session.get(Users, user_id)
+    if not user:
+        raise HTTPException(404, "User not found")
+
+    # 2) If token not present -> add it
+    if device_token not in user.device_tokens:
+        user.device_tokens.append(device_token)
+        session.add(user)
+        await session.commit()
+        await session.refresh(user)
+
+    return user.device_tokens
+
+
+async def get_leave_balance(session: AsyncSession, user_id) -> List[dict]:
+    # compute used for each leave_type and return
+    # using constants SICK_LIMIT and CASUAL_LIMIT
+    stmt = select(Leaves).where(Leaves.user_id == user_id, Leaves.status == "APPROVED")
+    rows = (await session.exec(stmt)).all()
+    used_sick = sum(r.days for r in rows if r.leave_type.lower() == "sick")
+    used_casual = sum(r.days for r in rows if r.leave_type.lower() == "casual")
+    return [
+        {
+            "leave_type": "Sick",
+            "limit": SICK_LIMIT,
+            "used": used_sick,
+            "remaining": max(0, SICK_LIMIT - used_sick),
+        },
+        {
+            "leave_type": "Casual",
+            "limit": CASUAL_LIMIT,
+            "used": used_casual,
+            "remaining": max(0, CASUAL_LIMIT - used_casual),
+        },
+    ]
+
 
 # In production, replace with DB storage
 USER_TOKEN_STORE = {}  # {google_user_id: {tokens}}

src/profile/utils.py

@@ -8,6 +8,94 @@ import httpx
 from typing import Dict, Optional
 from src.core.config import settings
 from urllib.parse import urlencode
+from datetime import date
+from typing import Tuple, Optional, List
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+from src.core.models import UserTeamsRole, Roles, Users, Teams  # adjust import path if differs
+from src.core.config import settings  # for FCM key if needed
+import httpx
+import math
+
+def calculate_days(from_date: date, to_date: date, include_weekends: bool = True) -> int:
+    """Calculate inclusive days. If you want to exclude weekends, add logic."""
+    delta = (to_date - from_date).days + 1
+    return max(0, delta)
+
+async def find_mentor_and_lead(session: AsyncSession, user_id) -> Tuple[Optional[dict], Optional[dict]]:
+    """
+    Return (mentor_user, lead_user) as dicts or None.
+    Uses your existing UserTeamsRole and Roles tables to find role members in same team.
+    """
+    # 1) find user's team mapping
+    stmt = select(UserTeamsRole).where(UserTeamsRole.user_id == user_id)
+    user_team = (await session.exec(stmt)).first()
+    if not user_team:
+        return None, None
+
+    # 2) find Mentor role id
+    mentor_role = (await session.exec(select(Roles).where(Roles.name == "Mentor"))).first()
+    lead_role = (await session.exec(select(Roles).where(Roles.name == "Team Lead"))).first()
+
+    mentor_user = None
+    lead_user = None
+
+    if mentor_role:
+        mentor_user = (await session.exec(
+            select(Users)
+            .join(UserTeamsRole)
+            .where(UserTeamsRole.team_id == user_team.team_id)
+            .where(UserTeamsRole.role_id == mentor_role.id)
+        )).first()
+
+    if lead_role:
+        lead_user = (await session.exec(
+            select(Users)
+            .join(UserTeamsRole)
+            .where(UserTeamsRole.team_id == user_team.team_id)
+            .where(UserTeamsRole.role_id == lead_role.id)
+        )).first()
+
+    return mentor_user, lead_user
+
+
+
+async def get_tokens_for_user(session: AsyncSession, user_id) -> list[str]:
+    user = await session.get(Users, user_id)
+    if not user:
+        return []
+    return user.device_tokens or []
+
+# Simple FCM send using legacy HTTP API (server key).
+# In production prefer FCM HTTP v1 (OAuth) or firebase-admin SDK.
+async def send_push_to_tokens(tokens: list[str], title: str, body: str, data: dict = None):
+    if not tokens:
+        return
+
+    server_key = getattr(settings, "FCM_SERVER_KEY", None)
+    if not server_key:
+        # no key configured: just log or skip
+        print("FCM_SERVER_KEY not configured, skipping push")
+        return
+
+    url = "https://fcm.googleapis.com/fcm/send"
+    headers = {
+        "Authorization": f"key={server_key}",
+        "Content-Type": "application/json",
+    }
+    payload = {
+        "registration_ids": tokens,
+        "notification": {"title": title, "body": body},
+    }
+    if data:
+        payload["data"] = data
+
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        r = await client.post(url, json=payload, headers=headers)
+        # handle response in logs
+        if r.status_code != 200:
+            print("FCM send failed:", r.status_code, r.text)
+