Mateus Tavares
2b1668206d
- padronizando o nome de get_db_connection e session para get_db_session, para não confundir com session do Flask ou sessoes web
- corrigindo potenciais erros
-- has_permission nao consegue com lazy load carregar permission depois de load_user fechar a conexao, entao joinedLoad com Permission antes de fechar
-- db.rollback não existe caso db = get_db_session() apareça muito depois dentro do try, padronizando antes de try
--- comparar role por nivel (Role.SECRETARIO_GERAL) e nao por nome ("Secretario Geral")
- unificacao de get_otp_qr_code
- mudança de nowutc() para now(UTC) conforme novo padrão
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
from datetime import datetime, UTC
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from functions.database import get_db_session, Controle as ControleModel
|
|
|
|
class Controle:
|
|
def __init__(self):
|
|
self.db = get_db_session()
|
|
|
|
def registrar_controle(self, militante_id: int, tipo: str, valor: float, observacao: str = None) -> bool:
|
|
"""
|
|
Registra um novo controle no sistema
|
|
|
|
Args:
|
|
militante_id: ID do militante
|
|
tipo: Tipo do controle (ex: 'pagamento', 'cota')
|
|
valor: Valor do controle
|
|
observacao: Observação opcional sobre o controle
|
|
|
|
Returns:
|
|
bool: True se o controle foi registrado com sucesso, False caso contrário
|
|
"""
|
|
try:
|
|
data_registro = datetime.now(UTC)
|
|
|
|
novo_controle = ControleModel(
|
|
militante_id=militante_id,
|
|
tipo=tipo,
|
|
valor=valor,
|
|
data_registro=data_registro,
|
|
observacao=observacao
|
|
)
|
|
|
|
self.db.add(novo_controle)
|
|
self.db.commit()
|
|
return True
|
|
|
|
except SQLAlchemyError as e:
|
|
self.db.rollback()
|
|
print(f"Erro ao registrar controle: {str(e)}")
|
|
return False
|
|
finally:
|
|
self.db.close()
|
|
|
|
def listar_controles(self, militante_id: int = None) -> list:
|
|
"""
|
|
Lista os controles registrados no sistema
|
|
|
|
Args:
|
|
militante_id: ID do militante para filtrar (opcional)
|
|
|
|
Returns:
|
|
list: Lista de controles encontrados
|
|
"""
|
|
try:
|
|
query = self.db.query(ControleModel)
|
|
|
|
if militante_id:
|
|
query = query.filter(ControleModel.militante_id == militante_id)
|
|
|
|
return query.all()
|
|
|
|
except SQLAlchemyError as e:
|
|
print(f"Erro ao listar controles: {str(e)}")
|
|
return []
|
|
finally:
|
|
self.db.close()
|
|
|
|
def buscar_controle(self, controle_id: int) -> ControleModel:
|
|
"""
|
|
Busca um controle específico pelo ID
|
|
|
|
Args:
|
|
controle_id: ID do controle
|
|
|
|
Returns:
|
|
ControleModel: Objeto do controle encontrado ou None
|
|
"""
|
|
try:
|
|
return self.db.query(ControleModel).filter(ControleModel.id == controle_id).first()
|
|
except SQLAlchemyError as e:
|
|
print(f"Erro ao buscar controle: {str(e)}")
|
|
return None
|
|
finally:
|
|
self.db.close() |