Módulo de Agravos - Documentação Técnica

O módulo de agravos implementa um sistema extensível para gerenciamento de diferentes tipos de agravos de saúde (dengue, zika, chikungunya, etc.) seguindo a arquitetura híbrida definida na ADR-006.

Arquitetura

Estrutura de Diretórios

modules/agravos/
├── __init__.py                 # Exporta AgravosModule
├── module_config.py            # Configuração e carregamento dinâmico
├── async_repository.py         # Repositório assíncrono para operações de BD
├── core/                       # Núcleo do módulo
│   ├── __init__.py
│   ├── base.py                # Classe abstrata BaseAgravo
│   ├── models.py              # Modelos SQLAlchemy
│   ├── registry.py            # Sistema de registro dinâmico
│   └── schemas.py             # Schemas Pydantic compartilhados
├── endpoints/                  # Endpoints REST
│   ├── __init__.py
│   └── casos.py               # CRUD de casos
└── tipos/                      # Implementações específicas
    ├── __init__.py
    ├── dengue.py              # Implementação para dengue
    ├── zika.py                # Implementação para zika
    └── chikungunya.py         # Implementação para chikungunya

Componentes Principais

1. Sistema de Registro (Registry Pattern)

O sistema usa um padrão Registry para permitir adição dinâmica de novos tipos de agravos:

# core/registry.py
from typing import Dict, Type, Optional, List
from abc import ABC

class AgravoRegistry:
    _registry: Dict[str, Type['BaseAgravo']] = {}
    
    @classmethod
    def register(cls, agravo_class: Type['BaseAgravo']) -> None:
        """Registra um novo tipo de agravo"""
        instance = agravo_class()
        codigo = instance.codigo
        cls._registry[codigo] = agravo_class
    
    @classmethod
    def get(cls, codigo: str) -> Optional[Type['BaseAgravo']]:
        """Obtém classe de agravo por código"""
        return cls._registry.get(codigo)
    
    @classmethod
    def list_all(cls) -> List[str]:
        """Lista todos os códigos registrados"""
        return list(cls._registry.keys())

# Decorator para auto-registro
def registrar_agravo(cls):
    AgravoRegistry.register(cls)
    return cls

2. Classe Base Abstrata

# core/base.py
from abc import ABC, abstractmethod
from typing import Dict, Any, List

class BaseAgravo(ABC):
    @property
    @abstractmethod
    def codigo(self) -> str:
        """Código único do agravo"""
        pass
    
    @property
    @abstractmethod
    def nome(self) -> str:
        """Nome do agravo"""
        pass
    
    @property
    @abstractmethod
    def cid10_codes(self) -> List[str]:
        """Códigos CID-10 associados"""
        pass
    
    @abstractmethod
    def validar_dados_especificos(self, dados: Dict[str, Any]) -> List[str]:
        """Valida dados específicos do agravo"""
        pass
    
    def processar_notificacao(self, dados: Dict[str, Any]) -> Dict[str, Any]:
        """Processa dados antes de salvar"""
        return dados

3. Modelos de Banco de Dados

# core/models.py
from sqlalchemy import Column, Integer, String, JSON, ForeignKey
from sqlalchemy.orm import relationship
from core.database import Base

class AgravoModel(Base):
    __tablename__ = "agravos"
    
    id = Column(Integer, primary_key=True)
    codigo = Column(String(50), unique=True, nullable=False)
    nome = Column(String(100), nullable=False)
    descricao = Column(String(500))
    cid10_codes = Column(JSON, default=list)
    notificacao_obrigatoria = Column(Boolean, default=True)
    ativo = Column(Boolean, default=True)
    
    # Relacionamentos
    casos = relationship("CasoAgravoModel", back_populates="agravo")

class CasoAgravoModel(Base):
    __tablename__ = "casos_agravos"
    
    id = Column(Integer, primary_key=True)
    agravo_id = Column(Integer, ForeignKey("agravos.id"))
    numero_notificacao = Column(String(50), unique=True)
    
    # Dados do paciente (anonimizados em produção)
    paciente_nome = Column(String(200))
    paciente_idade = Column(Integer)
    paciente_sexo = Column(String(1))
    
    # Localização
    municipio_residencia = Column(String(100))
    estado_residencia = Column(String(2))
    
    # Dados clínicos genéricos
    data_primeiros_sintomas = Column(Date)
    data_notificacao = Column(DateTime, default=func.now())
    classificacao_final = Column(String(50))
    evolucao = Column(String(50))
    
    # Dados específicos do agravo (JSON)
    dados_especificos = Column(JSON, default=dict)
    
    # Relacionamentos
    agravo = relationship("AgravoModel", back_populates="casos")
    exames = relationship("ExameAgravoModel", back_populates="caso")

Fluxo de Dados

1. Inicialização

Carregando diagrama...

2. Criação de Caso

Carregando diagrama...

3. Consultas

# async_repository.py
class AgravoRepository:
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def listar_casos(
        self,
        filtros: Dict[str, Any],
        paginacao: PaginacaoSchema
    ) -> Page[CasoAgravoSchema]:
        query = select(CasoAgravoModel).options(
            selectinload(CasoAgravoModel.agravo),
            selectinload(CasoAgravoModel.exames)
        )
        
        # Aplicar filtros
        if filtros.get("agravo_codigo"):
            query = query.join(AgravoModel).where(
                AgravoModel.codigo == filtros["agravo_codigo"]
            )
        
        if filtros.get("municipio"):
            query = query.where(
                CasoAgravoModel.municipio_residencia == filtros["municipio"]
            )
        
        # Executar com paginação
        return await paginate(self.db, query, paginacao)

API Endpoints

Públicos (Sem Autenticação)

GET /api/v1/agravos/tipos

Lista todos os tipos de agravos disponíveis no sistema.

curl https://api.sinapse.org.br/v1/agravos/tipos

Resposta:

{
  "total": 3,
  "tipos": [
    {
      "codigo": "dengue",
      "nome": "Dengue",
      "descricao": "Doença febril aguda causada por arbovírus",
      "cid10_codes": ["A90", "A91"],
      "notificacao_obrigatoria": true,
      "campos_especificos": {
        "classificacao_final": {
          "tipo": "select",
          "opcoes": ["dengue_classico", "dengue_com_sinais_alarme", "dengue_grave"],
          "obrigatorio": true
        },
        "sorotipo": {
          "tipo": "select",
          "opcoes": ["DENV-1", "DENV-2", "DENV-3", "DENV-4", "indeterminado"],
          "obrigatorio": false
        }
      }
    }
  ]
}

Autenticados

GET /api/v1/agravos/casos/

Permissão requerida: agravos.listar

Query Parameters:

Parâmetro	Tipo	Descrição
`page`	int	Página (default: 1)
`size`	int	Itens por página (max: 100)
`agravo_codigo`	string	Filtrar por tipo
`municipio_residencia`	string	Filtrar por município
`estado_residencia`	string	UF (2 letras)
`data_inicio`	date	Data inicial ISO
`data_fim`	date	Data final ISO
`classificacao_final`	string	Status do caso

POST /api/v1/agravos/casos/

Permissão requerida: agravos.criar

Request Body:

{
  "agravo_codigo": "dengue",
  "numero_notificacao": "2025-001234",
  "paciente_nome": "João Silva",
  "paciente_idade": 35,
  "paciente_sexo": "M",
  "municipio_residencia": "São Paulo",
  "estado_residencia": "SP",
  "data_primeiros_sintomas": "2025-07-20",
  "sintomas": {
    "febre": true,
    "cefaleia": true,
    "mialgia": true,
    "artralgia": false,
    "exantema": false,
    "nausea": true
  },
  "dados_especificos": {
    "classificacao_inicial": "dengue_classico",
    "sinais_alarme": false,
    "comorbidades": ["hipertensao", "diabetes"]
  }
}

Extensibilidade

Adicionando Novo Tipo de Agravo

1. Criar Implementação

# tipos/covid19.py
from ..core.registry import registrar_agravo
from ..core.base import BaseAgravo
from typing import Dict, Any, List

@registrar_agravo
class Covid19Agravo(BaseAgravo):
    @property
    def codigo(self) -> str:
        return "covid19"
    
    @property
    def nome(self) -> str:
        return "COVID-19"
    
    @property
    def cid10_codes(self) -> List[str]:
        return ["U07.1", "U07.2"]
    
    def validar_dados_especificos(self, dados: Dict[str, Any]) -> List[str]:
        erros = []
        
        # Validações específicas para COVID-19
        if dados.get("vacinacao"):
            doses = dados["vacinacao"].get("doses", 0)
            if not isinstance(doses, int) or doses < 0:
                erros.append("Número de doses deve ser inteiro não negativo")
        
        if dados.get("teste_realizado"):
            tipo_teste = dados.get("tipo_teste")
            if tipo_teste not in ["RT-PCR", "Antigeno", "Sorologia"]:
                erros.append("Tipo de teste inválido")
        
        return erros
    
    def processar_notificacao(self, dados: Dict[str, Any]) -> Dict[str, Any]:
        # Calcular gravidade baseado em fatores de risco
        fatores_risco = dados.get("fatores_risco", [])
        idade = dados.get("paciente_idade", 0)
        
        if idade > 60 or len(fatores_risco) > 2:
            dados["grupo_risco"] = "alto"
        elif idade > 40 or len(fatores_risco) > 0:
            dados["grupo_risco"] = "moderado"
        else:
            dados["grupo_risco"] = "baixo"
        
        return dados

2. Importar no module_config.py

# module_config.py
def _load_agravos(self):
    """Carrega todos os tipos de agravos"""
    from .tipos import dengue, zika, chikungunya, covid19  # Adicionar aqui
    logger.info(f"Agravos carregados: {AgravoRegistry.list_all()}")

3. Adicionar ao Banco de Dados

-- Migration ou script SQL
INSERT INTO agravos (codigo, nome, descricao, cid10_codes, notificacao_obrigatoria) 
VALUES (
    'covid19', 
    'COVID-19', 
    'Doença respiratória causada pelo coronavírus SARS-CoV-2',
    '["U07.1", "U07.2"]',
    true
);

Configuração

Variáveis de Ambiente

# Habilitar módulo
ENABLED_MODULES=auth,usuarios,agravos  # ou "all"

# Configurações específicas do módulo
AGRAVOS_CACHE_TTL=3600  # Cache de tipos em segundos
AGRAVOS_MAX_BULK_SIZE=1000  # Máximo de registros por importação
AGRAVOS_ANONIMIZAR_PRODUCAO=true  # Anonimizar dados em produção

Permissões do Módulo

Permissão	Descrição
`agravos.listar`	Visualizar casos
`agravos.criar`	Criar novos casos
`agravos.editar`	Editar casos existentes
`agravos.deletar`	Remover casos
`agravos.exportar`	Exportar dados
`agravos.importar`	Importação em massa
`agravos.estatisticas`	Ver estatísticas agregadas

Migrações

Aplicar Migrações

# Desenvolvimento local
alembic upgrade head

# Docker
docker-compose exec api alembic upgrade head

# Kubernetes
kubectl exec -it deployment/api -- alembic upgrade head

Criar Nova Migração

# Gerar automaticamente
alembic revision --autogenerate -m "adicionar_campo_agravo"

# Manual
alembic revision -m "ajuste_indices_agravos"

Rollback

# Voltar uma migração
alembic downgrade -1

# Voltar para revisão específica
alembic downgrade abc123def456

Testes

Estrutura de Testes

tests/modules/agravos/
├── __init__.py
├── conftest.py              # Fixtures específicas
├── test_registry.py         # Testes do sistema de registro
├── test_tipos.py           # Testes dos tipos de agravos
├── test_endpoints.py       # Testes de API
└── test_repository.py      # Testes de persistência

Executar Testes

# Todos os testes do módulo
pytest modules/agravos/tests/ -v

# Teste específico
pytest modules/agravos/tests/test_registry.py::test_registro_dinamico

# Com cobertura
pytest modules/agravos/tests/ --cov=modules.agravos --cov-report=html

Exemplo de Teste

# test_tipos.py
import pytest
from modules.agravos.tipos.dengue import DengueAgravo

class TestDengueAgravo:
    def test_validacao_dados_especificos(self):
        agravo = DengueAgravo()
        
        # Dados válidos
        dados_validos = {
            "classificacao_final": "dengue_classico",
            "sorotipo": "DENV-2"
        }
        assert agravo.validar_dados_especificos(dados_validos) == []
        
        # Dados inválidos
        dados_invalidos = {
            "classificacao_final": "tipo_invalido"
        }
        erros = agravo.validar_dados_especificos(dados_invalidos)
        assert len(erros) > 0
        assert "classificacao_final" in erros[0]

Performance

Índices de Banco

-- Índices para queries comuns
CREATE INDEX idx_casos_agravo_data ON casos_agravos(agravo_id, data_notificacao);
CREATE INDEX idx_casos_municipio_data ON casos_agravos(municipio_residencia, data_notificacao);
CREATE INDEX idx_casos_classificacao ON casos_agravos(classificacao_final);
CREATE INDEX idx_casos_numero_notif ON casos_agravos(numero_notificacao);

-- Índice para dados JSON (PostgreSQL)
CREATE INDEX idx_casos_dados_gin ON casos_agravos USING gin(dados_especificos);

Otimizações Implementadas

Paginação obrigatória - Máximo 100 registros por página
Eager loading - Usa selectinload() para evitar N+1
Cache de tipos - Redis cache para lista de agravos
Índices compostos - Para queries mais comuns
Particionamento - Tabelas particionadas por ano (produção)

Cache Strategy

from core.cache import cache

class AgravoService:
    @cache(ttl=3600, key="agravos:tipos")
    async def listar_tipos(self) -> List[AgravoTipoSchema]:
        # Busca do banco apenas se não estiver em cache
        return await self.repository.listar_tipos()
    
    async def criar_caso(self, dados: dict) -> CasoAgravoSchema:
        caso = await self.repository.criar(dados)
        # Invalida cache de estatísticas
        await cache.delete_pattern("agravos:stats:*")
        return caso

Troubleshooting

Problema: Módulo não carrega

Sintomas:

Endpoints retornam 404
Logs não mostram carregamento do módulo

Soluções:

Verificar ENABLED_MODULES no .env
Verificar se módulo está em modules/__init__.py
Checar logs de erro: docker logs sinapse_api | grep -i error

Problema: "expected str instance, property found"

Causa: Registry tentando acessar property sem instanciar classe

Solução aplicada:

# Em registry.py
def register(cls, agravo_class):
    instance = agravo_class()  # Instanciar temporariamente
    codigo = instance.codigo    # Acessar property
    cls._registry[codigo] = agravo_class  # Guardar classe

Problema: Tabelas não criadas

Soluções:

Verificar se modelos estão em models/__all_models.py

Gerar nova migração:

alembic revision --autogenerate -m "criar_tabelas_agravos"

Aplicar migração:
```
alembic upgrade head
```

Problema: Performance degradada

Diagnóstico:

-- Queries lentas
SELECT query, calls, mean_exec_time
FROM pg_stat_statements
WHERE query LIKE '%casos_agravos%'
ORDER BY mean_exec_time DESC
LIMIT 10;

-- Índices faltando
SELECT schemaname, tablename, attname, n_distinct, correlation
FROM pg_stats
WHERE tablename = 'casos_agravos'
AND n_distinct > 100;

Soluções:

Adicionar índices apropriados
Aumentar work_mem no PostgreSQL
Implementar cache mais agressivo
Considerar particionamento de tabelas

Monitoramento

Métricas Importantes

# Prometheus metrics
from prometheus_client import Counter, Histogram

casos_criados = Counter(
    'agravos_casos_criados_total',
    'Total de casos criados',
    ['tipo_agravo', 'municipio']
)

validacao_duration = Histogram(
    'agravos_validacao_duration_seconds',
    'Tempo de validação de dados',
    ['tipo_agravo']
)

# Uso
with validacao_duration.labels(tipo_agravo=codigo).time():
    erros = agravo.validar_dados_especificos(dados)

Queries de Monitoramento

-- Casos por dia
SELECT 
    DATE(data_notificacao) as dia,
    COUNT(*) as total,
    COUNT(CASE WHEN classificacao_final = 'confirmado' THEN 1 END) as confirmados
FROM casos_agravos
WHERE data_notificacao >= CURRENT_DATE - INTERVAL '30 days'
GROUP BY DATE(data_notificacao)
ORDER BY dia DESC;

-- Top municípios
SELECT 
    municipio_residencia,
    COUNT(*) as total_casos,
    COUNT(DISTINCT agravo_id) as tipos_diferentes
FROM casos_agravos
WHERE data_notificacao >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY municipio_residencia
ORDER BY total_casos DESC
LIMIT 10;

Esta documentação é mantida pela equipe Sinapse. Para dúvidas ou sugestões, abra uma issue no GitHub.

Módulo de Agravos - Documentação Técnica

On this page