feat: factory de conexao MySQL
This commit is contained in:
@@ -0,0 +1,22 @@
|
||||
"""
|
||||
Única porta de entrada para conexão MySQL do painel. Todo módulo em
|
||||
app/db/*.py chama get_connection() daqui — nunca abre pymysql.connect()
|
||||
por conta própria.
|
||||
"""
|
||||
import os
|
||||
|
||||
import pymysql
|
||||
import pymysql.cursors
|
||||
|
||||
|
||||
def get_connection() -> pymysql.connections.Connection:
|
||||
"""Abre uma conexão nova por requisição (escala MVP; sem pool)."""
|
||||
return pymysql.connect(
|
||||
host=os.environ["DB_HOST"],
|
||||
port=int(os.environ.get("DB_PORT", "3306")),
|
||||
user=os.environ["DB_USER"],
|
||||
password=os.environ["DB_PASSWORD"],
|
||||
database=os.environ["DB_NAME"],
|
||||
cursorclass=pymysql.cursors.DictCursor,
|
||||
autocommit=True,
|
||||
)
|
||||
@@ -0,0 +1,17 @@
|
||||
"""
|
||||
Requer o container de dev rodando (licenciamento-mysql-dev, porta 3307 —
|
||||
ver licenciamento-n8n/docs/configurar-banco-sistema.md). tests/conftest.py
|
||||
já configura as env vars de conexão com os mesmos defaults do .env.example.
|
||||
"""
|
||||
from app.db.connection import get_connection
|
||||
|
||||
|
||||
def test_conexao_abre_e_consulta():
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT 1 AS um")
|
||||
row = cur.fetchone()
|
||||
assert row == {"um": 1}
|
||||
finally:
|
||||
conn.close()
|
||||
Reference in New Issue
Block a user