127 lines
No EOL
4.5 KiB
Python
127 lines
No EOL
4.5 KiB
Python
|
|
__version__ = "1.0.0 - 18/02/2025" # VERSAO PARA TODOS OS SISTEMAS TRANSMITINDO PARA O SERVIÇO LOGGER_CONTROLLER DO SG_IC2
|
|
__version__ = "1.0.1 - 15/01/2026 - S.D.G. - Ajustes"
|
|
__version__ = "1.0.2 - 23/01/2026 - S.D.G. - ForwardCompatibleModel & SerializeAsAny"
|
|
from pydantic import BaseModel, field_validator, ValidationInfo, Field, ConfigDict, SerializeAsAny
|
|
from enum import StrEnum
|
|
from typing import Union, Any, Optional
|
|
|
|
class ForwardCompatibleModel(BaseModel):
|
|
"""
|
|
Base para todos os modelos que permite campos extras.
|
|
O Pydantic V2 gerencia automaticamente o dicionário 'model_extra'.
|
|
"""
|
|
model_config = ConfigDict(extra='allow')
|
|
|
|
class RealmType(StrEnum):
|
|
LOGS = "LOGS"
|
|
METRICS = "METRICS"
|
|
STATUS = "STATUS"
|
|
ERRORS = "ERRORS"
|
|
OPERATIONAL = "OPERATIONAL"
|
|
|
|
class OpStatus(StrEnum):
|
|
ACTIVE = "ACTIVE"
|
|
OK = "OK"
|
|
ERROR = "ERROR"
|
|
|
|
class ICLogs(ForwardCompatibleModel):
|
|
log_date : str
|
|
service_date : str
|
|
service : str
|
|
versionstr : str
|
|
origin : str
|
|
enterprise_id : str
|
|
enterprise_name : str
|
|
message : str
|
|
tenant : Optional[str] = Field(default='#', description="Tenant ou '#' para global")
|
|
|
|
class ICErrors(ForwardCompatibleModel):
|
|
log_date : str
|
|
service_date : str
|
|
service : str
|
|
versionstr : str
|
|
origin : str
|
|
enterprise_id : str
|
|
enterprise_name : str
|
|
message : str
|
|
tenant : Optional[str] = Field(default='#', description="Tenant ou '#' para global")
|
|
|
|
class ICMetrics(ForwardCompatibleModel):
|
|
metric_date : str
|
|
metric : str
|
|
metric_name : str
|
|
value : str
|
|
unit : str
|
|
tenant : Optional[str] = Field(default='#', description="Tenant ou '#' para global")
|
|
|
|
class ICOperational(ForwardCompatibleModel):
|
|
log_date : str # YYYY-MM-DD HH:MM:S
|
|
service_date : str # YYYY-MM-DD HH:MM:S
|
|
service_name : str
|
|
versionstr : str
|
|
origin : str
|
|
enterprise_id : str
|
|
enterprise_name : str
|
|
status : OpStatus
|
|
step_status : str
|
|
message : str
|
|
|
|
class ICStatus(ForwardCompatibleModel):
|
|
service : str
|
|
versionstr : str
|
|
system_name : str
|
|
startup_date : str
|
|
access_date : str
|
|
origin : str
|
|
sid : Optional[str] = ""
|
|
task_count : Optional[int] = 0
|
|
cpu : Optional[float] = 0.0
|
|
memory : Optional[float] = 0.0
|
|
disk : Optional[float] = 0.0
|
|
network : Optional[float] = 0.0
|
|
|
|
# Modelo polimórfico gerador de metadados JSON para o Exchange de Logs do Sistema Inteirador
|
|
class InfoConnector(ForwardCompatibleModel):
|
|
realm: RealmType
|
|
|
|
# SerializeAsAny garante que se 'messageIC' for uma subclasse (como 'Operation'),
|
|
# os campos extras da subclasse (como 'reserva_id') aparecerão no JSON final.
|
|
messageIC: SerializeAsAny[Union[ICLogs, ICErrors, ICMetrics, ICStatus, ICOperational]]
|
|
|
|
@field_validator('messageIC', mode='before')
|
|
@classmethod
|
|
def validate_schema(cls, v: Any, info: ValidationInfo) -> Any:
|
|
realm = info.data.get("realm")
|
|
if realm is None:
|
|
# Se realm não existe, o próprio Pydantic lançará erro de validação do campo 'realm'
|
|
return v
|
|
|
|
# Mapeamento de Realm para Classe Esperada
|
|
target_map = {
|
|
RealmType.LOGS : ICLogs,
|
|
RealmType.ERRORS : ICErrors,
|
|
RealmType.METRICS : ICMetrics,
|
|
RealmType.STATUS : ICStatus,
|
|
RealmType.OPERATIONAL: ICOperational,
|
|
}
|
|
|
|
expected_class = target_map.get(realm)
|
|
if not expected_class:
|
|
raise ValueError(f"Invalid realm: {realm}")
|
|
|
|
# Se for um dicionário (vindo de um JSON), validamos contra a classe mapeada
|
|
if isinstance(v, dict):
|
|
return expected_class.model_validate(v)
|
|
|
|
# Se já for um objeto instanciado (como no seu Builder), verificamos a herança
|
|
if not isinstance(v, expected_class):
|
|
raise ValueError(f"Expected {expected_class.__name__} for realm {realm}, got {type(v).__name__}")
|
|
|
|
return v
|
|
|
|
model_config = ConfigDict(
|
|
extra="allow",
|
|
frozen=False,
|
|
str_strip_whitespace=True
|
|
) |