#!/usr/bin/env python3 # -*- coding:utf-8 -*- from contextlib import contextmanager from typing import Generator from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import declarative_base from app.core.config import settings Base = declarative_base() SQLALCHEMY_DATABASE_URL = settings.mysql_dsn engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True, echo=False, pool_size=10, max_overflow=20) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # @contextmanager # def get_db() -> Generator: # db = SessionLocal() # try: # yield db # finally: # db.close() def get_db() -> Generator: try: db = SessionLocal() yield db finally: db.close() if __name__ == '__main__': with get_db() as db: print(db.info)