47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
from collections.abc import Iterator
|
|
from typing import Annotated
|
|
|
|
import pyodbc
|
|
from fastapi import Depends
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import URL
|
|
from sqlalchemy.pool import PoolProxiedConnection
|
|
|
|
from app.core.configs import settings
|
|
|
|
# O pool do SQLAlchemy substitui o pool interno do ODBC. Com os dois ligados,
|
|
# conexões que o SQLAlchemy descarta continuam abertas no driver.
|
|
pyodbc.pooling = False
|
|
|
|
|
|
def _escapar(valor: str) -> str:
|
|
# Envolve o valor em chaves pra que ; ou } na senha não quebrem a connection string
|
|
return '{' + valor.replace('}', '}}') + '}'
|
|
|
|
|
|
connection_string = (
|
|
f"DRIVER={_escapar(settings.db_driver)};"
|
|
f"SERVER={settings.db_host},{settings.db_porta};"
|
|
f"DATABASE={_escapar(settings.db_database)};"
|
|
f"UID={_escapar(settings.db_login)};"
|
|
f"PWD={_escapar(settings.db_senha.get_secret_value())};"
|
|
)
|
|
|
|
engine = create_engine(
|
|
URL.create('mssql+pyodbc', query={'odbc_connect': connection_string}),
|
|
pool_size=settings.db_pool_size, # conexões mantidas abertas
|
|
max_overflow=settings.db_pool_max_overflow, # extras permitidas em pico de acesso
|
|
pool_pre_ping=True, # testa a conexão antes de entregar (o servidor pode ter derrubado)
|
|
pool_recycle=1800, # renova conexões com mais de 30 min
|
|
)
|
|
|
|
|
|
def get_conexao() -> Iterator[PoolProxiedConnection]:
|
|
"""Pega uma conexão do pool e devolve ao final da requisição."""
|
|
conexao = engine.raw_connection()
|
|
try:
|
|
yield conexao
|
|
finally:
|
|
conexao.close() # devolve ao pool, não fecha de verdade
|
|
|
|
Conexao = Annotated[PoolProxiedConnection, Depends(get_conexao)]
|