SQLAlchemy 中的嵌套一对多关系

Question

所以我正在创建一个 LMS 系统，我已经尝试执行下面的代码以将一个深度嵌套的关系结构插入到数据库中。执行代码后，我收到错误消息。在这种情况下我做错了什么？

code:

async def create_program(db: AsyncSession, program: ProgramSchema) -> Program:
    new_program = Program(title=program.title, slug=program.slug)
    new_courses = []
    new_lessons = []
    new_sections = []

    for course in program.courses:
        new_course = Course(title=course.title, slug=course.slug)
        for lesson in course.lessons:
            new_lesson = Lesson(title=lesson.title, slug=lesson.slug)
            for section in lesson.sections:
                new_section = Section(
                    title=section.title,
                    slug=section.slug,
                    body=section.body,
                )
                new_lesson.sections.append(new_section)
                new_sections.append(new_section)
            new_course.lessons.append(new_lesson)
            new_lessons.append(new_lesson)
        new_program.courses.append(new_course)
        new_courses.append(new_course)
    db.add(new_program)
    db.add_all(new_courses)
    db.add_all(new_lessons)
    db.add_all(new_sections)
    await db.commit()
    await db.refresh(new_program)
    return new_program

error

This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: greenlet_spawn has not been called; can't call await_() here. Was IO attempted in an unexpected place? (Background on this error at: http://sqlalche.me/e/14/xd2s) (Background on this error at: http://sqlalche.me/e/14/7s2a)
./repositories/database_repository.py:24: RuntimeWarning: coroutine 'AsyncSession.rollback' was never awaited
  session.rollback()
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

Models:

class Program(Base):
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String, nullable=False, unique=True)
    slug = Column(String, nullable=False, unique=True)
    courses = relationship("Course", backref="program")
    created_on = Column(
        DateTime(timezone=True), default=datetime.utcnow, nullable=False
    )
    updated_on = Column(
        DateTime(timezone=True),
        default=datetime.utcnow,
        onupdate=datetime.utcnow(),
        nullable=False,
    )


class Course(Base):
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String, nullable=False, unique=True)
    slug = Column(String, nullable=False, unique=True)
    program_id = Column(Integer, ForeignKey("program.id"))
    lessons = relationship("Lesson", backref="course")
    likes = Column(Integer, nullable=False, default=0)
    dislikes = Column(Integer, nullable=False, default=0)
    created_on = Column(
        DateTime(timezone=True), default=datetime.utcnow, nullable=False
    )
    updated_on = Column(
        DateTime(timezone=True),
        default=datetime.utcnow,
        onupdate=datetime.utcnow(),
        nullable=False,
    )


class Lesson(Base):
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String, nullable=False, unique=True)
    slug = Column(String, nullable=False, unique=True)
    course_id = Column(Integer, ForeignKey("course.id"))
    sections = relationship("Section", backref="lesson")
    likes = Column(Integer, nullable=False, default=0)
    dislikes = Column(Integer, nullable=False, default=0)
    created_on = Column(
        DateTime(timezone=True), default=datetime.utcnow, nullable=False
    )
    updated_on = Column(
        DateTime(timezone=True),
        default=datetime.utcnow,
        onupdate=datetime.utcnow(),
        nullable=False,
    )


class Section(Base):
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String, nullable=False, unique=True)
    slug = Column(String, nullable=False, unique=True)
    body = Column(String, nullable=True)
    lesson_id = Column(Integer, ForeignKey("lesson.id"))
    likes = Column(Integer, nullable=False, default=0)
    dislikes = Column(Integer, nullable=False, default=0)
    created_on = Column(
        DateTime(timezone=True), default=datetime.utcnow, nullable=False
    )
    updated_on = Column(
        DateTime(timezone=True),
        default=datetime.utcnow,
        onupdate=datetime.utcnow(),
        nullable=False,
    )

Structure:

Program>List of Courses>List of Lessons> List of Sections

Answer 1

我在关系函数中添加了 , lazy="immediate"，它起作用了。

SQLAlchemy 中的嵌套一对多关系

Nested one-to-many relationships in SQLAlchemy

python

sqlalchemy

pydantic

fastapi