518 lines
12 KiB
Python
518 lines
12 KiB
Python
from enum import Enum
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from Models.config import Config
|
|
from Models.empresa import Empresa
|
|
from Models.nfe import NFe
|
|
from Models.nfse import NFSe
|
|
from Models.servico import Servico
|
|
from Models.tomador import Tomador
|
|
|
|
|
|
class Fila(str, Enum):
|
|
NFE = "nfe"
|
|
NFSE = "nfse"
|
|
CERTIFICADO_DIGITAL = "certificado_digital"
|
|
EMPRESA = "empresa"
|
|
SERVICO = "servico"
|
|
TOMADOR = "tomador"
|
|
RELATORIOS = "relatorios"
|
|
WEBHOOKS = "webhooks"
|
|
|
|
|
|
class NFeMetodo(str, Enum):
|
|
POST = "post"
|
|
GET = "get"
|
|
DOWNLOAD = "download"
|
|
DOWNLOAD_XML = "download_xml"
|
|
CANCELAR = "cancelar"
|
|
STATUS_CANCELAMENTO = "status_cancelamento"
|
|
CCE = "cce"
|
|
STATUS_CCE = "status_cce"
|
|
RESUMO = "resumo"
|
|
|
|
|
|
class NFSeMetodo(str, Enum):
|
|
POST = "post"
|
|
GET = "get"
|
|
DOWNLOAD = "download"
|
|
DOWNLOAD_XML = "download_xml"
|
|
CANCELAR = "cancelar"
|
|
STATUS_CANCELAMENTO = "status_cancelamento"
|
|
RESUMO = "resumo"
|
|
HISTORICO = "historico"
|
|
|
|
|
|
class CertificadoMetodo(str, Enum):
|
|
POST = "post"
|
|
GET = "get"
|
|
PUT = "put"
|
|
|
|
|
|
class EmpresaMetodo(str, Enum):
|
|
POST = "post"
|
|
GET = "get"
|
|
POST_FILE = "post_file"
|
|
PATCH = "patch"
|
|
|
|
|
|
class ServicoMetodo(str, Enum):
|
|
POST = "post"
|
|
GET = "get"
|
|
PUT = "put"
|
|
DELETE = "delete"
|
|
|
|
|
|
class TomadorMetodo(str, Enum):
|
|
POST = "post"
|
|
GET = "get"
|
|
PUT = "put"
|
|
DELETE = "delete"
|
|
|
|
|
|
class RelatorioMetodo(str, Enum):
|
|
NFE_EMITIDAS = "nfe_emitidas"
|
|
NFE_POR_CNPJ = "nfe_por_cnpj"
|
|
NFSE_EMITIDAS = "nfse_emitidas"
|
|
NFSE_POR_CNPJ = "nfse_por_cnpj"
|
|
|
|
|
|
class WebhookMetodo(str, Enum):
|
|
GET = "get"
|
|
POST = "post"
|
|
DELETE = "delete"
|
|
PUT = "put"
|
|
TEST = "test"
|
|
|
|
|
|
class BaseEnvelope(BaseModel):
|
|
cliente: str | None = None
|
|
config: Config
|
|
url_base: str | None = None
|
|
|
|
|
|
# --- Base Data Models ---
|
|
|
|
class BaseDados(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
id_usuario: int | None = 1
|
|
|
|
|
|
class NFePostDadosWrapped(BaseDados):
|
|
nfe: NFe
|
|
id_empresa: str | None = None
|
|
|
|
|
|
class NFePostDadosDirect(NFe):
|
|
id_usuario: int | None = 1
|
|
id_empresa: str | None = None
|
|
|
|
|
|
class NFeIdDados(BaseDados):
|
|
id: str
|
|
|
|
|
|
class NFeCancelarDados(NFeIdDados):
|
|
justificativa: str | None = None
|
|
|
|
|
|
class NFeCCEDados(NFeIdDados):
|
|
correcao: str
|
|
|
|
|
|
class NFeResumoDados(BaseDados):
|
|
cnpj: str
|
|
id_integracao: str | None = None
|
|
id: str | None = None
|
|
|
|
|
|
class NFSePostDados(NFSe):
|
|
id_usuario: int | None = 1
|
|
id_empresa: str | None = None
|
|
|
|
|
|
class NFSeIdDados(BaseDados):
|
|
id: str
|
|
|
|
|
|
class NFSeGetDados(BaseDados):
|
|
id: str | None = None
|
|
|
|
|
|
class NFSeResumoDados(NFSeIdDados):
|
|
cnpj: str
|
|
|
|
|
|
class NFSeHistoricoDados(NFSeIdDados):
|
|
cnpj: str | None = None
|
|
|
|
|
|
class CertificadoPostDados(BaseDados):
|
|
arquivo: str
|
|
senha: str
|
|
id_empresa: str | None = None
|
|
email: str | None = None
|
|
expiration_at: str | None = Field(None, alias="expirationAt")
|
|
data_expiracao: str | None = None
|
|
|
|
|
|
class CertificadoGetDados(BaseDados):
|
|
id: str | None = None
|
|
|
|
|
|
class CertificadoPutDados(CertificadoPostDados):
|
|
id: str
|
|
|
|
|
|
class EmpresaPostDados(Empresa):
|
|
id_usuario: int | None = 1
|
|
|
|
|
|
class EmpresaGetDados(BaseDados):
|
|
cnpj: str | None = None
|
|
|
|
|
|
class EmpresaPostFileDados(BaseDados):
|
|
file: str
|
|
cnpj: str
|
|
|
|
|
|
class EmpresaPatchDados(BaseDados):
|
|
cnpj: str
|
|
certificado: str
|
|
|
|
|
|
class ServicoPostDados(Servico):
|
|
id_usuario: int | None = 1
|
|
|
|
|
|
class ServicoGetDados(BaseDados):
|
|
id: str | None = None
|
|
|
|
|
|
class ServicoPutDados(Servico):
|
|
id: str
|
|
id_usuario: int | None = 1
|
|
|
|
|
|
class ServicoDeleteDados(BaseDados):
|
|
id: str
|
|
|
|
|
|
class TomadorPostDados(Tomador):
|
|
id_usuario: int | None = 1
|
|
|
|
|
|
class TomadorGetDados(BaseDados):
|
|
cpf_cnpj: str | None = None
|
|
|
|
|
|
class TomadorPutDados(Tomador):
|
|
cpf_cnpj: str
|
|
id_usuario: int | None = 1
|
|
|
|
|
|
class TomadorDeleteDados(BaseDados):
|
|
cpf_cnpj: str
|
|
|
|
|
|
class RelatorioVazioDados(BaseDados):
|
|
pass
|
|
|
|
|
|
class RelatorioCnpjDados(BaseDados):
|
|
cnpj: str
|
|
|
|
|
|
class RelatorioPeriodoDados(BaseDados):
|
|
data_inicial: str | None = None
|
|
data_final: str | None = None
|
|
|
|
|
|
class RelatorioCnpjPeriodoDados(RelatorioPeriodoDados):
|
|
cnpj: str
|
|
|
|
|
|
class WebhookGetDeleteTestDados(BaseDados):
|
|
cnpj: str
|
|
|
|
|
|
class WebhookPostPutDados(WebhookGetDeleteTestDados):
|
|
url: str
|
|
|
|
|
|
# --- Envelopes ---
|
|
|
|
class NFePostEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFE
|
|
metodo: NFeMetodo = NFeMetodo.POST
|
|
dados: NFePostDadosDirect | NFePostDadosWrapped
|
|
|
|
|
|
class NFeIdEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFE
|
|
metodo: NFeMetodo
|
|
dados: NFeIdDados
|
|
|
|
|
|
class NFeCancelarEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFE
|
|
metodo: NFeMetodo = NFeMetodo.CANCELAR
|
|
dados: NFeCancelarDados
|
|
|
|
|
|
class NFeCCEEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFE
|
|
metodo: NFeMetodo = NFeMetodo.CCE
|
|
dados: NFeCCEDados
|
|
|
|
|
|
class NFeResumoEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFE
|
|
metodo: NFeMetodo = NFeMetodo.RESUMO
|
|
dados: NFeResumoDados
|
|
|
|
|
|
class NFSePostEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFSE
|
|
metodo: NFSeMetodo = NFSeMetodo.POST
|
|
dados: NFSePostDados
|
|
|
|
|
|
class NFSeGetEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFSE
|
|
metodo: NFSeMetodo = NFSeMetodo.GET
|
|
dados: NFSeGetDados
|
|
|
|
|
|
class NFSeIdEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFSE
|
|
metodo: NFSeMetodo
|
|
dados: NFSeIdDados
|
|
|
|
|
|
class NFSeResumoEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFSE
|
|
metodo: NFSeMetodo = NFSeMetodo.RESUMO
|
|
dados: NFSeResumoDados
|
|
|
|
|
|
class NFSeHistoricoEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.NFSE
|
|
metodo: NFSeMetodo = NFSeMetodo.HISTORICO
|
|
dados: NFSeHistoricoDados
|
|
|
|
|
|
class CertificadoPostEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.CERTIFICADO_DIGITAL
|
|
metodo: CertificadoMetodo = CertificadoMetodo.POST
|
|
dados: CertificadoPostDados
|
|
|
|
|
|
class CertificadoGetEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.CERTIFICADO_DIGITAL
|
|
metodo: CertificadoMetodo = CertificadoMetodo.GET
|
|
dados: CertificadoGetDados
|
|
|
|
|
|
class CertificadoPutEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.CERTIFICADO_DIGITAL
|
|
metodo: CertificadoMetodo = CertificadoMetodo.PUT
|
|
dados: CertificadoPutDados
|
|
|
|
|
|
class EmpresaPostEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.EMPRESA
|
|
metodo: EmpresaMetodo = EmpresaMetodo.POST
|
|
dados: EmpresaPostDados
|
|
|
|
|
|
class EmpresaGetEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.EMPRESA
|
|
metodo: EmpresaMetodo = EmpresaMetodo.GET
|
|
dados: EmpresaGetDados
|
|
|
|
|
|
class EmpresaPostFileEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.EMPRESA
|
|
metodo: EmpresaMetodo = EmpresaMetodo.POST_FILE
|
|
dados: EmpresaPostFileDados
|
|
|
|
|
|
class EmpresaPatchEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.EMPRESA
|
|
metodo: EmpresaMetodo = EmpresaMetodo.PATCH
|
|
dados: EmpresaPatchDados
|
|
|
|
|
|
class ServicoPostEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.SERVICO
|
|
metodo: ServicoMetodo = ServicoMetodo.POST
|
|
dados: ServicoPostDados
|
|
|
|
|
|
class ServicoGetEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.SERVICO
|
|
metodo: ServicoMetodo = ServicoMetodo.GET
|
|
dados: ServicoGetDados
|
|
|
|
|
|
class ServicoPutEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.SERVICO
|
|
metodo: ServicoMetodo = ServicoMetodo.PUT
|
|
dados: ServicoPutDados
|
|
|
|
|
|
class ServicoDeleteEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.SERVICO
|
|
metodo: ServicoMetodo = ServicoMetodo.DELETE
|
|
dados: ServicoDeleteDados
|
|
|
|
|
|
class TomadorPostEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.TOMADOR
|
|
metodo: TomadorMetodo = TomadorMetodo.POST
|
|
dados: TomadorPostDados
|
|
|
|
|
|
class TomadorGetEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.TOMADOR
|
|
metodo: TomadorMetodo = TomadorMetodo.GET
|
|
dados: TomadorGetDados
|
|
|
|
|
|
class TomadorPutEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.TOMADOR
|
|
metodo: TomadorMetodo = TomadorMetodo.PUT
|
|
dados: TomadorPutDados
|
|
|
|
|
|
class TomadorDeleteEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.TOMADOR
|
|
metodo: TomadorMetodo = TomadorMetodo.DELETE
|
|
dados: TomadorDeleteDados
|
|
|
|
|
|
class RelatorioNFeEmitidasEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.RELATORIOS
|
|
metodo: RelatorioMetodo = RelatorioMetodo.NFE_EMITIDAS
|
|
dados: RelatorioVazioDados
|
|
|
|
|
|
class RelatorioNFeCnpjEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.RELATORIOS
|
|
metodo: RelatorioMetodo = RelatorioMetodo.NFE_POR_CNPJ
|
|
dados: RelatorioCnpjDados
|
|
|
|
|
|
class RelatorioNFSeEmitidasEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.RELATORIOS
|
|
metodo: RelatorioMetodo = RelatorioMetodo.NFSE_EMITIDAS
|
|
dados: RelatorioPeriodoDados
|
|
|
|
|
|
class RelatorioNFSeCnpjEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.RELATORIOS
|
|
metodo: RelatorioMetodo = RelatorioMetodo.NFSE_POR_CNPJ
|
|
dados: RelatorioCnpjPeriodoDados
|
|
|
|
|
|
class WebhookGetDeleteTestEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.WEBHOOKS
|
|
metodo: WebhookMetodo
|
|
dados: WebhookGetDeleteTestDados
|
|
|
|
|
|
class WebhookPostPutEnvelope(BaseEnvelope):
|
|
fila: Fila = Fila.WEBHOOKS
|
|
metodo: WebhookMetodo
|
|
dados: WebhookPostPutDados
|
|
|
|
|
|
# --- Generic Envelope for basic structural validation ---
|
|
|
|
class GenericEnvelope(BaseEnvelope):
|
|
"""Validates that minimum required fields exist before routing."""
|
|
fila: Fila
|
|
metodo: str
|
|
dados: dict | None = None
|
|
|
|
|
|
# --- Payload Validation Router ---
|
|
|
|
# Maps (fila, metodo) -> Envelope class
|
|
_ENVELOPE_MAP: dict[tuple[str, str], type[BaseEnvelope]] = {
|
|
# NFe
|
|
("nfe", "post"): NFePostEnvelope,
|
|
("nfe", "get"): NFeIdEnvelope,
|
|
("nfe", "download"): NFeIdEnvelope,
|
|
("nfe", "download_xml"): NFeIdEnvelope,
|
|
("nfe", "cancelar"): NFeCancelarEnvelope,
|
|
("nfe", "status_cancelamento"): NFeIdEnvelope,
|
|
("nfe", "cce"): NFeCCEEnvelope,
|
|
("nfe", "status_cce"): NFeIdEnvelope,
|
|
("nfe", "resumo"): NFeResumoEnvelope,
|
|
# NFSe
|
|
("nfse", "post"): NFSePostEnvelope,
|
|
("nfse", "get"): NFSeGetEnvelope,
|
|
("nfse", "download"): NFSeIdEnvelope,
|
|
("nfse", "download_xml"): NFSeIdEnvelope,
|
|
("nfse", "cancelar"): NFSeIdEnvelope,
|
|
("nfse", "status_cancelamento"): NFSeIdEnvelope,
|
|
("nfse", "resumo"): NFSeResumoEnvelope,
|
|
("nfse", "historico"): NFSeHistoricoEnvelope,
|
|
# Certificado Digital
|
|
("certificado_digital", "post"): CertificadoPostEnvelope,
|
|
("certificado_digital", "get"): CertificadoGetEnvelope,
|
|
("certificado_digital", "put"): CertificadoPutEnvelope,
|
|
# Empresa
|
|
("empresa", "post"): EmpresaPostEnvelope,
|
|
("empresa", "get"): EmpresaGetEnvelope,
|
|
("empresa", "post_file"): EmpresaPostFileEnvelope,
|
|
("empresa", "patch"): EmpresaPatchEnvelope,
|
|
# Serviço
|
|
("servico", "post"): ServicoPostEnvelope,
|
|
("servico", "get"): ServicoGetEnvelope,
|
|
("servico", "put"): ServicoPutEnvelope,
|
|
("servico", "delete"): ServicoDeleteEnvelope,
|
|
# Tomador
|
|
("tomador", "post"): TomadorPostEnvelope,
|
|
("tomador", "get"): TomadorGetEnvelope,
|
|
("tomador", "put"): TomadorPutEnvelope,
|
|
("tomador", "delete"): TomadorDeleteEnvelope,
|
|
# Relatórios
|
|
("relatorios", "nfe_emitidas"): RelatorioNFeEmitidasEnvelope,
|
|
("relatorios", "nfe_por_cnpj"): RelatorioNFeCnpjEnvelope,
|
|
("relatorios", "nfse_emitidas"): RelatorioNFSeEmitidasEnvelope,
|
|
("relatorios", "nfse_por_cnpj"): RelatorioNFSeCnpjEnvelope,
|
|
# Webhooks
|
|
("webhooks", "get"): WebhookGetDeleteTestEnvelope,
|
|
("webhooks", "post"): WebhookPostPutEnvelope,
|
|
("webhooks", "delete"): WebhookGetDeleteTestEnvelope,
|
|
("webhooks", "put"): WebhookPostPutEnvelope,
|
|
("webhooks", "test"): WebhookGetDeleteTestEnvelope,
|
|
}
|
|
|
|
|
|
def validate_payload(payload_dict: dict) -> BaseEnvelope:
|
|
# Step 1: Basic structural validation
|
|
generic = GenericEnvelope(**payload_dict)
|
|
|
|
fila = generic.fila.value
|
|
metodo = generic.metodo.lower()
|
|
|
|
# Step 2: Find specific envelope
|
|
key = (fila, metodo)
|
|
envelope_cls = _ENVELOPE_MAP.get(key)
|
|
|
|
if envelope_cls is None:
|
|
raise ValueError(f"Combinação fila/metodo desconhecida: {fila}/{metodo}")
|
|
|
|
# Step 3: Normalize metodo in the dict to match enum lowercase values
|
|
payload_dict['metodo'] = metodo
|
|
|
|
# Step 4: Full typed validation
|
|
return envelope_cls(**payload_dict)
|
|
|
|
|
|
|