出现IntegrityError时SQLAlchemy操作MariaDB引发内存泄漏求助

开发访问MariaDB的API时，使用SQLAlchemy进行数据库操作遇到异常内存泄漏：插入数据无错误时，事务结束后内存占用会正常回落；但当因表中唯一索引触发IntegrityError时，每次执行插入操作内存都会上涨且无法释放。

项目基于FastAPI+Uvicorn部署，已将问题精简为90行测试脚本并复现该现象（脚本会自动创建表，但需手动提前创建数据库etetete）。

环境信息 #

• Python版本：3.12
• MariaDB：Docker部署的最新版本
• 依赖包版本：

Package           Version
----------------- -------
greenlet          3.0.3
mariadb           1.1.10
packaging         24.0
pip               24.0
SQLAlchemy        2.0.29
typing_extensions 4.11.0

测试脚本 #

import time
from sqlalchemy import create_engine, Index, String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import Session


class Base(DeclarativeBase):
    pass


class Person(Base):
    __tablename__ = "person"
    __table_args__ = (
        Index("age", "children", "number", unique=True),
    )
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    age: Mapped[int]
    children: Mapped[int]
    number: Mapped[int]
    name: Mapped[str] = mapped_column(String(20))
    prename: Mapped[str] = mapped_column(String(20))


def generatePersons(nValues):
    persons = []
    for i in range(0, nValues):
        persons.append(Person(
            age=i,
            children=i,
            number=i,
            name="Smith",
            prename="John"
        ))
    return persons


# DB TEST FUNCTIONS---------------
def testDb(nValues):
    print("Start")
    engine = create_engine("mariadb+mariadbconnector://root:12345@localhost:3306/etetete")
    Base.metadata.create_all(engine)
    s = Session(engine)
    persons = generatePersons(nValues)
    try:
        s.add_all(persons)
        s.commit()
    except Exception as e:
        print("error" + str(e))
        s.rollback()
    finally:
        s.expunge_all()
        s.expire_all()
        s.close()
    engine.clear_compiled_cache()
    engine.dispose()
    print("End")


def testDbContextManager(nValues):
    print("Start")
    engine = create_engine("mariadb+mariadbconnector://root:12345@localhost:3306/etetete")
    Base.metadata.create_all(engine)
    persons = generatePersons(nValues)
    with Session(engine) as s:
        try:
            s.add_all(persons)
            s.commit()
        except Exception as e:
            print("error"+str(e))
    print("End")
# ---------------DB TEST FUNCTIONS


def executeTest():
    for i in range(0, 100):
        start = time.time()
        # testDb(200000)
        testDbContextManager(200000)
        end = time.time()
        res = end - start
        print("ExecutionTime: " + str(round(res, 2))+"s")
        input("Once again?")
        print()


executeTest()

已尝试的排查方案 #

• 使用/不使用上下文管理器管理Session
• 更换不同版本的SQLAlchemy
• 手动调用Python垃圾回收机制
• 先执行flush再执行commit
• 通过Windows任务管理器和memory-profiler工具监控内存变化

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