SDK Python
Exemplos completos com requests e httpx.
Cliente reutilizavel (requests)
import time
import requests
from decimal import Decimal
from typing import Any
class FastDelivClient:
"""Cliente HTTP para a API fast_deliv."""
BASE_URL = "https://fast-deliv-backend.vercel.app"
SUPABASE_URL = "https://<PROJECT>.supabase.co"
SUPABASE_ANON_KEY = "<ANON_KEY>"
def __init__(self, email: str, password: str):
self.email = email
self.password = password
self._token: str | None = None
self._refresh_token: str | None = None
self._token_expires_at: float = 0
# --- Autenticacao ---
def _authenticate(self) -> None:
resp = requests.post(
f"{self.SUPABASE_URL}/auth/v1/token?grant_type=password",
headers={"apikey": self.SUPABASE_ANON_KEY, "Content-Type": "application/json"},
json={"email": self.email, "password": self.password},
)
resp.raise_for_status()
data = resp.json()
self._token = data["access_token"]
self._refresh_token = data["refresh_token"]
self._token_expires_at = time.time() + data.get("expires_in", 3600) - 60
def _refresh(self) -> None:
resp = requests.post(
f"{self.SUPABASE_URL}/auth/v1/token?grant_type=refresh_token",
headers={"apikey": self.SUPABASE_ANON_KEY, "Content-Type": "application/json"},
json={"refresh_token": self._refresh_token},
)
resp.raise_for_status()
data = resp.json()
self._token = data["access_token"]
self._token_expires_at = time.time() + data.get("expires_in", 3600) - 60
@property
def headers(self) -> dict[str, str]:
if self._token is None:
self._authenticate()
elif time.time() >= self._token_expires_at:
self._refresh()
return {"Authorization": f"Bearer {self._token}"}
# --- Requests genericos ---
def get(self, path: str, **kwargs) -> Any:
resp = requests.get(f"{self.BASE_URL}{path}", headers=self.headers, **kwargs)
resp.raise_for_status()
return resp.json()
def post(self, path: str, json: dict, **kwargs) -> Any:
resp = requests.post(
f"{self.BASE_URL}{path}", headers=self.headers, json=json, **kwargs
)
resp.raise_for_status()
return resp.json()
def patch(self, path: str, **kwargs) -> Any:
resp = requests.patch(f"{self.BASE_URL}{path}", headers=self.headers, **kwargs)
resp.raise_for_status()
return resp.json()
def put(self, path: str, json: dict, **kwargs) -> Any:
resp = requests.put(
f"{self.BASE_URL}{path}", headers=self.headers, json=json, **kwargs
)
resp.raise_for_status()
return resp.json()
# --- Entregas ---
def list_deliveries(self, status: str | None = None) -> list[dict]:
params = {"status": status} if status else {}
return self.get("/api/v1/deliveries", params=params)
def create_delivery(
self,
title: str,
origin_address: str,
origin_lat: float,
origin_lng: float,
destination_address: str,
destination_lat: float,
destination_lng: float,
) -> dict:
return self.post(
"/api/v1/deliveries",
json={
"title": title,
"origin_address": origin_address,
"origin_lat": origin_lat,
"origin_lng": origin_lng,
"destination_address": destination_address,
"destination_lat": destination_lat,
"destination_lng": destination_lng,
},
)
def cancel_delivery(self, delivery_id: str) -> dict:
return self.patch(f"/api/v1/deliveries/{delivery_id}/cancel")
# --- Motoristas ---
def list_drivers(self) -> list[dict]:
return self.get("/api/v1/drivers")
# --- Carteira ---
def get_wallet(self) -> dict:
return self.get("/api/v1/wallets/me")
def get_transactions(self) -> list[dict]:
return self.get("/api/v1/wallets/me/transactions")
def withdraw(self, amount: str, pix_key: str, pix_key_type: str) -> dict:
return self.post(
"/api/v1/wallets/withdraw",
json={"amount": amount, "pix_key": pix_key, "pix_key_type": pix_key_type},
)
# --- Precos ---
def get_pricing(self) -> dict:
return self.get("/api/v1/pricing")
def update_pricing(self, **kwargs) -> dict:
return self.put("/api/v1/pricing", json=kwargs)
Uso basico
client = FastDelivClient("admin@empresa.com", "senha123")
# Listar entregas pendentes
pending = client.list_deliveries(status="pending")
print(f"{len(pending)} entregas pendentes")
# Criar entrega
delivery = client.create_delivery(
title="Entrega Setor Sul",
origin_address="Rua das Flores, 100, Goiania",
origin_lat=-16.6869,
origin_lng=-49.2648,
destination_address="Av. T-63, 800, Goiania",
destination_lat=-16.7012,
destination_lng=-49.2789,
)
print(f"Criada: {delivery['id']}")
Polling de status de entrega
import time
def wait_for_completion(client: FastDelivClient, delivery_id: str, timeout: int = 3600) -> dict:
"""Aguarda uma entrega ser concluida ou cancelada."""
deadline = time.time() + timeout
terminal_states = {"completed", "cancelled"}
while time.time() < deadline:
deliveries = client.list_deliveries()
delivery = next((d for d in deliveries if d["id"] == delivery_id), None)
if delivery is None:
raise ValueError(f"Entrega {delivery_id} nao encontrada")
status = delivery["status"]
print(f"Status atual: {status}")
if status in terminal_states:
return delivery
time.sleep(30) # verificar a cada 30 segundos
raise TimeoutError(f"Entrega {delivery_id} nao concluiu em {timeout}s")
# Uso
delivery = client.create_delivery(...)
resultado = wait_for_completion(client, delivery["id"])
print(f"Concluida! Ganho: R$ {resultado['driver_earnings']}")
Listener de webhook com FastAPI
import hashlib
import hmac
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
ASAAS_WEBHOOK_SECRET = "seu_webhook_secret"
@app.post("/webhooks/asaas")
async def handle_asaas_webhook(request: Request):
body = await request.body()
signature = request.headers.get("asaas-signature", "")
expected = hmac.new(
ASAAS_WEBHOOK_SECRET.encode(),
body,
hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(expected, signature):
raise HTTPException(status_code=401, detail="Assinatura invalida")
event = await request.json()
event_type = event.get("event")
transfer_id = event.get("payment", {}).get("id")
if event_type in ("TRANSFER_DONE", "TRANSFER_APPROVED"):
print(f"Saque {transfer_id} concluido!")
# Atualize seu sistema aqui
elif event_type in ("TRANSFER_FAILED", "TRANSFER_CANCELLED"):
reason = event.get("payment", {}).get("failReason", "motivo desconhecido")
print(f"Saque {transfer_id} falhou: {reason}")
# Notifique o motorista ou registre o evento
return {"received": True}
Versao async com httpx
import httpx
import asyncio
async def listar_entregas_async(token: str) -> list[dict]:
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://fast-deliv-backend.vercel.app/api/v1/deliveries",
headers={"Authorization": f"Bearer {token}"},
)
resp.raise_for_status()
return resp.json()
# Buscar varios recursos em paralelo
async def get_dashboard_data(token: str) -> dict:
headers = {"Authorization": f"Bearer {token}"}
async with httpx.AsyncClient() as client:
deliveries_task = client.get(
"https://fast-deliv-backend.vercel.app/api/v1/deliveries",
headers=headers,
)
drivers_task = client.get(
"https://fast-deliv-backend.vercel.app/api/v1/drivers",
headers=headers,
)
pricing_task = client.get(
"https://fast-deliv-backend.vercel.app/api/v1/pricing",
headers=headers,
)
deliveries_resp, drivers_resp, pricing_resp = await asyncio.gather(
deliveries_task, drivers_task, pricing_task
)
return {
"deliveries": deliveries_resp.json(),
"drivers": drivers_resp.json(),
"pricing": pricing_resp.json(),
}