27 lines
672 B
Python
27 lines
672 B
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from functools import lru_cache
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import Engine
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_engine() -> Engine:
|
|
"""Return a cached SQLAlchemy engine configured from DATABASE_URL."""
|
|
database_url = os.getenv(
|
|
"DATABASE_URL",
|
|
"mysql+pymysql://root:12345678@localhost:3306/data-ge?charset=utf8mb4",
|
|
)
|
|
connect_args = {}
|
|
if database_url.startswith("sqlite"):
|
|
connect_args["check_same_thread"] = False
|
|
|
|
return create_engine(
|
|
database_url,
|
|
pool_pre_ping=True,
|
|
future=True,
|
|
connect_args=connect_args,
|
|
)
|