feat: Implementar arquitetura de permissões no nível de dados
BREAKING CHANGES: - Sistema de permissões movido do nível de template para nível de dados - Menus sempre visíveis, controle transparente no backend - Templates nunca quebram, sempre renderizam com dados filtrados Features: - ✅ Arquitetura MVC completa implementada - ✅ Controllers com filtragem hierárquica de dados - ✅ Template helpers simplificados (user_can sempre True) - ✅ Controle de acesso baseado na hierarquia organizacional - ✅ Regra especial para tesoureiros (acesso completo) - ✅ Tratamento robusto de erros em todos os controllers Controllers implementados: - militante_controller.py - Filtragem por célula/setor/CR/CC - cota_controller.py - Controle baseado em permissões - material_controller.py - Acesso flexível por nível - pagamento_controller.py - Filtragem organizacional - auth_controller.py - Autenticação com OTP - home_controller.py - Dashboard com estatísticas - usuario_controller.py - Gestão de usuários Templates corrigidos: - listar_cotas.html - URLs corrigidas (nova_cota → cota.nova) - listar_tipos_materiais.html - Variáveis ajustadas (tipos → tipos_materiais) - base.html - Menus sempre visíveis - Diversos templates com correções de URLs e referências Services implementados: - auth_service.py - Lógica de autenticação - dashboard_service.py - Estatísticas do dashboard - cache_service.py - Integração com Redis - celula_service.py - Operações de células Models implementados: - militante_model.py - Operações de militantes - pagamento_model.py - Operações de pagamentos Documentação: - docs/permission_fixes_summary.md - Resumo completo das correções - docs/architecture_summary.md - Arquitetura MVC - docs/mvc_refactoring.md - Detalhes da refatoração - docs/permission_strategy.md - Estratégia de permissões - docs/redis_cache_setup.md - Setup do cache Redis - README.md atualizado com nova arquitetura Testes: - test_menu_navigation.py - Testes unitários de navegação - test_integration_menu.py - Testes de integração com Selenium Status dos testes: ✅ Funcionais: /, /dashboard, /pagamentos, /materiais ❌ Com problemas: /militantes, /cotas, /tipos-materiais, /admin/dashboard Hierarquia de permissões implementada: Admin → Acesso total CC → Acesso total CR → Dados do CR Setor → Dados do setor Célula → Dados da célula Próximos passos identificados: - Corrigir referências a Militante indefinido nos templates - Resolver problemas de campos inexistentes - Corrigir roteamento admin
This commit is contained in:
LS
1
models/__init__.py
Normal file
1
models/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# Models package
|
||||
4
models/entities/assinatura_anual.py
Normal file
4
models/entities/assinatura_anual.py
Normal file
@@ -0,0 +1,4 @@
|
||||
from models.entities.assinatura_jornal import AssinaturaJornal as AssinaturaAnual
|
||||
|
||||
# This file is a compatibility layer for code that uses AssinaturaAnual
|
||||
# The class has been renamed to AssinaturaJornal
|
||||
24
models/entities/celula.py
Normal file
24
models/entities/celula.py
Normal file
@@ -0,0 +1,24 @@
|
||||
from sqlalchemy import Column, Integer, String, ForeignKey
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from models.entities.base import Base
|
||||
|
||||
class Celula(Base):
|
||||
__tablename__ = 'celulas'
|
||||
|
||||
id = Column(Integer, primary_key=True, autoincrement=True)
|
||||
nome = Column(String(100), nullable=False)
|
||||
setor_id = Column(Integer, ForeignKey('setores.id', use_alter=True, name='fk_celula_setor'))
|
||||
cr_id = Column(Integer, ForeignKey('comites_regionais.id', use_alter=True, name='fk_celula_cr'))
|
||||
secretario = Column(Integer, ForeignKey('militantes.id', use_alter=True, name='fk_celula_secretario'))
|
||||
responsavel_financas = Column(Integer, ForeignKey('militantes.id', use_alter=True, name='fk_celula_responsavel_financas'))
|
||||
quadro_orientador = Column(String(255))
|
||||
|
||||
# Relacionamentos
|
||||
setor = relationship("Setor", back_populates="celulas")
|
||||
cr = relationship("ComiteRegional", back_populates="celulas")
|
||||
militantes = relationship("Militante", back_populates="celula", foreign_keys="[Militante.celula_id]")
|
||||
secretario_rel = relationship("Militante", foreign_keys=[secretario])
|
||||
responsavel_financas_rel = relationship("Militante", foreign_keys=[responsavel_financas])
|
||||
pagamentos = relationship("PagamentoCelula", back_populates="celula")
|
||||
usuarios = relationship("Usuario", back_populates="celula")
|
||||
328
models/militante_model.py
Normal file
328
models/militante_model.py
Normal file
@@ -0,0 +1,328 @@
|
||||
from functions.database import get_db_connection, Militante, EmailMilitante, Endereco
|
||||
from sqlalchemy.orm import joinedload
|
||||
from datetime import datetime
|
||||
from typing import List, Dict, Optional
|
||||
from services.cache_service import cache_service, cached, CacheKeys, invalidate_cache_pattern
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class MilitanteModel:
|
||||
"""Model para operações com militantes"""
|
||||
|
||||
@staticmethod
|
||||
@invalidate_cache_pattern("militantes:*")
|
||||
def criar_militante(data: Dict) -> Dict:
|
||||
"""Cria um novo militante"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
# Criar endereço se fornecido
|
||||
endereco_id = None
|
||||
if data.get('endereco'):
|
||||
endereco = Endereco(**data['endereco'])
|
||||
db.add(endereco)
|
||||
db.flush()
|
||||
endereco_id = endereco.id
|
||||
|
||||
# Criar militante
|
||||
militante = Militante(
|
||||
nome=data['nome'],
|
||||
cpf=data['cpf'],
|
||||
titulo_eleitoral=data.get('titulo_eleitoral'),
|
||||
data_nascimento=data.get('data_nascimento'),
|
||||
data_entrada_oci=data.get('data_entrada_oci'),
|
||||
data_efetivacao_oci=data.get('data_efetivacao_oci'),
|
||||
telefone1=data.get('telefone1'),
|
||||
telefone2=data.get('telefone2'),
|
||||
profissao=data.get('profissao'),
|
||||
regime_trabalho=data.get('regime_trabalho'),
|
||||
empresa=data.get('empresa'),
|
||||
contratante=data.get('contratante'),
|
||||
instituicao_ensino=data.get('instituicao_ensino'),
|
||||
tipo_instituicao=data.get('tipo_instituicao'),
|
||||
sindicato=data.get('sindicato'),
|
||||
cargo_sindical=data.get('cargo_sindical'),
|
||||
dirigente_sindical=data.get('dirigente_sindical', False),
|
||||
central_sindical=data.get('central_sindical'),
|
||||
endereco_id=endereco_id,
|
||||
celula_id=data.get('celula_id'),
|
||||
registrado_por=data.get('registrado_por')
|
||||
)
|
||||
|
||||
db.add(militante)
|
||||
db.flush()
|
||||
|
||||
# Criar email se fornecido
|
||||
if data.get('email'):
|
||||
email = EmailMilitante(
|
||||
militante_id=militante.id,
|
||||
endereco_email=data['email']
|
||||
)
|
||||
db.add(email)
|
||||
|
||||
db.commit()
|
||||
|
||||
# Cache the new militante
|
||||
cache_key = CacheKeys.militante_detail(militante.id)
|
||||
militante_data = MilitanteModel.formatar_dados_militante(militante)
|
||||
cache_service.set(cache_key, militante_data, 1800) # 30 minutes
|
||||
|
||||
logger.info(f"Militante {militante.id} criado e cacheado")
|
||||
|
||||
return {
|
||||
'status': 'success',
|
||||
'message': 'Militante criado com sucesso',
|
||||
'militante_id': militante.id
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
logger.error(f"Erro ao criar militante: {e}")
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': f'Erro ao criar militante: {str(e)}'
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
@cached(expire=1800, key_prefix="militantes") # Cache for 30 minutes
|
||||
def listar_militantes() -> List[Militante]:
|
||||
"""Lista todos os militantes"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
militantes = db.query(Militante).options(
|
||||
joinedload(Militante.emails),
|
||||
joinedload(Militante.endereco),
|
||||
joinedload(Militante.celula)
|
||||
).order_by(Militante.nome).all()
|
||||
|
||||
# Cache individual militantes
|
||||
for militante in militantes:
|
||||
cache_key = CacheKeys.militante_detail(militante.id)
|
||||
militante_data = MilitanteModel.formatar_dados_militante(militante)
|
||||
cache_service.set(cache_key, militante_data, 1800)
|
||||
|
||||
logger.debug(f"Listados {len(militantes)} militantes e cacheados")
|
||||
return militantes
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Erro ao listar militantes: {e}")
|
||||
return []
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def buscar_por_id(militante_id: int) -> Optional[Militante]:
|
||||
"""Busca um militante por ID"""
|
||||
# Try cache first
|
||||
cache_key = CacheKeys.militante_detail(militante_id)
|
||||
cached_militante = cache_service.get(cache_key)
|
||||
|
||||
if cached_militante:
|
||||
logger.debug(f"Cache hit para militante {militante_id}")
|
||||
# Convert cached data back to Militante object if needed
|
||||
return cached_militante
|
||||
|
||||
# Cache miss, get from database
|
||||
db = get_db_connection()
|
||||
try:
|
||||
militante = db.query(Militante).options(
|
||||
joinedload(Militante.emails),
|
||||
joinedload(Militante.endereco)
|
||||
).get(militante_id)
|
||||
|
||||
if militante:
|
||||
# Cache the militante
|
||||
militante_data = MilitanteModel.formatar_dados_militante(militante)
|
||||
cache_service.set(cache_key, militante_data, 1800)
|
||||
logger.debug(f"Cache miss para militante {militante_id}, cacheado")
|
||||
|
||||
return militante
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Erro ao buscar militante {militante_id}: {e}")
|
||||
return None
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
@invalidate_cache_pattern("militantes:*")
|
||||
def atualizar_militante(militante_id: int, data: Dict) -> Dict:
|
||||
"""Atualiza um militante existente"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
militante = db.query(Militante).get(militante_id)
|
||||
|
||||
if not militante:
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': 'Militante não encontrado'
|
||||
}
|
||||
|
||||
# Atualizar dados básicos
|
||||
militante.nome = data.get('nome', militante.nome)
|
||||
militante.cpf = data.get('cpf', militante.cpf)
|
||||
militante.titulo_eleitoral = data.get('titulo_eleitoral', militante.titulo_eleitoral)
|
||||
militante.telefone1 = data.get('telefone1', militante.telefone1)
|
||||
militante.telefone2 = data.get('telefone2', militante.telefone2)
|
||||
militante.profissao = data.get('profissao', militante.profissao)
|
||||
militante.regime_trabalho = data.get('regime_trabalho', militante.regime_trabalho)
|
||||
militante.empresa = data.get('empresa', militante.empresa)
|
||||
militante.contratante = data.get('contratante', militante.contratante)
|
||||
militante.instituicao_ensino = data.get('instituicao_ensino', militante.instituicao_ensino)
|
||||
militante.tipo_instituicao = data.get('tipo_instituicao', militante.tipo_instituicao)
|
||||
militante.sindicato = data.get('sindicato', militante.sindicato)
|
||||
militante.cargo_sindical = data.get('cargo_sindical', militante.cargo_sindical)
|
||||
militante.dirigente_sindical = data.get('dirigente_sindical', militante.dirigente_sindical)
|
||||
militante.central_sindical = data.get('central_sindical', militante.central_sindical)
|
||||
|
||||
# Atualizar datas
|
||||
if data.get('data_nascimento'):
|
||||
militante.data_nascimento = data['data_nascimento']
|
||||
if data.get('data_entrada_oci'):
|
||||
militante.data_entrada_oci = data['data_entrada_oci']
|
||||
if data.get('data_efetivacao_oci'):
|
||||
militante.data_efetivacao_oci = data['data_efetivacao_oci']
|
||||
|
||||
# Atualizar endereço
|
||||
if data.get('endereco') and militante.endereco:
|
||||
endereco = militante.endereco
|
||||
endereco.cep = data['endereco'].get('cep', endereco.cep)
|
||||
endereco.estado = data['endereco'].get('estado', endereco.estado)
|
||||
endereco.cidade = data['endereco'].get('cidade', endereco.cidade)
|
||||
endereco.bairro = data['endereco'].get('bairro', endereco.bairro)
|
||||
endereco.rua = data['endereco'].get('rua', endereco.rua)
|
||||
endereco.numero = data['endereco'].get('numero', endereco.numero)
|
||||
endereco.complemento = data['endereco'].get('complemento', endereco.complemento)
|
||||
|
||||
# Atualizar email
|
||||
if data.get('email') and militante.emails:
|
||||
militante.emails[0].endereco_email = data['email']
|
||||
|
||||
db.commit()
|
||||
|
||||
# Update cache
|
||||
cache_key = CacheKeys.militante_detail(militante_id)
|
||||
militante_data = MilitanteModel.formatar_dados_militante(militante)
|
||||
cache_service.set(cache_key, militante_data, 1800)
|
||||
|
||||
logger.info(f"Militante {militante_id} atualizado e cache atualizado")
|
||||
|
||||
return {
|
||||
'status': 'success',
|
||||
'message': 'Militante atualizado com sucesso'
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
logger.error(f"Erro ao atualizar militante {militante_id}: {e}")
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': f'Erro ao atualizar militante: {str(e)}'
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
@invalidate_cache_pattern("militantes:*")
|
||||
def excluir_militante(militante_id: int) -> Dict:
|
||||
"""Exclui um militante"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
militante = db.query(Militante).get(militante_id)
|
||||
if not militante:
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': 'Militante não encontrado'
|
||||
}
|
||||
|
||||
db.delete(militante)
|
||||
db.commit()
|
||||
|
||||
# Remove from cache
|
||||
cache_key = CacheKeys.militante_detail(militante_id)
|
||||
cache_service.delete(cache_key)
|
||||
|
||||
logger.info(f"Militante {militante_id} excluído e removido do cache")
|
||||
|
||||
return {
|
||||
'status': 'success',
|
||||
'message': 'Militante excluído com sucesso'
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
logger.error(f"Erro ao excluir militante {militante_id}: {e}")
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': f'Erro ao excluir militante: {str(e)}'
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
@cached(expire=1800, key_prefix="militantes")
|
||||
def buscar_por_cpf(cpf: str) -> Optional[Militante]:
|
||||
"""Busca um militante por CPF"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
militante = db.query(Militante).filter_by(cpf=cpf).first()
|
||||
if militante:
|
||||
# Cache the militante
|
||||
cache_key = CacheKeys.militante_detail(militante.id)
|
||||
militante_data = MilitanteModel.formatar_dados_militante(militante)
|
||||
cache_service.set(cache_key, militante_data, 1800)
|
||||
return militante
|
||||
except Exception as e:
|
||||
logger.error(f"Erro ao buscar militante por CPF {cpf}: {e}")
|
||||
return None
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def formatar_dados_militante(militante: Militante) -> Dict:
|
||||
"""Formata os dados de um militante para retorno JSON"""
|
||||
def formatar_data_segura(data):
|
||||
try:
|
||||
if not data:
|
||||
return None
|
||||
return data.strftime('%Y-%m-%d')
|
||||
except Exception as e:
|
||||
logger.error(f"Erro ao formatar data: {str(e)}, valor: {data}")
|
||||
return None
|
||||
|
||||
return {
|
||||
'id': militante.id,
|
||||
'nome': militante.nome,
|
||||
'cpf': militante.cpf,
|
||||
'titulo_eleitoral': militante.titulo_eleitoral,
|
||||
'data_nascimento': formatar_data_segura(militante.data_nascimento),
|
||||
'data_entrada_oci': formatar_data_segura(militante.data_entrada_oci),
|
||||
'data_efetivacao_oci': formatar_data_segura(militante.data_efetivacao_oci),
|
||||
'telefone1': militante.telefone1,
|
||||
'telefone2': militante.telefone2,
|
||||
'profissao': militante.profissao,
|
||||
'regime_trabalho': militante.regime_trabalho,
|
||||
'empresa': militante.empresa,
|
||||
'contratante': militante.contratante,
|
||||
'instituicao_ensino': militante.instituicao_ensino,
|
||||
'tipo_instituicao': militante.tipo_instituicao,
|
||||
'sindicato': militante.sindicato,
|
||||
'cargo_sindical': militante.cargo_sindical,
|
||||
'dirigente_sindical': militante.dirigente_sindical,
|
||||
'central_sindical': militante.central_sindical,
|
||||
'responsabilidades': militante.responsabilidades,
|
||||
'estado': militante.estado.value if militante.estado else None,
|
||||
'celula_id': militante.celula_id,
|
||||
'email': militante.emails[0].endereco_email if militante.emails else None,
|
||||
'endereco': {
|
||||
'cep': militante.endereco.cep if militante.endereco else None,
|
||||
'estado': militante.endereco.estado if militante.endereco else None,
|
||||
'cidade': militante.endereco.cidade if militante.endereco else None,
|
||||
'bairro': militante.endereco.bairro if militante.endereco else None,
|
||||
'rua': militante.endereco.rua if militante.endereco else None,
|
||||
'numero': militante.endereco.numero if militante.endereco else None,
|
||||
'complemento': militante.endereco.complemento if militante.endereco else None
|
||||
} if militante.endereco else None
|
||||
}
|
||||
184
models/pagamento_model.py
Normal file
184
models/pagamento_model.py
Normal file
@@ -0,0 +1,184 @@
|
||||
from functions.database import get_db_connection, Pagamento, Militante, TipoPagamento
|
||||
from sqlalchemy.orm import joinedload
|
||||
from datetime import datetime
|
||||
from typing import List, Dict, Optional
|
||||
|
||||
class PagamentoModel:
|
||||
"""Model para operações com pagamentos"""
|
||||
|
||||
@staticmethod
|
||||
def criar_pagamento(data: Dict) -> Dict:
|
||||
"""Cria um novo pagamento"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
pagamento = Pagamento(
|
||||
militante_id=data['militante_id'],
|
||||
tipo_pagamento_id=data.get('tipo_pagamento_id'),
|
||||
valor=data['valor'],
|
||||
data_pagamento=data['data_pagamento']
|
||||
)
|
||||
db.add(pagamento)
|
||||
db.commit()
|
||||
|
||||
return {
|
||||
'status': 'success',
|
||||
'message': 'Pagamento criado com sucesso',
|
||||
'pagamento_id': pagamento.id
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': f'Erro ao criar pagamento: {str(e)}'
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def listar_pagamentos() -> List[Pagamento]:
|
||||
"""Lista todos os pagamentos"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
return db.query(Pagamento).join(Militante).order_by(Pagamento.data_pagamento.desc()).all()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def buscar_por_id(pagamento_id: int) -> Optional[Pagamento]:
|
||||
"""Busca um pagamento por ID"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
return db.query(Pagamento).get(pagamento_id)
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def atualizar_pagamento(pagamento_id: int, data: Dict) -> Dict:
|
||||
"""Atualiza um pagamento existente"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
pagamento = db.query(Pagamento).get(pagamento_id)
|
||||
|
||||
if not pagamento:
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': 'Pagamento não encontrado'
|
||||
}
|
||||
|
||||
pagamento.militante_id = data.get('militante_id', pagamento.militante_id)
|
||||
pagamento.tipo_pagamento_id = data.get('tipo_pagamento_id', pagamento.tipo_pagamento_id)
|
||||
pagamento.valor = data.get('valor', pagamento.valor)
|
||||
pagamento.data_pagamento = data.get('data_pagamento', pagamento.data_pagamento)
|
||||
|
||||
db.commit()
|
||||
|
||||
return {
|
||||
'status': 'success',
|
||||
'message': 'Pagamento atualizado com sucesso'
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': f'Erro ao atualizar pagamento: {str(e)}'
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def excluir_pagamento(pagamento_id: int) -> Dict:
|
||||
"""Exclui um pagamento"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
pagamento = db.query(Pagamento).get(pagamento_id)
|
||||
if not pagamento:
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': 'Pagamento não encontrado'
|
||||
}
|
||||
|
||||
db.delete(pagamento)
|
||||
db.commit()
|
||||
|
||||
return {
|
||||
'status': 'success',
|
||||
'message': 'Pagamento excluído com sucesso'
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
return {
|
||||
'status': 'error',
|
||||
'message': f'Erro ao excluir pagamento: {str(e)}'
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def listar_por_celula(celula_id: int) -> List[Pagamento]:
|
||||
"""Lista pagamentos de uma célula específica"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
return db.query(Pagamento).filter_by(celula_id=celula_id).all()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def listar_por_setor(setor_id: int) -> List[Pagamento]:
|
||||
"""Lista pagamentos de um setor específico"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
return db.query(Pagamento).join(Usuario).filter(Usuario.setor_id == setor_id).all()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def listar_por_cr(cr_id: int) -> List[Pagamento]:
|
||||
"""Lista pagamentos de um CR específico"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
return db.query(Pagamento).join(Usuario).filter(Usuario.cr_id == cr_id).all()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def listar_por_militante(militante_id: int) -> List[Pagamento]:
|
||||
"""Lista pagamentos de um militante específico"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
return db.query(Pagamento).filter_by(militante_id=militante_id).order_by(Pagamento.data_pagamento.desc()).all()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def obter_tipos_pagamento() -> List[TipoPagamento]:
|
||||
"""Obtém todos os tipos de pagamento"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
return db.query(TipoPagamento).order_by(TipoPagamento.descricao).all()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def obter_militantes() -> List[Militante]:
|
||||
"""Obtém todos os militantes"""
|
||||
db = get_db_connection()
|
||||
try:
|
||||
return db.query(Militante).order_by(Militante.nome).all()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@staticmethod
|
||||
def formatar_dados_pagamento(pagamento: Pagamento) -> Dict:
|
||||
"""Formata os dados de um pagamento para retorno JSON"""
|
||||
return {
|
||||
'id': pagamento.id,
|
||||
'militante_id': pagamento.militante_id,
|
||||
'tipo_pagamento_id': pagamento.tipo_pagamento_id,
|
||||
'valor': float(pagamento.valor) if pagamento.valor else 0.0,
|
||||
'data_pagamento': pagamento.data_pagamento.strftime('%Y-%m-%d') if pagamento.data_pagamento else None,
|
||||
'militante_nome': pagamento.militante.nome if pagamento.militante else None,
|
||||
'tipo_pagamento_nome': pagamento.tipo_pagamento.descricao if pagamento.tipo_pagamento else None
|
||||
}
|
||||
Reference in New Issue
Block a user