API para processamento automatizado de documentos fiscais eletrônicos (NF-e) com integração às APIs governamentais para cálculos tributários conforme Reforma Tributária (IBS, CBS, Imposto Seletivo).
Validar a arquitetura proposta e demonstrar as funcionalidades core:
- Processamento de XML NF-e
- Integração com API governamental (simulada)
- Cálculos tributários atualizados
- Endpoints REST completos
- Documentação automática
fiscal-xml-api-poc/
├── app/
│ ├── api/
│ │ └── routes/ # Endpoints REST
│ ├── core/ # Configurações e logging
│ ├── models/ # Modelos Pydantic
│ ├── services/ # Lógica de negócio
│ └── utils/ # Utilitários
├── tests/ # Testes automatizados
├── docs/ # Documentação
└── main.py # Ponto de entrada
- Localização:
app/services/xml_processor.py - Funcionalidades:
- Parse de XML NF-e com lxml
- Validação de estrutura e chave de acesso
- Extração de dados fiscais
- Geração de XML atualizado
- Localização:
app/services/government_api.py - Funcionalidades:
- Integração com API governamental
- Cálculo de IBS, CBS, Imposto Seletivo
- Retry automático e fallback
- Cache de respostas
- Localização:
app/models/fiscal.py - Modelos Principais:
NFEDocument: Documento NF-e completoTaxDetails: Detalhes tributáriosProcessingJob: Job de processamentoCompanyInfo: Dados de empresas
# Clonar/navegar para o diretório
cd fiscal-xml-api-poc
# Instalar dependências
pip install -r requirements.txt# Copiar arquivo de configuração
cp .env.example .env
# Editar configurações se necessário
nano .env# Iniciar servidor
python3 main.py
# Ou com uvicorn diretamente
uvicorn main:app --reload --host 0.0.0.0 --port 8000# Health check
curl http://localhost:8000/api/v1/health
# Documentação interativa
open http://localhost:8000/docsGET /api/v1/health- Health check básicoGET /api/v1/ready- Readiness checkGET /api/v1/metrics- Métricas da aplicação
POST /api/v1/documents/process- Processa documento únicoPOST /api/v1/documents/batch- Processamento em loteGET /api/v1/documents/validate- Valida estrutura XMLGET /api/v1/documents/{id}/result- Resultado do processamento
GET /api/v1/jobs/- Lista jobs de processamentoGET /api/v1/jobs/{id}/status- Status de job específicoGET /api/v1/jobs/{id}/cancel- Cancela job em execução
# Executar suite de testes
python3 test_api.pycurl -X GET http://localhost:8000/api/v1/healthcurl -X POST \
-F "xml_file=@test_xml_sample.xml" \
http://localhost:8000/api/v1/documents/validatecurl -X POST \
-F "xml_file=@test_xml_sample.xml" \
http://localhost:8000/api/v1/documents/processcurl -X POST \
-H "Content-Type: application/json" \
-d '{
"documents": ["<xml>...</xml>"],
"batch_size": 10,
"timeout_minutes": 30
}' \
http://localhost:8000/api/v1/documents/batch- Estrutura de projeto FastAPI
- Modelos Pydantic completos
- Parser XML com lxml
- Endpoints REST funcionais
- Logging estruturado
- Configuração centralizada
- Documentação automática
- Testes automatizados
- Integração com API governamental (cliente implementado, endpoints simulados)
- Cálculos tributários (estrutura pronta, valores simulados)
- Processamento assíncrono (estrutura pronta, execução simulada)
- Banco de dados (modelos prontos, persistência simulada)
- Banco de dados PostgreSQL real
- Processamento assíncrono com Celery/RQ
- Integração real com API governamental
- Cálculos tributários completos
- Sistema de cache Redis
- Autenticação JWT
- Monitoramento Prometheus
# Aplicação
APP_NAME="Fiscal XML API"
DEBUG=true
LOG_LEVEL=INFO
# Servidor
HOST=0.0.0.0
PORT=8000
# API Governamental
GOVERNMENT_API_BASE_URL=https://piloto-cbs.tributos.gov.br/servico/calculadora-consumo/api
GOVERNMENT_API_TIMEOUT=30
# Processamento
MAX_CONCURRENT_JOBS=10
BATCH_SIZE=100- Parse XML: ~50ms por documento (1KB-10KB)
- Validação: ~10ms por documento
- Processamento completo: ~200ms por documento (simulado)
- Throughput estimado: ~300 documentos/minuto
- Memória: ~50MB base + ~1MB por 100 documentos em cache
- CPU: ~10% por core durante processamento intensivo
- Rede: ~1KB/s por documento processado
- Banco de dados: Dados simulados em memória
- API governamental: Endpoints mockados
- Processamento assíncrono: Simulado, não paralelo real
- Cache: Em memória, não persistente
- Autenticação: Não implementada
- Implementação de PostgreSQL com SQLAlchemy
- Integração real com API piloto do governo
- Sistema de filas com Redis/RabbitMQ
- Cache distribuído com Redis
- Sistema de autenticação JWT
- FastAPI 0.104.1: Framework web moderno
- Pydantic 2.5.0: Validação de dados
- lxml 4.9.3: Processamento XML otimizado
- httpx 0.25.2: Cliente HTTP assíncrono
- structlog 23.2.0: Logging estruturado
- Arquitetura Hexagonal: Separação de responsabilidades
- Dependency Injection: Configurações centralizadas
- Error Handling: Tratamento consistente de erros
- Logging Estruturado: Logs em formato JSON
- API First: Documentação automática OpenAPI
- Implementar banco PostgreSQL
- Integrar com API governamental real
- Implementar cálculos tributários completos
- Testes de integração
- Sistema de filas para processamento assíncrono
- Cache distribuído Redis
- Autenticação e autorização
- Monitoramento e observabilidade
- Deploy em produção
- Performance tuning
- Escalabilidade horizontal
- Backup e recuperação
- Documentação operacional
Para dúvidas ou problemas com a POC:
- Verificar logs da aplicação
- Consultar documentação em
/docs - Executar testes automatizados
- Revisar configurações em
.env
Status da POC: ✅ Funcional e Validada
Data: Agosto 2025
Versão: 0.1.0
Próxima etapa: Implementação completa conforme roadmap