Created
September 5, 2023 07:22
-
-
Save wwnbb/c06899383e2cc1aa6dec96a9cd95fc3f to your computer and use it in GitHub Desktop.
This gist show how to use many to many relationships in async sqlalchemy v2
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import asyncio | |
| from sqlalchemy import Column, Integer, String, Table, ForeignKey, select | |
| from sqlalchemy.ext.asyncio import create_async_engine | |
| from sqlalchemy.orm import relationship | |
| from sqlalchemy.orm import joinedload | |
| from sqlalchemy.orm import declarative_base | |
| from sqlalchemy.ext.asyncio import async_sessionmaker | |
| Base = declarative_base() | |
| # Create the association table for many-to-many relationship | |
| student_course_association = Table( | |
| "student_course", | |
| Base.metadata, | |
| Column("student_id", Integer, ForeignKey("students.id")), | |
| Column("course_id", Integer, ForeignKey("courses.id")), | |
| ) | |
| # Create the Student model | |
| class Student(Base): | |
| __tablename__ = "students" | |
| id = Column(Integer, primary_key=True) | |
| name = Column(String) | |
| # Define the many-to-many relationship using the association table | |
| courses = relationship( | |
| "Course", secondary=student_course_association, back_populates="students" | |
| ) | |
| # Create the Course model | |
| class Course(Base): | |
| __tablename__ = "courses" | |
| id = Column(Integer, primary_key=True) | |
| name = Column(String) | |
| # Define the many-to-many relationship using the association table | |
| students = relationship( | |
| "Student", secondary=student_course_association, back_populates="courses" | |
| ) | |
| # Create an SQLite database and initialize it | |
| engine = create_async_engine("sqlite+aiosqlite:///test.db") | |
| # Create the tables | |
| async_session = async_sessionmaker(engine, expire_on_commit=False) | |
| async def main(): | |
| async with engine.begin() as conn: | |
| await conn.run_sync(Base.metadata.create_all) | |
| async with async_session() as session: | |
| # Create students and courses | |
| student1 = Student(name="Alice") | |
| student2 = Student(name="Bob") | |
| course1 = Course(name="Math") | |
| course2 = Course(name="Science") | |
| # Add courses to students | |
| student1.courses.append(course1) | |
| student1.courses.append(course2) | |
| student2.courses.append(course1) | |
| # Add students and courses to the session and commit | |
| session.add(student1) | |
| session.add(student2) | |
| await session.commit() | |
| # Query all students and use joinedload for courses | |
| students_with_courses = ( | |
| ( | |
| await session.execute( | |
| select(Student).options(joinedload(Student.courses)) | |
| ) | |
| ) | |
| .unique() | |
| .scalars() | |
| .all() | |
| ) | |
| # Iterate through each student and their courses | |
| for student in students_with_courses: | |
| print(f"Student: {student.name}") | |
| for course in student.courses: | |
| print(f" Course: {course.name}") | |
| if __name__ == "__main__": | |
| # Create a new session | |
| asyncio.run(main()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment