Como Criar um Sistema de Upload de Arquivos com Feedback em Tempo Real Usando FastAPI e JavaScript

O processo de upload de arquivos é uma parte fundamental de muitas aplicações web. No entanto, a experiência do usuário pode ser muito mais envolvente e eficiente se o sistema fornecer um retorno visual imediato sobre o progresso do upload e, ao final, fornecer um link direto para o arquivo carregado. Aqui, exploraremos como implementar essa funcionalidade utilizando o FastAPI no backend e JavaScript no frontend.

Primeiro, no frontend, utilizamos a API drag and drop para permitir que os usuários arrastem arquivos diretamente para uma área de upload na interface. O comportamento de destaque dessa área, quando o usuário arrasta o arquivo sobre ela, é tratado por meio de eventos de dragover e dragleave, garantindo uma interação visual clara e agradável. Ao soltar o arquivo, um evento drop é acionado, que inicia o processo de upload.

javascript
Copy code
dropArea.addEventListener(eventName, e => { 
    e.preventDefault(); 
    e.stopPropagation(); 
    dropArea.classList.remove('active'); 
});
dropArea.addEventListener('drop', e => { 
    const files = e.dataTransfer.files; 
    if (files.length > 0) { 
        uploadFile(files[0]); 
    } 
});
function uploadFile(file) { 
    progressContainer.textContent = ''; 
    fileUrl.textContent = ''; 
    const xhr = new XMLHttpRequest(); 

    xhr.open('POST', '/upload', true); 

    xhr.upload.onprogress = (e) => { 
        if (e.lengthComputable) { 
            const percent = Math.round((e.loaded / e.total) * 100); 
            progressContainer.textContent = `Uploading: ${percent}% (${e.loaded} / ${e.total} bytes)`; 
        } 
    }; 
    xhr.onload = () => { 
        if (xhr.status === 200) { 
            progressContainer.textContent = 'Upload complete!'; 

            const resp = JSON.parse(xhr.responseText); 

            fileUrl.innerHTML = `File URL: ${resp.url}`; 
        } else { 
            progressContainer.textContent = 'Upload failed: ' + xhr.responseText; 
        } 
    }; 
    xhr.onerror = () => { 
        progressContainer.textContent = 'An error occurred during upload.'; 
    }; 
    const formData = new FormData(); 
    formData.append('file', file); 
    xhr.send(formData); 
}

Ao enviar o arquivo, o backend processa e armazena o arquivo, retornando uma URL pública para acesso. Em FastAPI, o código responsável por receber e salvar os arquivos é simples e direto. A seguir, mostramos como configurá-lo utilizando armazenamento local. O FastAPI oferece uma rota POST que lida com o upload de arquivos, atribuindo a cada arquivo um identificador único (UUID) para garantir que não haja colisões de nomes.

python
Copy code
from fastapi import APIRouter, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from pathlib import Path
import shutil
import uuid
router = APIRouter()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
@router.post("/upload")

async def upload_file(file: UploadFile = File(...)):

    file_id = str(uuid.uuid4())
    file_ext = Path(file.filename).suffix
    save_path = UPLOAD_DIR / f"{file_id}{file_ext}"
    try:
        with save_path.open("wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
    except Exception:
        raise HTTPException(status_code=500, detail="Failed to save file")
    url = f"/files/{file_id}{file_ext}"
    return JSONResponse({"url": url})

O FastAPI também facilita o servir de arquivos locais através do suporte a arquivos estáticos. Ao configurar a aplicação para servir a pasta de uploads, os arquivos tornam-se acessíveis via URL, permitindo que os usuários façam download deles diretamente da aplicação.

python
Copy code
from fastapi import FastAPI

from fastapi.staticfiles import StaticFiles
from routes.uploads import router as uploads_router
app = FastAPI()
app.include_router(uploads_router)
app.mount("/files", StaticFiles(directory="uploads"), name="files")

Em um ambiente de produção, a escalabilidade do armazenamento local pode ser limitada. Para lidar com arquivos grandes e garantir maior robustez e acessibilidade global, o armazenamento em nuvem como o AWS S3 se torna uma excelente opção. Usando a biblioteca boto3, o FastAPI pode ser configurado para enviar arquivos diretamente para o S3, oferecendo alta disponibilidade e redundância. O código a seguir mostra como fazer isso:

python
Copy code
import os
import boto3
USE_S3 = os.environ.get("USE_S3", "false").lower() == "true"
S3_BUCKET = os.environ.get("S3_BUCKET", "")

S3_CLIENT = boto3.client("s3") if USE_S3 else None

@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    file_id = str(uuid.uuid4())
    file_ext = Path(file.filename).suffix
    filename = f"{file_id}{file_ext}"
    if USE_S3 and S3_CLIENT:
        try:
            S3_CLIENT.upload_fileobj(file.file, S3_BUCKET, filename)
            url = f"https://{S3_BUCKET}.s3.amazonaws.com/{filename}"
            return JSONResponse({"url": url})
        except Exception as e:

            raise HTTPException(status_code=500, detail=f"Failed to upload to S3: {str(e)}")

    else:
        save_path = UPLOAD_DIR / filename
        try:
            with save_path.open("wb") as buffer:
                shutil.copyfileobj(file.file, buffer)
        except Exception:
            raise HTTPException(status_code=500, detail="Failed to save file")
        url = f"/files/{filename}"
        return JSONResponse({"url": url})

Além disso, ao implementar o envio de arquivos com feedback em tempo real, o usuário tem uma experiência de uso muito mais interativa e satisfatória. O monitoramento contínuo do progresso do upload torna o processo mais transparente e reduz a frustração associada ao envio de arquivos pesados. Este tipo de funcionalidade é particularmente útil em sistemas que lidam com grandes volumes de dados ou arquivos de mídia, como vídeos e imagens.

É importante notar que, enquanto o código apresentado é adequado para ambientes de desenvolvimento e pequenas implementações, sistemas de produção exigem mais cuidados com a segurança e a performance. Por exemplo, é essencial validar os arquivos enviados, limitando o tamanho e tipos de arquivos aceitos, além de garantir que o sistema de armazenamento esteja preparado para lidar com grandes volumes de dados de forma eficiente e segura.

Como Implementar Notificações Eficientes em Aplicações Web com FastAPI, Celery e Twilio

Em sistemas modernos, a implementação de notificações em tempo real se tornou uma necessidade primordial, especialmente quando se trata de interações com usuários. Para tornar esse processo mais eficiente e escalável, soluções como FastAPI, Celery e Twilio são extremamente úteis. Esses componentes não apenas facilitam o envio de notificações via e-mail e SMS, mas também garantem que o sistema continue performático e robusto, mesmo em cenários de alto tráfego.

Um aspecto fundamental no design de sistemas modernos é o uso de tarefas assíncronas. Com o Celery, podemos enfileirar tarefas de envio de e-mail ou SMS, garantindo que o processo de notificação não bloqueie as interações do usuário. Em vez de realizar a tarefa imediatamente, o sistema coloca a solicitação na fila e retorna ao usuário rapidamente. Isso é possível graças à integração com o Celery, que gerencia essas tarefas de fundo de maneira eficiente.

Enfileiramento de Tarefas de E-mail

Ao enviar notificações por e-mail, é importante considerar a escalabilidade e a confiabilidade. Usando o Celery, podemos colocar as tarefas de envio de e-mail em uma fila, permitindo que o trabalhador Celery as execute quando for possível, sem afetar o desempenho da API principal. Por exemplo, ao registrar um novo usuário, o sistema pode enfileirar o envio do e-mail de boas-vindas, permitindo que o processo de registro seja concluído rapidamente. O código a seguir ilustra essa abordagem:

python
Copy code
from fastapi import APIRouter
from app.tasks import send_email_task
router = APIRouter()
@router.post("/register")
def register_user(email: str):
    subject = "Bem-vindo!"
    html_body = "Obrigado por se inscrever!"
    send_email_task.delay(email, subject, html_body)
    return {"msg": "Cadastro completo. Verifique seu e-mail."}

Aqui, o método .delay() coloca a tarefa na fila do Redis, e o trabalhador Celery a processa assim que possível, permitindo uma resposta rápida ao usuário sem bloqueios.

Monitoramento de Tarefas do Celery

A ferramenta Flower é um recurso essencial para monitorar as tarefas do Celery em tempo real. Com ela, é possível visualizar o status de cada tarefa, incluindo entregas bem-sucedidas, falhas, tentativas de reenvio e o estado geral da fila. Essa visibilidade em tempo real é vital para a manutenção e auditoria do sistema de notificações, ajudando na identificação e resolução de problemas sem a necessidade de vasculhar arquivos de log. Para executar o Flower, basta utilizar os seguintes comandos:

bash
Copy code
pip install flower

celery -A app.celery_worker.celery_app flower --port=5555

Após executar, o painel do Flower estará disponível em http://localhost:5555, onde você poderá monitorar as tarefas ativas, os resultados das tarefas, as tentativas de reenvio e detalhes completos sobre cada tarefa.

Envio de Notificações SMS com Twilio

Além das notificações por e-mail, as mensagens SMS têm se tornado uma forma cada vez mais popular de comunicação em tempo real com os usuários. O Twilio é uma das plataformas líderes para envio e rastreamento de SMS em nível global. Em um design robusto, o envio de SMS deve ser assíncrono e não bloquear a execução da aplicação. Para isso, utilizamos Celery para enfileirar as tarefas de envio, o HTTPX para fazer as requisições assíncronas e monitoramento em tempo real via webhooks do Twilio.

Primeiro, é necessário configurar as credenciais do Twilio e armazená-las como variáveis de ambiente:

python
Copy code
import os
TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID")
TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")
TWILIO_FROM_NUMBER = os.getenv("TWILIO_FROM_NUMBER")
TWILIO_API_URL = f"https://api.twilio.com/2010-04-01/Accounts/{TWILIO_ACCOUNT_SID}/Messages.json"

Em seguida, criamos uma função para enviar o SMS de forma assíncrona utilizando o HTTPX:

python
Copy code
import httpx

from config import TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_FROM_NUMBER, TWILIO_API_URL
async def send_sms_via_twilio(to_number: str, body: str, status_callback_url: str = None):
    data = {
        "From": TWILIO_FROM_NUMBER,
        "To": to_number,
        "Body": body,
    }
    if status_callback_url:
        data["StatusCallback"] = status_callback_url
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            TWILIO_API_URL,
            data=data,
            auth=(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)
        )
        resp.raise_for_status()
        return resp.json()

Para enfileirar o envio do SMS, o Celery pode ser utilizado para processar a tarefa em segundo plano, sem impactar a performance do sistema principal:

python
Copy code
from app.celery_worker import celery_app
from app.sms_utils import send_sms_via_twilio
import asyncio

@celery_app.task(bind=True, max_retries=4, default_retry_delay=40)

def send_sms_task(self, to_number, message, status_callback_url=None):
    try:
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(
            send_sms_via_twilio(to_number, message, status_callback_url)
        )
        return result
    except Exception as exc:
        raise self.retry(exc=exc)

Monitoramento e Resiliência com Webhooks do Twilio

python
Copy code
from fastapi import APIRouter, Request, status
from fastapi.responses import JSONResponse
router = APIRouter()
@router.post("/webhooks/twilio-sms-status")

async def twilio_sms_status_webhook(request: Request):

    data = await request.form()
    message_sid = data.get("MessageSid")
    message_status = data.get("MessageStatus")
    to_number = data.get("To")
    print(f"Twilio SMS: {message_sid} para {to_number} está agora {message_status}")
    return JSONResponse({"status": "received"}, status_code=status.HTTP_200_OK)

Essa configuração permite que o sistema monitore as entregas e reaja a problemas, como falhas de rede ou erros temporários na API do Twilio.

Escalabilidade e Confiabilidade

Além disso, a arquitetura assíncrona e a integração com ferramentas como Flower para monitoramento e o Twilio para rastreamento de status de SMS proporcionam visibilidade completa do processo de notificação. Isso permite identificar e resolver problemas de forma rápida e eficaz, sem impactar a experiência do usuário final.

Como Criar Consultas Dinâmicas e Eficientes para Filtragem e Exportação de Dados em Aplicações de Grande Escala

Ao criar filtros dinâmicos, uma das abordagens mais eficazes é a utilização de uma cadeia de condições, onde cada novo filtro é adicionado sem duplicação de lógica, o que mantém o código limpo e de fácil manutenção. No caso do SQLAlchemy, cada novo filtro é simplesmente anexado à consulta SQL existente, o que evita a necessidade de estruturas complexas como blocos de "if-else" aninhados. Isso também significa que a implementação de novos filtros no futuro se torna trivial, e a consulta gerada reflete apenas as condições necessárias para a execução, o que resulta em um código eficiente.

Em sistemas com grande volume de tráfego, a capacidade de operar com consultas assíncronas é fundamental para garantir escalabilidade. O uso de operações assíncronas permite que a aplicação realize múltiplas operações simultaneamente, sem bloquear a execução de outras tarefas. No contexto do FastAPI, isso é possível ao definir endpoints como funções assíncronas, usando a palavra-chave async def, e realizando operações de banco de dados com sessões assíncronas, utilizando o comando await para cada operação de consulta. Essa abordagem é especialmente útil para aplicações que precisam lidar com dados em tempo real ou em larga escala.

No entanto, quando o sistema precisa lidar com grandes volumes de dados para importação ou exportação, é essencial adotar práticas que assegurem não apenas a correção dos dados, mas também a performance e a escalabilidade. As operações de importação e exportação em massa devem ser projetadas para processar grandes volumes de dados sem comprometer os recursos de memória ou a performance do sistema. Uma das técnicas chave para isso é o "streaming", que permite enviar os dados em pequenas partes, sem precisar carregar o conjunto de dados inteiro na memória de uma vez.

No caso da exportação de dados, o streaming pode ser realizado de forma eficiente utilizando o módulo csv do Python, que escreve as linhas conforme são processadas. Para dados no formato JSON, um padrão semelhante é utilizado, garantindo que cada objeto seja enviado conforme é gerado, sem sobrecarregar a memória. O FastAPI oferece suporte robusto para essas operações de streaming, permitindo que os dados sejam exportados de forma eficiente, seja no formato CSV ou JSON.

Um exemplo básico de exportação de dados em formato CSV usando o FastAPI pode ser visto em código, onde os dados são gerados e enviados ao cliente conforme são processados, sem sobrecarregar a memória:

python
Copy code
from fastapi.responses import StreamingResponse
import csv
from io import StringIO
@app.get("/books/export/csv")
def export_books_csv():
    def generate():
        output = StringIO()
        writer = csv.writer(output)

        writer.writerow(['id', 'title', 'author', 'description', 'year'])

        yield output.getvalue()
        output.seek(0)
        output.truncate(0)
        for book in book_service.list_books():
            writer.writerow([book.id, book.title, book.author, book.description, book.year])
            yield output.getvalue()
            output.seek(0)
            output.truncate(0)
    return StreamingResponse(generate(), media_type="text/csv", headers={
        "Content-Disposition": "attachment; filename=books.csv"
    })

No exemplo acima, a função generate() cria a resposta de streaming, escrevendo os dados no formato CSV e enviando-os em pequenos blocos. Esse processo é altamente eficiente e evita problemas de memória, independentemente do tamanho da base de dados.

A exportação de dados no formato JSON pode seguir um padrão semelhante. No entanto, é necessário garantir que a sintaxe do JSON seja válida, o que inclui o manejo adequado das vírgulas entre os objetos. O código abaixo mostra como isso pode ser feito de forma eficiente:

python
Copy code
import json
@app.get("/books/export/json")
def export_books_json():
    def generate():
        yield '['
        first = True
        for book in book_service.list_books():
            if not first:
                yield ','
            yield json.dumps(book.dict())
            first = False
        yield ']'
    return StreamingResponse(generate(), media_type="application/json", headers={
        "Content-Disposition": "attachment; filename=books.json"
    })

Com esse método, os dados são enviados como um array JSON, onde cada objeto é enviado separadamente e a memória é mantida sob controle.