提到 SQLAlchemy 和 Flask-SQLAlchemy,相信大家都不会感到陌生。Flask-SQLAlchemy 封装后使用起来非常方便。然而,当我们脱离 Flask 使用原生 SQLAlchemy 时,经常会遇到数据库连接丢失(Lost connection to MySQL server)的问题。此类错误即使通过调整数据库的等待时长也无法有效解决,每次都需要重启服务才能重新连接成功。那么如何正确地使用原生 SQLAlchemy 并避免数据库连接丢失的问题呢?本文将详细介绍这一过程。
安装与配置
首先,确保你已经安装了 SQLAlchemy 和 SQLAlchemy 的异步扩展模块:
pip install sqlalchemy sqlalchemy[asyncio] asyncpg
然后,在你的项目中进行如下配置:
创建数据库引擎
import asyncio
from contextlib import contextmanager, asynccontextmanager
import glob
import importlib
import os
import re
from typing import Generator, AsyncGenerator
from loguru import logger
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_scoped_session, async_sessionmaker
from sqlalchemy.orm import sessionmaker, scoped_session
from config import settings
# 创建同步数据库引擎
engine = create_engine(
settings.SQLALCHEMY_DATABASE_URI,
echo=settings.SQLALCHEMY_ECHO,
pool_recycle=settings.SQLALCHEMY_