""" Cakto API Python Client ======================== Cliente Python para integração com a API da Cakto. Plataforma de pagamentos para negócios digitais. Uso Síncrono: from cakto import CaktoClient client = CaktoClient(client_id="xxx", client_secret="yyy") products = client.products.list() Uso Assíncrono: from cakto import AsyncCaktoClient async with AsyncCaktoClient(client_id="xxx", client_secret="yyy") as client: products = await client.products.list() Autor: Claude Code Documentação: https://docs.cakto.com.br """ from __future__ import annotations import requests from typing import Optional, Dict, List, Any, Iterator, AsyncIterator, TYPE_CHECKING from dataclasses import dataclass from datetime import datetime, timedelta from urllib.parse import urljoin, urlencode import time import asyncio # Importação condicional do httpx try: import httpx HTTPX_AVAILABLE = True except ImportError: HTTPX_AVAILABLE = False httpx = None # type: ignore # ============================================================================= # Exceptions # ============================================================================= class CaktoError(Exception): """Erro base da API Cakto.""" pass class CaktoAuthError(CaktoError): """Erro de autenticação.""" pass class CaktoNotFoundError(CaktoError): """Recurso não encontrado.""" pass class CaktoValidationError(CaktoError): """Erro de validação dos dados.""" pass class CaktoRateLimitError(CaktoError): """Rate limit excedido.""" pass # ============================================================================= # Data Classes # ============================================================================= @dataclass class PaginatedResponse: """Resposta paginada da API.""" count: int next: Optional[str] previous: Optional[str] results: List[Dict[str, Any]] @property def has_next(self) -> bool: return self.next is not None @property def has_previous(self) -> bool: return self.previous is not None @dataclass class Token: """Token de acesso OAuth2.""" access_token: str token_type: str expires_in: int created_at: datetime @property def is_expired(self) -> bool: """Verifica se o token expirou (com margem de 60s).""" expiry = self.created_at + timedelta(seconds=self.expires_in - 60) return datetime.now() >= expiry # ============================================================================= # Base Resource # ============================================================================= class BaseResource: """Classe base para recursos da API.""" def __init__(self, client: 'CaktoClient'): self._client = client def _request( self, method: str, endpoint: str, params: Optional[Dict] = None, data: Optional[Dict] = None, **kwargs ) -> Any: return self._client._request(method, endpoint, params, data, **kwargs) def _paginate( self, endpoint: str, params: Optional[Dict] = None, limit: int = 50 ) -> Iterator[Dict[str, Any]]: """Itera sobre todos os resultados paginados.""" params = params or {} params['limit'] = limit page = 1 while True: params['page'] = page response = self._request('GET', endpoint, params=params) paginated = PaginatedResponse(**response) for item in paginated.results: yield item if not paginated.has_next: break page += 1 # ============================================================================= # Products Resource # ============================================================================= class ProductsResource(BaseResource): """ Gerenciamento de Produtos. Exemplo: # Listar todos os produtos products = client.products.list() # Filtrar por status active = client.products.list(status='active') # Obter produto específico product = client.products.get('product_id') # Atualizar produto client.products.update('product_id', name='Novo Nome') """ ENDPOINT = '/public_api/products/' def list( self, *, page: int = 1, limit: int = 50, search: Optional[str] = None, status: Optional[str] = None, type: Optional[str] = None, ordering: Optional[str] = None, **filters ) -> PaginatedResponse: """ Lista produtos com filtros. Args: page: Número da página limit: Resultados por página search: Busca por nome, id, short_id status: active, blocked, deleted type: unique, subscription ordering: Ordenação (name, price, -createdAt, etc) **filters: Filtros adicionais (price__gte, createdAt__lte, etc) Returns: PaginatedResponse com lista de produtos """ params = {'page': page, 'limit': limit} if search: params['search'] = search if status: params['status'] = status if type: params['type'] = type if ordering: params['ordering'] = ordering params.update(filters) response = self._request('GET', self.ENDPOINT, params=params) return PaginatedResponse(**response) def list_all(self, **filters) -> Iterator[Dict[str, Any]]: """Itera sobre todos os produtos (todas as páginas).""" return self._paginate(self.ENDPOINT, params=filters) def get(self, product_id: str) -> Dict[str, Any]: """Obtém um produto pelo ID.""" return self._request('GET', f'{self.ENDPOINT}{product_id}/') def update(self, product_id: str, **data) -> Dict[str, Any]: """ Atualiza um produto. Args: product_id: ID do produto **data: Campos a atualizar (name, description, price, etc) Returns: Produto atualizado """ return self._request('PUT', f'{self.ENDPOINT}{product_id}/', data=data) # ============================================================================= # Offers Resource # ============================================================================= class OffersResource(BaseResource): """ Gerenciamento de Ofertas. Exemplo: # Listar ofertas offers = client.offers.list() # Criar oferta offer = client.offers.create( name='Oferta Especial', price=97.00, product='product_uuid', type='unique', intervalType='lifetime' ) # Atualizar oferta client.offers.update('offer_id', price=147.00) # Deletar oferta client.offers.delete('offer_id') """ ENDPOINT = '/public_api/offers/' def list( self, *, page: int = 1, limit: int = 50, search: Optional[str] = None, status: Optional[str] = None, type: Optional[str] = None, product: Optional[str] = None, ordering: Optional[str] = None, **filters ) -> PaginatedResponse: """ Lista ofertas com filtros. Args: page: Número da página limit: Resultados por página search: Busca por nome, id, product.name status: active, disabled, deleted type: unique, subscription product: ID do produto (comma-separated para múltiplos) ordering: Ordenação (name, price, -createdAt, etc) **filters: Filtros adicionais Returns: PaginatedResponse com lista de ofertas """ params = {'page': page, 'limit': limit} if search: params['search'] = search if status: params['status'] = status if type: params['type'] = type if product: params['product'] = product if ordering: params['ordering'] = ordering params.update(filters) response = self._request('GET', self.ENDPOINT, params=params) return PaginatedResponse(**response) def list_all(self, **filters) -> Iterator[Dict[str, Any]]: """Itera sobre todas as ofertas (todas as páginas).""" return self._paginate(self.ENDPOINT, params=filters) def get(self, offer_id: str) -> Dict[str, Any]: """Obtém uma oferta pelo ID.""" return self._request('GET', f'{self.ENDPOINT}{offer_id}/') def create( self, name: str, price: float, product: str, type: str = 'unique', *, units: int = 1, default: bool = False, # Campos para tipo 'unique' intervalType: Optional[str] = None, interval: Optional[int] = None, # Campos para tipo 'subscription' recurrence_period: Optional[int] = None, quantity_recurrences: Optional[int] = None, trial_days: Optional[int] = None, max_retries: Optional[int] = None, retry_interval: Optional[int] = None, **extra ) -> Dict[str, Any]: """ Cria uma nova oferta. Args: name: Nome da oferta price: Preço product: UUID do produto type: 'unique' ou 'subscription' units: Unidades por compra default: Oferta padrão intervalType: Para unique - week, month, year, lifetime interval: Para unique - número de intervalos recurrence_period: Para subscription - dias entre cobranças quantity_recurrences: Para subscription - número de cobranças (-1 = infinito) trial_days: Para subscription - dias de trial max_retries: Para subscription - tentativas de cobrança retry_interval: Para subscription - dias entre tentativas Returns: Oferta criada """ data = { 'name': name, 'price': price, 'product': product, 'type': type, 'units': units, 'default': default, } if type == 'unique': if intervalType: data['intervalType'] = intervalType if interval: data['interval'] = interval elif type == 'subscription': if recurrence_period: data['recurrence_period'] = recurrence_period if quantity_recurrences is not None: data['quantity_recurrences'] = quantity_recurrences if trial_days is not None: data['trial_days'] = trial_days if max_retries is not None: data['max_retries'] = max_retries if retry_interval is not None: data['retry_interval'] = retry_interval data.update(extra) return self._request('POST', self.ENDPOINT, data=data) def update(self, offer_id: str, **data) -> Dict[str, Any]: """ Atualiza uma oferta. Args: offer_id: ID da oferta **data: Campos a atualizar Returns: Oferta atualizada """ return self._request('PUT', f'{self.ENDPOINT}{offer_id}/', data=data) def delete(self, offer_id: str) -> None: """Deleta uma oferta.""" self._request('DELETE', f'{self.ENDPOINT}{offer_id}/') # ============================================================================= # Orders Resource # ============================================================================= class OrdersResource(BaseResource): """ Gerenciamento de Pedidos. Exemplo: # Listar pedidos pagos orders = client.orders.list(status='paid') # Filtrar por data orders = client.orders.list( paidAt__gte='2025-01-01', paidAt__lte='2025-01-31' ) # Obter pedido order = client.orders.get('order_id') # Reembolsar client.orders.refund('order_id') # Reenviar email de aprovação client.orders.resend_email('order_id') """ ENDPOINT = '/public_api/orders/' # Status possíveis STATUS_PAID = 'paid' STATUS_REFUNDED = 'refunded' STATUS_WAITING = 'waiting_payment' STATUS_REFUSED = 'refused' STATUS_BLOCKED = 'blocked' STATUS_CHARGEDBACK = 'chargedback' # Tipos de oferta OFFER_TYPE_MAIN = 'main' OFFER_TYPE_UPSELL = 'upsell' OFFER_TYPE_DOWNSELL = 'downsell' OFFER_TYPE_ORDERBUMP = 'orderbump' def list( self, *, page: int = 1, limit: int = 50, search: Optional[str] = None, status: Optional[str] = None, type: Optional[str] = None, offer_type: Optional[str] = None, product: Optional[str] = None, customer: Optional[str] = None, affiliate: Optional[str] = None, paymentMethod: Optional[str] = None, ordering: Optional[str] = None, **filters ) -> PaginatedResponse: """ Lista pedidos com filtros avançados. Args: page: Número da página limit: Resultados por página search: Busca textual status: paid, refunded, waiting_payment, refused, blocked, chargedback type: unique, subscription offer_type: main, upsell, downsell, orderbump product: ID do produto customer: ID do cliente affiliate: ID do afiliado paymentMethod: Método de pagamento ordering: Ordenação (paidAt, createdAt, amount, -refId) **filters: Filtros adicionais: - amount__gte, amount__lte: Filtro por valor - paidAt__gte, paidAt__lte: Filtro por data de pagamento - createdAt__gte, createdAt__lte: Filtro por data de criação - utm_source, utm_medium, utm_campaign: Filtros UTM Returns: PaginatedResponse com lista de pedidos """ params = {'page': page, 'limit': limit} if search: params['search'] = search if status: params['status'] = status if type: params['type'] = type if offer_type: params['offer_type'] = offer_type if product: params['product'] = product if customer: params['customer'] = customer if affiliate: params['affiliate'] = affiliate if paymentMethod: params['paymentMethod'] = paymentMethod if ordering: params['ordering'] = ordering params.update(filters) response = self._request('GET', self.ENDPOINT, params=params) return PaginatedResponse(**response) def list_all(self, **filters) -> Iterator[Dict[str, Any]]: """Itera sobre todos os pedidos (todas as páginas).""" return self._paginate(self.ENDPOINT, params=filters) def get(self, order_id: str) -> Dict[str, Any]: """Obtém um pedido pelo ID.""" return self._request('GET', f'{self.ENDPOINT}{order_id}/') def refund(self, order_id: str) -> Dict[str, Any]: """ Solicita reembolso de um pedido. Args: order_id: ID do pedido Returns: Resposta da API """ return self._request('POST', f'{self.ENDPOINT}{order_id}/refund/') def resend_email(self, order_id: str) -> Dict[str, Any]: """ Reenvia o email de confirmação de pagamento. Args: order_id: ID do pedido Returns: Resposta da API """ return self._request('POST', f'{self.ENDPOINT}{order_id}/resend_approved_email/') def resend_access(self, order_id: str) -> Dict[str, Any]: """ Cria ou atualiza acessos de um pedido. Args: order_id: ID do pedido Returns: Resposta da API """ return self._request('POST', f'{self.ENDPOINT}{order_id}/resend_access/') # ============================================================================= # Webhooks Resource # ============================================================================= class WebhooksResource(BaseResource): """ Gerenciamento de Webhooks. Exemplo: # Listar webhooks webhooks = client.webhooks.list() # Criar webhook webhook = client.webhooks.create( name='Meu Webhook', url='https://meusite.com/webhook', events=['purchase_approved', 'refund'] ) # Testar webhook client.webhooks.test('webhook_id') # Ver histórico de eventos history = client.webhooks.event_history() """ ENDPOINT = '/public_api/webhook/' # Eventos disponíveis EVENT_INITIATE_CHECKOUT = 'initiate_checkout' EVENT_CHECKOUT_ABANDONMENT = 'checkout_abandonment' EVENT_PURCHASE_APPROVED = 'purchase_approved' EVENT_PURCHASE_REFUSED = 'purchase_refused' EVENT_PIX_GERADO = 'pix_gerado' EVENT_BOLETO_GERADO = 'boleto_gerado' EVENT_PICPAY_GERADO = 'picpay_gerado' EVENT_OPENFINANCE_NUBANK = 'openfinance_nubank_gerado' EVENT_CHARGEBACK = 'chargeback' EVENT_REFUND = 'refund' EVENT_SUBSCRIPTION_CREATED = 'subscription_created' EVENT_SUBSCRIPTION_CANCELED = 'subscription_canceled' EVENT_SUBSCRIPTION_RENEWED = 'subscription_renewed' EVENT_SUBSCRIPTION_RENEWAL_REFUSED = 'subscription_renewal_refused' ALL_EVENTS = [ EVENT_INITIATE_CHECKOUT, EVENT_CHECKOUT_ABANDONMENT, EVENT_PURCHASE_APPROVED, EVENT_PURCHASE_REFUSED, EVENT_PIX_GERADO, EVENT_BOLETO_GERADO, EVENT_PICPAY_GERADO, EVENT_OPENFINANCE_NUBANK, EVENT_CHARGEBACK, EVENT_REFUND, EVENT_SUBSCRIPTION_CREATED, EVENT_SUBSCRIPTION_CANCELED, EVENT_SUBSCRIPTION_RENEWED, EVENT_SUBSCRIPTION_RENEWAL_REFUSED, ] def list( self, *, page: int = 1, limit: int = 50, search: Optional[str] = None, status: Optional[str] = None, events: Optional[List[str]] = None, products: Optional[List[str]] = None, ordering: Optional[str] = None, **filters ) -> PaginatedResponse: """ Lista webhooks com filtros. Args: page: Número da página limit: Resultados por página search: Busca por nome ou URL status: active, disabled, waiting_config events: Lista de eventos para filtrar products: Lista de IDs de produtos ordering: Ordenação (status, name, url, -createdAt) Returns: PaginatedResponse com lista de webhooks """ params = {'page': page, 'limit': limit} if search: params['search'] = search if status: params['status'] = status if events: params['events'] = ','.join(events) if products: params['products'] = ','.join(products) if ordering: params['ordering'] = ordering params.update(filters) response = self._request('GET', self.ENDPOINT, params=params) return PaginatedResponse(**response) def list_all(self, **filters) -> Iterator[Dict[str, Any]]: """Itera sobre todos os webhooks (todas as páginas).""" return self._paginate(self.ENDPOINT, params=filters) def get(self, webhook_id: str) -> Dict[str, Any]: """Obtém um webhook pelo ID.""" return self._request('GET', f'{self.ENDPOINT}{webhook_id}/') def create( self, name: str, url: str, events: List[str], *, products: Optional[List[str]] = None, status: str = 'active', **extra ) -> Dict[str, Any]: """ Cria um novo webhook. Args: name: Nome do webhook url: URL de destino (max 2048 caracteres) events: Lista de eventos para receber products: Lista de IDs de produtos (opcional) status: active, disabled Returns: Webhook criado """ data = { 'name': name, 'url': url, 'events': events, 'status': status, } if products: data['products'] = products data.update(extra) return self._request('POST', self.ENDPOINT, data=data) def update(self, webhook_id: str, **data) -> Dict[str, Any]: """ Atualiza um webhook. Args: webhook_id: ID do webhook **data: Campos a atualizar Returns: Webhook atualizado """ return self._request('PUT', f'{self.ENDPOINT}{webhook_id}/', data=data) def delete(self, webhook_id: str) -> None: """Deleta um webhook.""" self._request('DELETE', f'{self.ENDPOINT}{webhook_id}/') def event_history( self, *, page: int = 1, limit: int = 50, **filters ) -> PaginatedResponse: """ Obtém histórico de eventos enviados. Args: page: Número da página limit: Resultados por página **filters: Filtros adicionais Returns: PaginatedResponse com histórico de eventos """ params = {'page': page, 'limit': limit} params.update(filters) response = self._request('GET', f'{self.ENDPOINT}event_history/', params=params) return PaginatedResponse(**response) def resend_event(self, event_id: str) -> Dict[str, Any]: """ Reenvia um evento específico. Args: event_id: ID do evento Returns: Resposta da API """ return self._request('POST', f'{self.ENDPOINT}event_resend/{event_id}/') def test(self, webhook_id: str) -> Dict[str, Any]: """ Envia evento de teste para um webhook. Args: webhook_id: ID do webhook Returns: Resposta da API """ return self._request('POST', f'{self.ENDPOINT}event_test/{webhook_id}/') # ============================================================================= # Main Client # ============================================================================= class CaktoClient: """ Cliente Python para a API Cakto. Exemplo: # Inicializar cliente client = CaktoClient( client_id='seu_client_id', client_secret='seu_client_secret' ) # Listar produtos ativos products = client.products.list(status='active') for product in products.results: print(f"{product['name']}: R${product['price']}") # Listar todos os pedidos pagos do mês orders = client.orders.list( status='paid', paidAt__gte='2025-01-01', ordering='-paidAt' ) # Iterar sobre TODOS os pedidos (paginação automática) for order in client.orders.list_all(status='paid'): print(f"Pedido {order['refId']}: R${order['amount']}") # Criar webhook para vendas client.webhooks.create( name='Notificação de Vendas', url='https://meusite.com/webhook/cakto', events=['purchase_approved', 'refund'] ) Attributes: products: Gerenciamento de produtos offers: Gerenciamento de ofertas orders: Gerenciamento de pedidos webhooks: Gerenciamento de webhooks """ BASE_URL = 'https://api.cakto.com.br/' AUTH_ENDPOINT = '/oauth/token/' # Endpoint típico OAuth2 def __init__( self, client_id: str, client_secret: str, *, base_url: Optional[str] = None, timeout: int = 30, auto_retry: bool = True, max_retries: int = 3 ): """ Inicializa o cliente Cakto. Args: client_id: Client ID da API (obtido no painel Cakto) client_secret: Client Secret da API base_url: URL base alternativa (opcional) timeout: Timeout das requisições em segundos auto_retry: Tentar novamente em caso de erro de rate limit max_retries: Número máximo de tentativas """ self.client_id = client_id self.client_secret = client_secret self.base_url = base_url or self.BASE_URL self.timeout = timeout self.auto_retry = auto_retry self.max_retries = max_retries self._token: Optional[Token] = None self._session = requests.Session() # Inicializa recursos self.products = ProductsResource(self) self.offers = OffersResource(self) self.orders = OrdersResource(self) self.webhooks = WebhooksResource(self) def _get_token(self) -> str: """Obtém ou renova o token de acesso.""" if self._token and not self._token.is_expired: return self._token.access_token return self._authenticate() def _authenticate(self) -> str: """ Autentica via OAuth2 e obtém access_token. Returns: Access token válido Raises: CaktoAuthError: Se a autenticação falhar """ url = urljoin(self.base_url, self.AUTH_ENDPOINT) data = { 'grant_type': 'client_credentials', 'client_id': self.client_id, 'client_secret': self.client_secret, } try: response = self._session.post( url, data=data, timeout=self.timeout ) response.raise_for_status() token_data = response.json() self._token = Token( access_token=token_data['access_token'], token_type=token_data.get('token_type', 'Bearer'), expires_in=token_data.get('expires_in', 3600), created_at=datetime.now() ) return self._token.access_token except requests.exceptions.HTTPError as e: raise CaktoAuthError(f"Falha na autenticação: {e}") except Exception as e: raise CaktoAuthError(f"Erro ao autenticar: {e}") def _request( self, method: str, endpoint: str, params: Optional[Dict] = None, data: Optional[Dict] = None, retry_count: int = 0 ) -> Any: """ Executa uma requisição à API. Args: method: Método HTTP (GET, POST, PUT, DELETE) endpoint: Endpoint da API params: Query parameters data: Body da requisição (JSON) retry_count: Contador de tentativas (interno) Returns: Resposta da API (JSON) Raises: CaktoError: Em caso de erro na requisição """ url = urljoin(self.base_url, endpoint) token = self._get_token() headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json', 'Accept': 'application/json', } try: response = self._session.request( method=method, url=url, headers=headers, params=params, json=data, timeout=self.timeout ) # Trata erros HTTP if response.status_code == 401: # Token expirado, tenta renovar self._token = None if retry_count < 1: return self._request(method, endpoint, params, data, retry_count + 1) raise CaktoAuthError("Token inválido ou expirado") if response.status_code == 404: raise CaktoNotFoundError(f"Recurso não encontrado: {endpoint}") if response.status_code == 400: error_detail = response.json() if response.text else {} raise CaktoValidationError(f"Erro de validação: {error_detail}") if response.status_code == 429: if self.auto_retry and retry_count < self.max_retries: # Rate limit - aguarda e tenta novamente retry_after = int(response.headers.get('Retry-After', 5)) time.sleep(retry_after) return self._request(method, endpoint, params, data, retry_count + 1) raise CaktoRateLimitError("Rate limit excedido") response.raise_for_status() # DELETE retorna 204 No Content if response.status_code == 204: return None return response.json() if response.text else None except requests.exceptions.Timeout: raise CaktoError(f"Timeout na requisição para {endpoint}") except requests.exceptions.ConnectionError: raise CaktoError(f"Erro de conexão com a API") except (CaktoError, CaktoAuthError, CaktoNotFoundError, CaktoValidationError, CaktoRateLimitError): raise except Exception as e: raise CaktoError(f"Erro na requisição: {e}") def test_connection(self) -> bool: """ Testa a conexão com a API. Returns: True se conectado com sucesso Raises: CaktoError: Se não conseguir conectar """ try: self._authenticate() return True except CaktoAuthError as e: raise CaktoError(f"Falha ao conectar: {e}") # ============================================================================= # Async Base Resource # ============================================================================= class AsyncBaseResource: """Classe base para recursos assíncronos da API.""" def __init__(self, client: 'AsyncCaktoClient'): self._client = client async def _request( self, method: str, endpoint: str, params: Optional[Dict] = None, data: Optional[Dict] = None, **kwargs ) -> Any: return await self._client._request(method, endpoint, params, data, **kwargs) async def _paginate( self, endpoint: str, params: Optional[Dict] = None, limit: int = 50 ) -> AsyncIterator[Dict[str, Any]]: """Itera assincronamente sobre todos os resultados paginados.""" params = params or {} params['limit'] = limit page = 1 while True: params['page'] = page response = await self._request('GET', endpoint, params=params) paginated = PaginatedResponse(**response) for item in paginated.results: yield item if not paginated.has_next: break page += 1 # ============================================================================= # Async Products Resource # ============================================================================= class AsyncProductsResource(AsyncBaseResource): """Gerenciamento assíncrono de Produtos.""" ENDPOINT = '/public_api/products/' async def list( self, *, page: int = 1, limit: int = 50, search: Optional[str] = None, status: Optional[str] = None, type: Optional[str] = None, ordering: Optional[str] = None, **filters ) -> PaginatedResponse: """Lista produtos com filtros.""" params = {'page': page, 'limit': limit} if search: params['search'] = search if status: params['status'] = status if type: params['type'] = type if ordering: params['ordering'] = ordering params.update(filters) response = await self._request('GET', self.ENDPOINT, params=params) return PaginatedResponse(**response) async def list_all(self, **filters) -> AsyncIterator[Dict[str, Any]]: """Itera sobre todos os produtos (todas as páginas).""" async for item in self._paginate(self.ENDPOINT, params=filters): yield item async def get(self, product_id: str) -> Dict[str, Any]: """Obtém um produto pelo ID.""" return await self._request('GET', f'{self.ENDPOINT}{product_id}/') async def update(self, product_id: str, **data) -> Dict[str, Any]: """Atualiza um produto.""" return await self._request('PUT', f'{self.ENDPOINT}{product_id}/', data=data) # ============================================================================= # Async Offers Resource # ============================================================================= class AsyncOffersResource(AsyncBaseResource): """Gerenciamento assíncrono de Ofertas.""" ENDPOINT = '/public_api/offers/' async def list( self, *, page: int = 1, limit: int = 50, search: Optional[str] = None, status: Optional[str] = None, type: Optional[str] = None, product: Optional[str] = None, ordering: Optional[str] = None, **filters ) -> PaginatedResponse: """Lista ofertas com filtros.""" params = {'page': page, 'limit': limit} if search: params['search'] = search if status: params['status'] = status if type: params['type'] = type if product: params['product'] = product if ordering: params['ordering'] = ordering params.update(filters) response = await self._request('GET', self.ENDPOINT, params=params) return PaginatedResponse(**response) async def list_all(self, **filters) -> AsyncIterator[Dict[str, Any]]: """Itera sobre todas as ofertas (todas as páginas).""" async for item in self._paginate(self.ENDPOINT, params=filters): yield item async def get(self, offer_id: str) -> Dict[str, Any]: """Obtém uma oferta pelo ID.""" return await self._request('GET', f'{self.ENDPOINT}{offer_id}/') async def create( self, name: str, price: float, product: str, type: str = 'unique', *, units: int = 1, default: bool = False, intervalType: Optional[str] = None, interval: Optional[int] = None, recurrence_period: Optional[int] = None, quantity_recurrences: Optional[int] = None, trial_days: Optional[int] = None, max_retries: Optional[int] = None, retry_interval: Optional[int] = None, **extra ) -> Dict[str, Any]: """Cria uma nova oferta.""" data = { 'name': name, 'price': price, 'product': product, 'type': type, 'units': units, 'default': default, } if type == 'unique': if intervalType: data['intervalType'] = intervalType if interval: data['interval'] = interval elif type == 'subscription': if recurrence_period: data['recurrence_period'] = recurrence_period if quantity_recurrences is not None: data['quantity_recurrences'] = quantity_recurrences if trial_days is not None: data['trial_days'] = trial_days if max_retries is not None: data['max_retries'] = max_retries if retry_interval is not None: data['retry_interval'] = retry_interval data.update(extra) return await self._request('POST', self.ENDPOINT, data=data) async def update(self, offer_id: str, **data) -> Dict[str, Any]: """Atualiza uma oferta.""" return await self._request('PUT', f'{self.ENDPOINT}{offer_id}/', data=data) async def delete(self, offer_id: str) -> None: """Deleta uma oferta.""" await self._request('DELETE', f'{self.ENDPOINT}{offer_id}/') # ============================================================================= # Async Orders Resource # ============================================================================= class AsyncOrdersResource(AsyncBaseResource): """Gerenciamento assíncrono de Pedidos.""" ENDPOINT = '/public_api/orders/' STATUS_PAID = 'paid' STATUS_REFUNDED = 'refunded' STATUS_WAITING = 'waiting_payment' STATUS_REFUSED = 'refused' STATUS_BLOCKED = 'blocked' STATUS_CHARGEDBACK = 'chargedback' OFFER_TYPE_MAIN = 'main' OFFER_TYPE_UPSELL = 'upsell' OFFER_TYPE_DOWNSELL = 'downsell' OFFER_TYPE_ORDERBUMP = 'orderbump' async def list( self, *, page: int = 1, limit: int = 50, search: Optional[str] = None, status: Optional[str] = None, type: Optional[str] = None, offer_type: Optional[str] = None, product: Optional[str] = None, customer: Optional[str] = None, affiliate: Optional[str] = None, paymentMethod: Optional[str] = None, ordering: Optional[str] = None, **filters ) -> PaginatedResponse: """Lista pedidos com filtros avançados.""" params = {'page': page, 'limit': limit} if search: params['search'] = search if status: params['status'] = status if type: params['type'] = type if offer_type: params['offer_type'] = offer_type if product: params['product'] = product if customer: params['customer'] = customer if affiliate: params['affiliate'] = affiliate if paymentMethod: params['paymentMethod'] = paymentMethod if ordering: params['ordering'] = ordering params.update(filters) response = await self._request('GET', self.ENDPOINT, params=params) return PaginatedResponse(**response) async def list_all(self, **filters) -> AsyncIterator[Dict[str, Any]]: """Itera sobre todos os pedidos (todas as páginas).""" async for item in self._paginate(self.ENDPOINT, params=filters): yield item async def get(self, order_id: str) -> Dict[str, Any]: """Obtém um pedido pelo ID.""" return await self._request('GET', f'{self.ENDPOINT}{order_id}/') async def refund(self, order_id: str) -> Dict[str, Any]: """Solicita reembolso de um pedido.""" return await self._request('POST', f'{self.ENDPOINT}{order_id}/refund/') async def resend_email(self, order_id: str) -> Dict[str, Any]: """Reenvia o email de confirmação de pagamento.""" return await self._request('POST', f'{self.ENDPOINT}{order_id}/resend_approved_email/') async def resend_access(self, order_id: str) -> Dict[str, Any]: """Cria ou atualiza acessos de um pedido.""" return await self._request('POST', f'{self.ENDPOINT}{order_id}/resend_access/') # ============================================================================= # Async Webhooks Resource # ============================================================================= class AsyncWebhooksResource(AsyncBaseResource): """Gerenciamento assíncrono de Webhooks.""" ENDPOINT = '/public_api/webhook/' EVENT_INITIATE_CHECKOUT = 'initiate_checkout' EVENT_CHECKOUT_ABANDONMENT = 'checkout_abandonment' EVENT_PURCHASE_APPROVED = 'purchase_approved' EVENT_PURCHASE_REFUSED = 'purchase_refused' EVENT_PIX_GERADO = 'pix_gerado' EVENT_BOLETO_GERADO = 'boleto_gerado' EVENT_PICPAY_GERADO = 'picpay_gerado' EVENT_OPENFINANCE_NUBANK = 'openfinance_nubank_gerado' EVENT_CHARGEBACK = 'chargeback' EVENT_REFUND = 'refund' EVENT_SUBSCRIPTION_CREATED = 'subscription_created' EVENT_SUBSCRIPTION_CANCELED = 'subscription_canceled' EVENT_SUBSCRIPTION_RENEWED = 'subscription_renewed' EVENT_SUBSCRIPTION_RENEWAL_REFUSED = 'subscription_renewal_refused' ALL_EVENTS = [ EVENT_INITIATE_CHECKOUT, EVENT_CHECKOUT_ABANDONMENT, EVENT_PURCHASE_APPROVED, EVENT_PURCHASE_REFUSED, EVENT_PIX_GERADO, EVENT_BOLETO_GERADO, EVENT_PICPAY_GERADO, EVENT_OPENFINANCE_NUBANK, EVENT_CHARGEBACK, EVENT_REFUND, EVENT_SUBSCRIPTION_CREATED, EVENT_SUBSCRIPTION_CANCELED, EVENT_SUBSCRIPTION_RENEWED, EVENT_SUBSCRIPTION_RENEWAL_REFUSED, ] async def list( self, *, page: int = 1, limit: int = 50, search: Optional[str] = None, status: Optional[str] = None, events: Optional[List[str]] = None, products: Optional[List[str]] = None, ordering: Optional[str] = None, **filters ) -> PaginatedResponse: """Lista webhooks com filtros.""" params = {'page': page, 'limit': limit} if search: params['search'] = search if status: params['status'] = status if events: params['events'] = ','.join(events) if products: params['products'] = ','.join(products) if ordering: params['ordering'] = ordering params.update(filters) response = await self._request('GET', self.ENDPOINT, params=params) return PaginatedResponse(**response) async def list_all(self, **filters) -> AsyncIterator[Dict[str, Any]]: """Itera sobre todos os webhooks (todas as páginas).""" async for item in self._paginate(self.ENDPOINT, params=filters): yield item async def get(self, webhook_id: str) -> Dict[str, Any]: """Obtém um webhook pelo ID.""" return await self._request('GET', f'{self.ENDPOINT}{webhook_id}/') async def create( self, name: str, url: str, events: List[str], *, products: Optional[List[str]] = None, status: str = 'active', **extra ) -> Dict[str, Any]: """Cria um novo webhook.""" data = { 'name': name, 'url': url, 'events': events, 'status': status, } if products: data['products'] = products data.update(extra) return await self._request('POST', self.ENDPOINT, data=data) async def update(self, webhook_id: str, **data) -> Dict[str, Any]: """Atualiza um webhook.""" return await self._request('PUT', f'{self.ENDPOINT}{webhook_id}/', data=data) async def delete(self, webhook_id: str) -> None: """Deleta um webhook.""" await self._request('DELETE', f'{self.ENDPOINT}{webhook_id}/') async def event_history( self, *, page: int = 1, limit: int = 50, **filters ) -> PaginatedResponse: """Obtém histórico de eventos enviados.""" params = {'page': page, 'limit': limit} params.update(filters) response = await self._request('GET', f'{self.ENDPOINT}event_history/', params=params) return PaginatedResponse(**response) async def resend_event(self, event_id: str) -> Dict[str, Any]: """Reenvia um evento específico.""" return await self._request('POST', f'{self.ENDPOINT}event_resend/{event_id}/') async def test(self, webhook_id: str) -> Dict[str, Any]: """Envia evento de teste para um webhook.""" return await self._request('POST', f'{self.ENDPOINT}event_test/{webhook_id}/') # ============================================================================= # Async Client # ============================================================================= class AsyncCaktoClient: """ Cliente assíncrono Python para a API Cakto. Exemplo: # Usando context manager (recomendado) async with AsyncCaktoClient(client_id='xxx', client_secret='yyy') as client: products = await client.products.list(status='active') for product in products.results: print(f"{product['name']}: R${product['price']}") # Sem context manager client = AsyncCaktoClient(client_id='xxx', client_secret='yyy') try: orders = await client.orders.list(status='paid') finally: await client.close() # Requisições paralelas async with AsyncCaktoClient(client_id='xxx', client_secret='yyy') as client: products, orders, webhooks = await asyncio.gather( client.products.list(), client.orders.list(status='paid'), client.webhooks.list() ) # Paginação assíncrona async with AsyncCaktoClient(client_id='xxx', client_secret='yyy') as client: async for order in client.orders.list_all(status='paid'): print(f"Pedido: {order['refId']}") Attributes: products: Gerenciamento de produtos offers: Gerenciamento de ofertas orders: Gerenciamento de pedidos webhooks: Gerenciamento de webhooks """ BASE_URL = 'https://api.cakto.com.br/' AUTH_ENDPOINT = '/oauth/token/' def __init__( self, client_id: str, client_secret: str, *, base_url: Optional[str] = None, timeout: float = 30.0, auto_retry: bool = True, max_retries: int = 3 ): """ Inicializa o cliente assíncrono Cakto. Args: client_id: Client ID da API client_secret: Client Secret da API base_url: URL base alternativa timeout: Timeout das requisições em segundos auto_retry: Tentar novamente em caso de rate limit max_retries: Número máximo de tentativas Raises: ImportError: Se httpx não estiver instalado """ if not HTTPX_AVAILABLE: raise ImportError( "httpx é necessário para o cliente assíncrono. " "Instale com: pip install httpx" ) self.client_id = client_id self.client_secret = client_secret self.base_url = base_url or self.BASE_URL self.timeout = timeout self.auto_retry = auto_retry self.max_retries = max_retries self._token: Optional[Token] = None self._client: Optional[httpx.AsyncClient] = None # Inicializa recursos self.products = AsyncProductsResource(self) self.offers = AsyncOffersResource(self) self.orders = AsyncOrdersResource(self) self.webhooks = AsyncWebhooksResource(self) async def __aenter__(self) -> 'AsyncCaktoClient': """Context manager entry.""" self._client = httpx.AsyncClient(timeout=self.timeout) return self async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: """Context manager exit.""" await self.close() async def close(self) -> None: """Fecha o cliente HTTP.""" if self._client: await self._client.aclose() self._client = None def _get_client(self) -> httpx.AsyncClient: """Obtém ou cria o cliente HTTP.""" if self._client is None: self._client = httpx.AsyncClient(timeout=self.timeout) return self._client async def _get_token(self) -> str: """Obtém ou renova o token de acesso.""" if self._token and not self._token.is_expired: return self._token.access_token return await self._authenticate() async def _authenticate(self) -> str: """Autentica via OAuth2 e obtém access_token.""" url = urljoin(self.base_url, self.AUTH_ENDPOINT) client = self._get_client() data = { 'grant_type': 'client_credentials', 'client_id': self.client_id, 'client_secret': self.client_secret, } try: response = await client.post(url, data=data) response.raise_for_status() token_data = response.json() self._token = Token( access_token=token_data['access_token'], token_type=token_data.get('token_type', 'Bearer'), expires_in=token_data.get('expires_in', 3600), created_at=datetime.now() ) return self._token.access_token except httpx.HTTPStatusError as e: raise CaktoAuthError(f"Falha na autenticação: {e}") except Exception as e: raise CaktoAuthError(f"Erro ao autenticar: {e}") async def _request( self, method: str, endpoint: str, params: Optional[Dict] = None, data: Optional[Dict] = None, retry_count: int = 0 ) -> Any: """Executa uma requisição assíncrona à API.""" url = urljoin(self.base_url, endpoint) token = await self._get_token() client = self._get_client() headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json', 'Accept': 'application/json', } try: response = await client.request( method=method, url=url, headers=headers, params=params, json=data ) # Trata erros HTTP if response.status_code == 401: self._token = None if retry_count < 1: return await self._request(method, endpoint, params, data, retry_count + 1) raise CaktoAuthError("Token inválido ou expirado") if response.status_code == 404: raise CaktoNotFoundError(f"Recurso não encontrado: {endpoint}") if response.status_code == 400: error_detail = response.json() if response.text else {} raise CaktoValidationError(f"Erro de validação: {error_detail}") if response.status_code == 429: if self.auto_retry and retry_count < self.max_retries: retry_after = int(response.headers.get('Retry-After', 5)) await asyncio.sleep(retry_after) return await self._request(method, endpoint, params, data, retry_count + 1) raise CaktoRateLimitError("Rate limit excedido") response.raise_for_status() if response.status_code == 204: return None return response.json() if response.text else None except httpx.TimeoutException: raise CaktoError(f"Timeout na requisição para {endpoint}") except httpx.ConnectError: raise CaktoError("Erro de conexão com a API") except (CaktoError, CaktoAuthError, CaktoNotFoundError, CaktoValidationError, CaktoRateLimitError): raise except Exception as e: raise CaktoError(f"Erro na requisição: {e}") async def test_connection(self) -> bool: """Testa a conexão com a API.""" try: await self._authenticate() return True except CaktoAuthError as e: raise CaktoError(f"Falha ao conectar: {e}") # ============================================================================= # Convenience Functions # ============================================================================= def create_client(client_id: str, client_secret: str, **kwargs) -> CaktoClient: """ Cria uma instância do cliente Cakto. Args: client_id: Client ID da API client_secret: Client Secret da API **kwargs: Argumentos adicionais para CaktoClient Returns: Instância configurada do CaktoClient """ return CaktoClient(client_id, client_secret, **kwargs) # ============================================================================= # CLI / Main # ============================================================================= if __name__ == '__main__': import os # Exemplo de uso print("Cakto API Python Client") print("=" * 40) print() print("Uso:") print(" from cakto import CaktoClient") print() print(" client = CaktoClient(") print(" client_id='seu_client_id',") print(" client_secret='seu_client_secret'") print(" )") print() print(" # Listar produtos") print(" products = client.products.list()") print() print(" # Listar pedidos pagos") print(" orders = client.orders.list(status='paid')") print() print(" # Criar webhook") print(" client.webhooks.create(") print(" name='Vendas',") print(" url='https://meusite.com/webhook',") print(" events=['purchase_approved']") print(" )") print() print("Documentação: https://docs.cakto.com.br")