赞
踩
提到 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_
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。