47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
import os
|
|
from dotenv import load_dotenv
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
# .env laden
|
|
load_dotenv()
|
|
|
|
# Datenbank-Zugangsdaten aus .env
|
|
DB_USER = os.getenv("DB_USER")
|
|
DB_PASSWORD = os.getenv("DB_PASSWORD")
|
|
DB_HOST = os.getenv("DB_HOST", "localhost")
|
|
DB_PORT = os.getenv("DB_PORT", "3306")
|
|
DB_NAME = os.getenv("DB_NAME")
|
|
|
|
# Pooling Parameter aus .env (optional mit Default-Werten)
|
|
POOL_SIZE = int(os.getenv("POOL_SIZE", 10))
|
|
MAX_OVERFLOW = int(os.getenv("MAX_OVERFLOW", 20))
|
|
POOL_TIMEOUT = int(os.getenv("POOL_TIMEOUT", 30))
|
|
POOL_RECYCLE = int(os.getenv("POOL_RECYCLE", 1800))
|
|
|
|
# Connection-String zusammenbauen
|
|
DATABASE_URL = (
|
|
f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
|
|
)
|
|
|
|
# Engine mit Pooling konfigurieren
|
|
engine = create_engine(
|
|
DATABASE_URL,
|
|
pool_size=POOL_SIZE,
|
|
max_overflow=MAX_OVERFLOW,
|
|
pool_timeout=POOL_TIMEOUT,
|
|
pool_recycle=POOL_RECYCLE,
|
|
echo=True, # für Debug, später False
|
|
)
|
|
|
|
# Session Factory
|
|
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
|
|
|
|
def get_session():
|
|
return SessionLocal()
|
|
|
|
def execute_query(query):
|
|
with engine.connect() as connection:
|
|
result = connection.execute(text(query))
|
|
return [dict(row) for row in result]
|