comunatec/controles
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- inits centralizados, READMEs atualizados

Browse Source 
- 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
This commit is contained in:
Mateus Tavares
2026-02-20 17:19:15 -03:00
parent 6882b57081
commit 2b1668206d
38 changed files with 1250 additions and 1187 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
models/militante_model.py
View File

@@ -1,4 +1,4 @@
from functions.database import get_db_connection, Militante, EmailMilitante, Endereco
from functions.database import get_db_session, Militante, EmailMilitante, Endereco
from sqlalchemy.orm import joinedload
from datetime import datetime
from typing import List, Dict, Optional
@@ -14,7 +14,7 @@ class MilitanteModel:
@invalidate_cache_pattern("militantes:*")
def criar_militante(data: Dict) -> Dict:
"""Cria um novo militante"""
db = get_db_connection()
db = get_db_session()
try:
# Criar endereço se fornecido
endereco_id = None
@@ -89,7 +89,7 @@ class MilitanteModel:
@cached(expire=1800, key_prefix="militantes") # Cache for 30 minutes
def listar_militantes() -> List[Militante]:
"""Lista todos os militantes"""
db = get_db_connection()
db = get_db_session()
try:
militantes = db.query(Militante).options(
joinedload(Militante.emails),
@@ -125,7 +125,7 @@ class MilitanteModel:
return cached_militante
# Cache miss, get from database
db = get_db_connection()
db = get_db_session()
try:
militante = db.query(Militante).options(
joinedload(Militante.emails),
@@ -150,7 +150,7 @@ class MilitanteModel:
@invalidate_cache_pattern("militantes:*")
def atualizar_militante(militante_id: int, data: Dict) -> Dict:
"""Atualiza um militante existente"""
db = get_db_connection()
db = get_db_session()
try:
militante = db.query(Militante).get(militante_id)
@@ -228,7 +228,7 @@ class MilitanteModel:
@invalidate_cache_pattern("militantes:*")
def excluir_militante(militante_id: int) -> Dict:
"""Exclui um militante"""
db = get_db_connection()
db = get_db_session()
try:
militante = db.query(Militante).get(militante_id)
if not militante:
@@ -265,7 +265,7 @@ class MilitanteModel:
@cached(expire=1800, key_prefix="militantes")
def buscar_por_cpf(cpf: str) -> Optional[Militante]:
"""Busca um militante por CPF"""
db = get_db_connection()
db = get_db_session()
try:
militante = db.query(Militante).filter_by(cpf=cpf).first()
if militante:

24
models/pagamento_model.py
View File

@@ -1,4 +1,4 @@
from functions.database import get_db_connection, Pagamento, Militante, TipoPagamento
from functions.database import get_db_session, Pagamento, Militante, TipoPagamento
from sqlalchemy.orm import joinedload
from datetime import datetime
from typing import List, Dict, Optional
@@ -9,7 +9,7 @@ class PagamentoModel:
@staticmethod
def criar_pagamento(data: Dict) -> Dict:
"""Cria um novo pagamento"""
db = get_db_connection()
db = get_db_session()
try:
pagamento = Pagamento(
militante_id=data['militante_id'],
@@ -38,7 +38,7 @@ class PagamentoModel:
@staticmethod
def listar_pagamentos() -> List[Pagamento]:
"""Lista todos os pagamentos"""
db = get_db_connection()
db = get_db_session()
try:
return db.query(Pagamento).join(Militante).order_by(Pagamento.data_pagamento.desc()).all()
finally:
@@ -47,7 +47,7 @@ class PagamentoModel:
@staticmethod
def buscar_por_id(pagamento_id: int) -> Optional[Pagamento]:
"""Busca um pagamento por ID"""
db = get_db_connection()
db = get_db_session()
try:
return db.query(Pagamento).get(pagamento_id)
finally:
@@ -56,7 +56,7 @@ class PagamentoModel:
@staticmethod
def atualizar_pagamento(pagamento_id: int, data: Dict) -> Dict:
"""Atualiza um pagamento existente"""
db = get_db_connection()
db = get_db_session()
try:
pagamento = db.query(Pagamento).get(pagamento_id)
@@ -90,7 +90,7 @@ class PagamentoModel:
@staticmethod
def excluir_pagamento(pagamento_id: int) -> Dict:
"""Exclui um pagamento"""
db = get_db_connection()
db = get_db_session()
try:
pagamento = db.query(Pagamento).get(pagamento_id)
if not pagamento:
@@ -119,7 +119,7 @@ class PagamentoModel:
@staticmethod
def listar_por_celula(celula_id: int) -> List[Pagamento]:
"""Lista pagamentos de uma célula específica"""
db = get_db_connection()
db = get_db_session()
try:
return db.query(Pagamento).filter_by(celula_id=celula_id).all()
finally:
@@ -128,7 +128,7 @@ class PagamentoModel:
@staticmethod
def listar_por_setor(setor_id: int) -> List[Pagamento]:
"""Lista pagamentos de um setor específico"""
db = get_db_connection()
db = get_db_session()
try:
return db.query(Pagamento).join(Usuario).filter(Usuario.setor_id == setor_id).all()
finally:
@@ -137,7 +137,7 @@ class PagamentoModel:
@staticmethod
def listar_por_cr(cr_id: int) -> List[Pagamento]:
"""Lista pagamentos de um CR específico"""
db = get_db_connection()
db = get_db_session()
try:
return db.query(Pagamento).join(Usuario).filter(Usuario.cr_id == cr_id).all()
finally:
@@ -146,7 +146,7 @@ class PagamentoModel:
@staticmethod
def listar_por_militante(militante_id: int) -> List[Pagamento]:
"""Lista pagamentos de um militante específico"""
db = get_db_connection()
db = get_db_session()
try:
return db.query(Pagamento).filter_by(militante_id=militante_id).order_by(Pagamento.data_pagamento.desc()).all()
finally:
@@ -155,7 +155,7 @@ class PagamentoModel:
@staticmethod
def obter_tipos_pagamento() -> List[TipoPagamento]:
"""Obtém todos os tipos de pagamento"""
db = get_db_connection()
db = get_db_session()
try:
return db.query(TipoPagamento).order_by(TipoPagamento.descricao).all()
finally:
@@ -164,7 +164,7 @@ class PagamentoModel:
@staticmethod
def obter_militantes() -> List[Militante]:
"""Obtém todos os militantes"""
db = get_db_connection()
db = get_db_session()
try:
return db.query(Militante).order_by(Militante.nome).all()
finally: