many to many orm

Steven Chen · Steven Chen · commit 1fdc70a2a1e3 · 2023-11-11T01:45:42.000-08:00
diff --git a/11_many_to_many_orm/__init__.py b/11_many_to_many_orm/__init__.py
diff --git a/11_many_to_many_orm/db_init.py b/11_many_to_many_orm/db_init.py
@@ -0,0 +1,53 @@
+import datetime
+
+from sqlalchemy import create_engine, String, ForeignKey, Table, Column
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker, Mapped, mapped_column, relationship
+from typing_extensions import Annotated
+from typing import List, Set
+
+
+engine = create_engine('mysql://root:test@localhost/testdb', echo=True)
+Base = declarative_base()
+
+
+int_pk = Annotated[int, mapped_column(primary_key=True)]
+required_unique_name = Annotated[str, mapped_column(String(128), unique=True, nullable=False)]
+required_string = Annotated[str, mapped_column(String(128), nullable=False)]
+
+
+association_table = Table(
+    "user_role",
+    Base.metadata,
+    Column("user_id", ForeignKey("users.id"), primary_key=True),
+    Column("role_id", ForeignKey("roles.id"), primary_key=True)
+)
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int_pk]
+    name: Mapped[required_unique_name]
+    password: Mapped[required_string]
+
+    roles: Mapped[List["Role"]] = relationship(secondary=association_table, lazy=False, back_populates="users")
+
+    def __repr__(self):
+        return f'id: {self.id}, name: {self.name}'
+
+
+class Role(Base):
+    __tablename__ = "roles"
+
+    id: Mapped[int_pk]
+    name: Mapped[required_unique_name]
+
+    users: Mapped[List["User"]] = relationship(secondary=association_table, lazy=True, back_populates="roles")
+
+    def __repr__(self):
+        return f'id: {self.id}, name: {self.name}'
+
+
+Base.metadata.create_all(engine)
+Session = sessionmaker(bind=engine)
diff --git a/11_many_to_many_orm/query.py b/11_many_to_many_orm/query.py
@@ -0,0 +1,38 @@
+from db_init import Session, User, Role
+
+
+def insert_records(s):
+    role1 = Role(name="Admin")
+    role2 = Role(name="Operator")
+
+    user1 = User(name="Jack", password="111")
+    user2 = User(name="Tom", password="222")
+    user3 = User(name="Mary", password="333")
+
+    user1.roles.append(role1)
+    user1.roles.append(role2)
+
+    user2.roles.append(role1)
+    user3.roles.append(role2)
+
+    s.add_all([user1, user2, user3])
+
+    s.commit()
+
+
+def select_user(s):
+    u = s.query(User).filter(User.id == 1).one()
+    print(u)
+    print(u.roles)
+
+
+def select_role(s):
+    r = s.query(Role).filter(Role.id == 2).one()
+    print(r)
+    print(r.users)
+
+
+session = Session()
+# insert_records(session)
+# select_user(session)
+select_role(session)
