Handle 402 quota_exhausted

import os
from datetime import datetime, timezone
 
import requests
 
 
class QuotaPaused(Exception):
    def __init__(self, limit, resets_at, message):
        super().__init__(message)
        self.limit, self.resets_at = limit, resets_at
 
 
def call(body: dict) -> dict:
    r = requests.post(
        "https://app.syntic.ai/v1/chat/completions",
        headers={"Authorization": f"Bearer {os.environ['SYNTIC_API_KEY']}"},
        json={**body, "stream": False},
        timeout=(10, 300),
    )
    if r.status_code == 402:
        e = r.json()["error"]
        if e.get("limit") == "unavailable":
            raise TimeoutError("usage check unavailable; retry shortly")
        resets = datetime.fromisoformat(e["resets_at"].replace("Z", "+00:00")) if e.get("resets_at") else None
        raise QuotaPaused(e.get("limit"), resets, user_message(e.get("limit"), resets))
    r.raise_for_status()
    return r.json()
 
 
def user_message(limit: str, resets: datetime | None) -> str:
    when = f" It will be available again in about {wait_text(resets)}." if resets else ""
    return {
        "session": "The assistant has reached its usage limit for now." + when,
        "weekly": "The assistant has reached its weekly usage limit." + when,
        "monthly": "The assistant has reached this month's usage limit. An administrator can add credits.",
        "spend_limit": "The assistant has reached its spending limit. An administrator can raise it.",
        "no_credits": "The assistant is out of credits. An administrator can add more.",
    }.get(limit, "The assistant is temporarily unavailable.")
 
 
def wait_text(resets: datetime) -> str:
    minutes = max(1, int((resets - datetime.now(timezone.utc)).total_seconds() // 60))
    return f"{minutes} minutes" if minutes < 120 else f"{minutes // 60} hours"

type QuotaError = { type: string; code: string; message?: string; limit?: string; resets_at?: string | null }
 
export class QuotaPaused extends Error {
  constructor(public limit: string, public resetsAt: Date | null, message: string) { super(message) }
}
 
export async function call(body: object) {
  const res = await fetch('https://app.syntic.ai/v1/chat/completions', {
    method: 'POST',
    headers: { Authorization: `Bearer ${process.env.SYNTIC_API_KEY}`, 'Content-Type': 'application/json' },
    body: JSON.stringify({ ...body, stream: false }),
  })
  if (res.status === 402) {
    const e = ((await res.json()) as { error: QuotaError }).error
    if (e.limit === 'unavailable') throw new Error('usage check unavailable; retry shortly')
    const resetsAt = e.resets_at ? new Date(e.resets_at) : null
    throw new QuotaPaused(e.limit ?? 'unknown', resetsAt, messageFor(e.limit, resetsAt))
  }
  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`)
  return res.json()
}
 
function messageFor(limit: string | undefined, resetsAt: Date | null) {
  const when = resetsAt ? ` It will be available again ${resetsAt.toLocaleString()}.` : ''
  switch (limit) {
    case 'session': return 'The assistant has reached its usage limit for now.' + when
    case 'weekly': return "The assistant has reached this week's usage limit." + when
    case 'monthly': return "The assistant has reached this month's usage limit. An administrator can add credits."
    case 'spend_limit': return 'The assistant has reached its spending limit. An administrator can raise it.'
    case 'no_credits': return 'The assistant is out of credits. An administrator can add more.'
    default: return 'The assistant is temporarily unavailable.'
  }
}