[Python] Problema ao fechar conexões do SQLalchemy

Estou desenvolvendo uma aplicação web em python e flask-sqlalchemy e estou tendo problemas com o pool de conexões, as conexões não estão sendo fechadas após as operações no banco de dados, mesmo eu utilizando o gerenciador de contexto with. a configuração do meu projeto esta da seguinte forma:

__init__.py:

db = SQLAlchemy()
scheduler = APScheduler()

def create_app():
app = Flask(__name__)
CORS(app)
app.config.from_object(Config)

db.init_app(app)
scheduler.init_app(app)
scheduler.start()



run.py:

from app import create_app, db

app = create_app()

with app.app_context():
db.create_all()


if __name__ == "__main__":
app.run(host='0.0.0.0', port=4000, debug=True)


config.py:

from urllib.parse import quote_plus
import os
from dotenv import load_dotenv

# Carregando dotenv para variáveis de ambiente
load_dotenv()

# Declaração de variáveis de ambiente
APP_ID = os.getenv('APP_ID')
SECRET_KEY = os.getenv('SECRET_KEY')
REDIRECT_URI_NGROK = os.getenv('REDIRECT_URI_NGROK')

# Variáveis de ambiente para conexão com o banco de dados
hostname = os.getenv('SQL_HOSTNAME')
database = os.getenv('SQL_DATABASE')
username = os.getenv('SQL_USERNAME')
password = os.getenv('password')
password_encoded = quote_plus(password)
domain = "http://127.0.0.1:4000"

class Config:
SECRET_KEY = SECRET_KEY
print(f"SECRET_KEY:{SECRET_KEY}")
REDIRECT_URI = REDIRECT_URI_NGROK
print(f"REDIRECT_URI:{REDIRECT_URI}")

# URI do banco de dados com MySQL e PyMySQL
SQLALCHEMY_DATABASE_URI = f'mysql+pymysql://{username}:{password_encoded}@{hostname}/{database}'


SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_AUTO_FLUSH = True

# Configurações para o pool de conexões
SQLALCHEMY_ENGINE_OPTIONS = {
"pool_size":20, # Número máximo de conexões no pool
"max_overflow":1, # Excesso de conexões além do pool_size
"pool_timeout":30, # Tempo máximo para esperar por uma conexão disponível
"pool_recycle":3600, # Tempo para reciclar (renovar) conexões
"pool_pre_ping":True
}

# Configurações do APScheduler
SCHEDULER_API_ENABLED = True


e por ultimo, um exemplo de uma função que uso para manipular o DB:

def get_total_visits_ontem(self, variacao_selecionada):
with db.session() as session:

ontem = datetime.now() - timedelta(days=1)
ontem = ontem.date() # Obtém apenas a parte da data

total_visits = session.query(func.sum(Acompanhamento_concorrentes.visits_ontem)).filter(
Acompanhamento_concorrentes.mlb_filho == variacao_selecionada,
func.date(Acompanhamento_concorrentes.date_checked) == ontem # Compara apenas a data
).scalar()

return total_visits or 0


eu estou importando o db como: from app import db, para utilizar nesse arquivo, mas por algum motivo há conexões ficando em aberto, esgotando o pool de conexões

Continue reading...