143 lines
4.7 KiB
Python
143 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any, Dict
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, HTTPException, Path, status
|
|
|
|
_CNPJ_API_TEMPLATE = "https://www.receitaws.com.br/v1/cnpj/{cnpj}"
|
|
_CNPJ_SUCCESS_EXAMPLE: Dict[str, Any] = {
|
|
"status": "OK",
|
|
"cnpj": "27865757000102",
|
|
"tipo": "MATRIZ",
|
|
"abertura": "10/05/2010",
|
|
"nome": "SAVE IN MED COMERCIO DE MEDICAMENTOS LTDA",
|
|
"fantasia": "SAVE IN MED",
|
|
"porte": "DEMAIS",
|
|
"natureza_juridica": "206-2 - Sociedade Empresária Limitada",
|
|
"situacao": "ATIVA",
|
|
"capital_social": "50000.00",
|
|
"atividade_principal": [
|
|
{
|
|
"code": "47.71-7-01",
|
|
"text": "Comércio varejista de produtos farmacêuticos, sem manipulação de fórmulas",
|
|
}
|
|
],
|
|
"atividades_secundarias": [
|
|
{
|
|
"code": "47.89-0-99",
|
|
"text": "Comércio varejista de outros produtos não especificados anteriormente",
|
|
}
|
|
],
|
|
"qsa": [
|
|
{
|
|
"nome": "MARIA DE SOUZA LIMA",
|
|
"qual": "16-Sócio",
|
|
}
|
|
],
|
|
}
|
|
|
|
router = APIRouter(
|
|
prefix="/public/cnpj",
|
|
tags=["Público • CNPJ"],
|
|
)
|
|
|
|
|
|
def _normalize_cnpj(raw: str) -> str:
|
|
return re.sub(r"\D", "", raw or "")
|
|
|
|
|
|
@router.get(
|
|
"/{cnpj:path}",
|
|
summary="Consulta pública de CNPJ (ReceitaWS)",
|
|
description=(
|
|
"Consulta o serviço público da ReceitaWS para obter informações do CNPJ fornecido."
|
|
),
|
|
responses={
|
|
status.HTTP_200_OK: {
|
|
"description": "Retorno da ReceitaWS",
|
|
"content": {"application/json": {"example": _CNPJ_SUCCESS_EXAMPLE}},
|
|
},
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": "CNPJ não encontrado",
|
|
"content": {"application/json": {"example": {"detail": "CNPJ não encontrado"}}},
|
|
},
|
|
status.HTTP_422_UNPROCESSABLE_ENTITY: {
|
|
"description": "CNPJ inválido",
|
|
"content": {
|
|
"application/json": {
|
|
"example": {"detail": "CNPJ deve conter 14 dígitos numéricos."}
|
|
}
|
|
},
|
|
},
|
|
status.HTTP_429_TOO_MANY_REQUESTS: {
|
|
"description": "Serviço externo indisponível",
|
|
"content": {
|
|
"application/json": {
|
|
"example": {"detail": "Serviço externo indisponível"}
|
|
}
|
|
},
|
|
},
|
|
status.HTTP_502_BAD_GATEWAY: {
|
|
"description": "Erro ao acessar serviço externo",
|
|
"content": {
|
|
"application/json": {
|
|
"example": {"detail": "Serviço externo indisponível"}
|
|
}
|
|
},
|
|
},
|
|
},
|
|
)
|
|
async def consultar_cnpj(
|
|
cnpj: str = Path(
|
|
...,
|
|
description="CNPJ a ser consultado. Aceita formatos com ou sem máscara.",
|
|
examples={"mascarado": {"value": "27.865.757/0001-02"}},
|
|
),
|
|
) -> Dict[str, Any]:
|
|
normalized = _normalize_cnpj(cnpj)
|
|
if len(normalized) != 14:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="CNPJ deve conter 14 dígitos numéricos.",
|
|
)
|
|
|
|
url = _CNPJ_API_TEMPLATE.format(cnpj=normalized)
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
external_response = await client.get(url, headers={"Accept": "application/json"})
|
|
except httpx.RequestError as exc: # pragma: no cover - proteção contra falhas de rede reais
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail="Serviço externo indisponível",
|
|
) from exc
|
|
|
|
status_code = external_response.status_code
|
|
try:
|
|
data = external_response.json()
|
|
except ValueError:
|
|
data = None
|
|
|
|
if status_code in {status.HTTP_429_TOO_MANY_REQUESTS} or status_code >= 500:
|
|
message = ""
|
|
if isinstance(data, dict) and isinstance(data.get("message"), str):
|
|
message = f": {data['message']}"
|
|
raise HTTPException(
|
|
status_code=status_code,
|
|
detail=f"Serviço externo indisponível{message}",
|
|
)
|
|
|
|
if status_code == status.HTTP_404_NOT_FOUND:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="CNPJ não encontrado")
|
|
|
|
if isinstance(data, dict) and data.get("status") == "ERROR":
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="CNPJ não encontrado")
|
|
|
|
if status_code != status.HTTP_200_OK or not isinstance(data, dict):
|
|
detail_message = "Erro ao consultar CNPJ."
|
|
if isinstance(data, dict) and isinstance(data.get("message"), str):
|
|
detail_message = data["message"]
|
|
raise HTTPException(status_code=status_code, detail=detail_message)
|
|
|
|
return data
|