Integração completa para geração de PDFs em Django/Python
pip install requests
CLOUD2PDF_API_URL = 'https://api.cloud2pdf.com/v1'
CLOUD2PDF_API_TOKEN = 'seu_token_aqui'
Copie o código da seção abaixo
Crie myapp/services/cloud2pdf_service.py
import requests
import time
from django.conf import settings
from typing import Optional, Dict, Any
class Cloud2PdfService:
def __init__(self):
self.api_url = settings.CLOUD2PDF_API_URL
self.api_token = settings.CLOUD2PDF_API_TOKEN
self.headers = {
'Authorization': f'Bearer {self.api_token}',
'Content-Type': 'application/json',
'Accept': 'application/json'
}
def generate_async(
self,
html: str,
reference: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
use_cdn: bool = True
) -> Dict[str, Any]:
"""Gera PDF de forma assíncrona"""
if options is None:
options = {}
payload = {
'html': html,
'is_disk': use_cdn,
'reference': reference,
'options': {
'format': options.get('format', 'A4'),
'landscape': options.get('landscape', False),
**options
}
}
response = requests.post(
f'{self.api_url}/pdf/generate-async',
json=payload,
headers=self.headers,
timeout=30
)
response.raise_for_status()
return response.json()
def check_status(self, pdf_id: int) -> Dict[str, Any]:
"""Verifica status do PDF"""
response = requests.get(
f'{self.api_url}/pdf/status/{pdf_id}',
headers=self.headers,
timeout=10
)
response.raise_for_status()
return response.json()
def wait_for_completion(
self,
pdf_id: int,
max_attempts: int = 60,
delay_seconds: int = 5
) -> Dict[str, Any]:
"""Aguarda conclusão do PDF (polling)"""
for _ in range(max_attempts):
status = self.check_status(pdf_id)
if status['status'] == 'completed':
return status
if status['status'] == 'failed':
raise Exception(f"PDF generation failed: {status.get('error', 'Unknown error')}")
time.sleep(delay_seconds)
raise Exception('PDF generation timeout')
def download(self, pdf_id: int) -> bytes:
"""Baixa PDF como bytes"""
response = requests.get(
f'{self.api_url}/pdf/download/{pdf_id}',
headers={'Authorization': f'Bearer {self.api_token}'},
timeout=30
)
response.raise_for_status()
return response.content
def download_and_save(self, pdf_id: int, file_path: str) -> str:
"""Baixa e salva PDF localmente"""
content = self.download(pdf_id)
with open(file_path, 'wb') as f:
f.write(content)
return file_path
def generate_sync(self, html: str, options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Gera PDF síncrono (apenas para PDFs pequenos!)"""
if options is None:
options = {}
payload = {
'html': html,
'is_disk': False,
'options': options
}
response = requests.post(
f'{self.api_url}/pdf/generate',
json=payload,
headers=self.headers,
timeout=60
)
response.raise_for_status()
return response.json()
from django.shortcuts import render, get_object_or_404
from django.http import HttpResponse, JsonResponse
from django.template.loader import render_to_string
from .models import Invoice
from .services.cloud2pdf_service import Cloud2PdfService
def generate_invoice_pdf(request, invoice_id):
"""Gerar PDF assíncrono"""
invoice = get_object_or_404(Invoice, pk=invoice_id)
# Renderizar HTML
html = render_to_string('invoices/pdf.html', {'invoice': invoice})
# Gerar PDF
pdf_service = Cloud2PdfService()
result = pdf_service.generate_async(
html,
reference=f'invoice-{invoice.id}',
options={
'format': 'A4',
'landscape': False,
'margins': {
'top': '20mm',
'right': '15mm',
'bottom': '20mm',
'left': '15mm'
}
}
)
# Salvar PDF ID
invoice.pdf_id = result['pdf_id']
invoice.pdf_status = 'processing'
invoice.save()
return JsonResponse({
'message': 'PDF generation started',
'pdf_id': result['pdf_id'],
'status_url': f'/invoices/{invoice.id}/pdf/status/'
})
def check_invoice_pdf_status(request, invoice_id):
"""Verificar status do PDF"""
invoice = get_object_or_404(Invoice, pk=invoice_id)
if not invoice.pdf_id:
return JsonResponse({'error': 'No PDF generated'}, status=404)
pdf_service = Cloud2PdfService()
status = pdf_service.check_status(invoice.pdf_id)
# Atualizar status no banco
if status['status'] == 'completed':
invoice.pdf_status = 'completed'
invoice.pdf_url = status.get('public_link', {}).get('url')
invoice.pdf_size = status.get('metadata', {}).get('file_size')
invoice.save()
return JsonResponse(status)
def download_invoice_pdf(request, invoice_id):
"""Download do PDF"""
invoice = get_object_or_404(Invoice, pk=invoice_id)
if not invoice.pdf_id:
return HttpResponse('No PDF generated', status=404)
pdf_service = Cloud2PdfService()
pdf_content = pdf_service.download(invoice.pdf_id)
response = HttpResponse(pdf_content, content_type='application/pdf')
response['Content-Disposition'] = f'attachment; filename="invoice-{invoice.id}.pdf"'
return response
Para processamento verdadeiramente assíncrono
from celery import shared_task
from .models import Invoice
from .services.cloud2pdf_service import Cloud2PdfService
@shared_task
def generate_invoice_pdf_task(invoice_id):
"""Gera PDF em background com Celery"""
try:
invoice = Invoice.objects.get(pk=invoice_id)
# Renderizar HTML
from django.template.loader import render_to_string
html = render_to_string('invoices/pdf.html', {'invoice': invoice})
# Gerar PDF
pdf_service = Cloud2PdfService()
result = pdf_service.generate_async(
html,
reference=f'invoice-{invoice.id}'
)
# Salvar PDF ID
invoice.pdf_id = result['pdf_id']
invoice.pdf_status = 'processing'
invoice.save()
# Aguardar conclusão (opcional)
completed = pdf_service.wait_for_completion(result['pdf_id'])
# Atualizar com dados finais
invoice.pdf_status = 'completed'
invoice.pdf_url = completed.get('public_link', {}).get('url')
invoice.pdf_size = completed.get('metadata', {}).get('file_size')
invoice.save()
# Opcional: Baixar e salvar localmente
file_path = f'pdfs/invoices/invoice-{invoice.id}.pdf'
pdf_service.download_and_save(result['pdf_id'], file_path)
invoice.pdf_local_path = file_path
invoice.save()
return {'success': True, 'pdf_id': result['pdf_id']}
except Exception as e:
invoice.pdf_status = 'failed'
invoice.save()
raise
# Uso na view:
def create_invoice(request):
# ... criar invoice ...
# Disparar task
generate_invoice_pdf_task.delay(invoice.id)
return JsonResponse({'message': 'Invoice created, PDF generation started'})
from django.views.decorators.csrf import csrf_exempt
from django.http import JsonResponse
import hmac
import hashlib
import json
@csrf_exempt
def cloud2pdf_webhook(request):
"""Recebe webhooks do Cloud2Pdf"""
if request.method != 'POST':
return JsonResponse({'error': 'Method not allowed'}, status=405)
# Validar signature
signature = request.headers.get('X-Webhook-Signature', '')
secret = settings.CLOUD2PDF_WEBHOOK_SECRET
expected_signature = hmac.new(
secret.encode(),
request.body,
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected_signature):
return JsonResponse({'error': 'Invalid signature'}, status=401)
# Processar evento
data = json.loads(request.body)
event = data.get('event')
pdf_id = data.get('pdf_id')
reference = data.get('reference', '')
if event == 'pdf.completed':
# Extrair ID da reference
invoice_id = reference.replace('invoice-', '')
try:
invoice = Invoice.objects.get(pk=invoice_id)
invoice.pdf_status = 'completed'
invoice.pdf_url = data.get('public_link')
invoice.pdf_size = data.get('file_size')
invoice.credits_used = data.get('credits_used')
invoice.save()
# Enviar email ao cliente
# send_invoice_ready_email(invoice)
except Invoice.DoesNotExist:
pass
elif event == 'pdf.failed':
invoice_id = reference.replace('invoice-', '')
Invoice.objects.filter(pk=invoice_id).update(pdf_status='failed')
return JsonResponse({'success': True})
# urls.py
from django.urls import path
urlpatterns = [
path('webhooks/cloud2pdf/', cloud2pdf_webhook, name='cloud2pdf_webhook'),
]
Não bloqueie requests HTTP com polling
Use render_to_string() para converter templates em HTML
Facilita rastreamento e webhooks