SQLAlchemy异步PostgreSQL配置循环导入问题求助

问题背景 #

配置SQLAlchemy异步ORM连接PostgreSQL时，参考《SQLAlchemy Async ORM IS Finally Here》教程，在models.py中用类方法实现增改查逻辑，但遇到database.py与models.py的循环导入问题：

• models.py依赖database.py中的async_db会话实例和Base基类
• database.py依赖models.py中的Employee类方法处理数据操作

核心需求：搭建一套支持增改查的规范数据库配置，解决循环导入问题。

现有代码示例 #

models.py #

from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy import Column, Integer, BigInteger, String, Float, DateTime, func, ForeignKey, Boolean
from sqlalchemy import update as sqlalchemy_update
from sqlalchemy.future import select
from helpers.database import async_db, Base


class ModelAdmin:
    @classmethod
    async def create(cls, **kwargs):
        session = await async_db.get_session()
        session.add(cls(**kwargs))
        await session.commit()

    @classmethod
    async def update(cls, id, **kwargs):
        query = (
            sqlalchemy_update(cls)
            .where(cls.id == id)
            .values(**kwargs)
            .execution_options(synchronize_session="fetch")
        )
        session = await async_db.get_session()
        await session.execute(query)
        await session.commit()

    @classmethod
    async def get(cls, id):
        query = select(cls).where(cls.id == id)
        session = await async_db.get_session()
        results = await session.execute(query)
        (result,) = results.one()
        return result


class Employee(Base, ModelAdmin):
    __tablename__ = "employees"
    id = Column(BigInteger, primary_key=True)
    username = Column(String(100))
    displayname = Column(String(100))
    email = Column(String(100), default="")
    emp_type = Column(String(30), default="")
    team = Column(String(100), default="")
    redline = Column(Float, default=-1)
    tier = Column(Integer, default=-1)
    cur_kw = Column(Float, default=0.0)
    emp_over = Column(BigInteger, default=-1)
    emp_under = Column(String, default="")
    deals = relationship("Deal")
    __mapper_args__ = {"eager_defaults": True}

    def get_tier(self) -> int:
        return .2

    def set_tier(self, tier: int) -> None:
        if self.emp_type == "canvasser":
            self.tier = tier

    def add_emp_over(self, emp):
        '''Adds an employee OVER another, AKA makes them a parent of the employee'''
        if emp.id not in self.emp_over.keys():
            self.emp_over[emp.id] = emp
            return f"Added {emp.username} as parent of {self.username}"

    def add_emp_under(self, emp):
        '''Adds an employee UNDER another, AKA makes them a child of the employee'''
        if emp.id not in self.emp_under.keys():
            self.emp_under[emp.id] = emp


class Deal(Base, ModelAdmin):
    __tablename__ = "deals"
    id = Column(Integer, primary_key=True)
    closer_id = Column(ForeignKey("employees.id"))
    deal_size = Column(Float)
    ppw = Column(Float)
    dealer_fee = Column(Float)
    lead_owner = relationship("Employee")
    installed = Column(Boolean)
    create_date = Column(DateTime, server_default=func.now())

    def get_commission(self):
        '''
            Gets the commission from this deal for the closer and the canvasser if the canvasser is set (otherwise it's a self-gen)
        '''
        commission = 0
        if not self.canvasser:
            commission = (self.deal_size * 1000) * \
                ((self.ppw * self.dealer_fee) - self.closer.blueline)
            return {commission}
        else:
            commission = {(self.deal_size * 1000) *
                ((self.ppw * self.dealer_fee) - self.closer.blueline)}
            commission[1] = commission[0] * self.canvasser.getTier()
            commission[0] = commission[0] - commission[1]
            return commission

database.py #

import discord
import asyncpg
import os
import json
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from dotenv import load_dotenv
from helpers.configuration import Config

load_dotenv()

db_name = os.environ.get('DB_NAME')
db_user = os.environ.get('DB_USER')
db_pass = os.environ.get('DB_PASS')
db_host = os.environ.get('DB_URL')


Base = declarative_base()


class AsyncDatabaseSession:
    '''Handles database connections, updates the employee list, receives member role updates, that good stuff'''
    from helpers.models import Employee

    def __init__(self):
        # here we should connect to and setup the database (if not setup) as well as
        # scan and index all members into Employee's then keep their database counterpart updated as well
        # any role updates should also ping here, let's get it
        self.employee_list = {}
        self._session = None
        self._engine = None

    async def init(self):
        self._engine = create_async_engine(
            f'postgresql+asyncpg://{db_user}:{db_pass}@{db_host}/{db_name}',
            echo=True,
            future=True
        )

        self._session = sessionmaker(
            self._engine, expire_on_commit=False, class_=AsyncSession
        )

    async def create_all(self):
        async with self._engine.begin() as conn:
            await conn.run_sync(Base.metadata.drop_all)
            await conn.run_sync(Base.metadata.create_all)

    async def async_session_generator(self):
        return sessionmaker(
            self._engine, expire_on_commit=False, class_1=AsyncSession
        )

    async def get_session(self):
        try:
            async_session = await self.async_session_generator()

            async with async_session() as session:
                yield session
        except:
            await session.rollback()
            raise
        finally:
            await session.close()
    
    async def update_employee(self, user: Employee):
        """Updates an employee"""
        from helpers.models import Employee
        if not len(user.emp_under) > 0:
            await Employee.update(
                id=user.user_id, username=user.username, displayname=user.displayname, email=user.email,
                emp_type=user.emp_type, team=user.team, redline=user.redline, tier=user.tier, cur_kw=user.cur_kw,
                emp_over=user.emp_over
            )
        else:
            await Employee.update(
                id=user.user_id, username=user.username, displayname=user.displayname, email=user.email,
                emp_type=user.emp_type, redline=user.redline, tier=user.tier, cur_kw=user.cur_kw,
                emp_over=user.emp_over, emp_under=json.dumps(user.emp_under)
            )

async_db = AsyncDatabaseSession()

解决方案 #

1. 拆分Base类到独立文件 #

创建base.py文件，统一管理SQLAlchemy基类，避免双向依赖：

# base.py
from sqlalchemy.orm import declarative_base

Base = declarative_base()

修改models.py和database.py的导入：

• models.py：from helpers.base import Base
• database.py：from helpers.base import Base

2. 延迟导入会话实例 #

在models.py的ModelAdmin类中，将async_db的导入移到方法内部，避免模块级循环导入，同时修正会话获取方式（原方法是生成器，需用async for遍历）：

class ModelAdmin:
    @classmethod
    async def create(cls, **kwargs):
        from helpers.database import async_db
        async for session in async_db.get_session():
            session.add(cls(**kwargs))

    @classmethod
    async def update(cls, id, **kwargs):
        from helpers.database import async_db
        query = (
            sqlalchemy_update(cls)
            .where(cls.id == id)
            .values(**kwargs)
            .execution_options(synchronize_session="fetch")
        )
        async for session in async_db.get_session():
            await session.execute(query)

    @classmethod
    async def get(cls, id):
        from helpers.database import async_db
        query = select(cls).where(cls.id == id)
        async for session in async_db.get_session():
            results = await session.execute(query)
            (result,) = results.one()
            return result

3. 修复会话生成方法错误 #

database.py中async_session_generator的参数名错误（class_1改为class_），同时简化get_session实现：

def async_session_generator(self):
    return sessionmaker(
        self._engine, expire_on_commit=False, class_=AsyncSession
    )

async def get_session(self):
    async_session = self.async_session_generator()
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

4. 调整模型依赖的导入时机 #

移除AsyncDatabaseSession类的模块级导入，仅在update_employee方法内导入Employee，并用字符串类型提示避免提前依赖：

class AsyncDatabaseSession:
    '''Handles database connections, updates the employee list, receives member role updates, that good stuff'''

    def __init__(self):
        self.employee_list = {}
        self._session = None
        self._engine = None

    # ... 其他方法保持不变 ...
    
    async def update_employee(self, user: "Employee"):
        """Updates an employee"""
        from helpers.models import Employee
        if not len(user.emp_under) > 0:
            await Employee.update(
                id=user.id, username=user.username, displayname=user.displayname, email=user.email,
                emp_type=user.emp_type, team=user.team, redline=user.redline, tier=user.tier, cur_kw=user.cur_kw,
                emp_over=user.emp_over
            )
        else:
            await Employee.update(
                id=user.id, username=user.username, displayname=user.displayname, email=user.email,
                emp_type=user.emp_type, redline=user.redline, tier=user.tier, cur_kw=user.cur_kw,
                emp_over=user.emp_over, emp_under=json.dumps(user.emp_under)
            )

注：原代码中user.user_id改为user.id，匹配Employee类的主键字段。

5. 规范项目初始化顺序 #

在项目入口文件（如main.py）中，先导入所有模型再初始化数据库，确保模型被Base元数据收录：

# main.py
from helpers import models
from helpers.database import async_db

async def main():
    await async_db.init()
    await async_db.create_all()
    # 其他业务逻辑

最终效果 #

通过以上调整，彻底解决循环导入问题，同时保留ModelAdmin类的增改查封装逻辑，数据库配置结构更清晰：

• base.py：统一管理SQLAlchemy基类
• models.py：定义数据模型与业务逻辑，延迟导入会话避免依赖冲突
• database.py：管理数据库连接与会话生成，内部导入模型避免循环引用

内容的提问来源于stack exchange，提问作者Zach Handley