orm query

Steven Chen · Steven Chen · commit f94c8ded2f2a · 2023-11-19T22:09:19.000-08:00
diff --git a/13_orm_query/__init__.py b/13_orm_query/__init__.py
diff --git a/13_orm_query/db_init.py b/13_orm_query/db_init.py
@@ -0,0 +1,49 @@
+import datetime
+
+from sqlalchemy import create_engine, String, ForeignKey
+from sqlalchemy.orm import DeclarativeBase, sessionmaker, Mapped, mapped_column, relationship
+from typing_extensions import Annotated
+from typing import List
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+engine = create_engine('mysql://root:test@localhost/testdb', echo=True)
+
+
+int_pk = Annotated[int, mapped_column(primary_key=True)]
+required_unique_string = Annotated[str, mapped_column(String(128), unique=True, nullable=False)]
+required_string = Annotated[str, mapped_column(String(128), nullable=False)]
+timestamp_not_null = Annotated[datetime.datetime, mapped_column(nullable=False)]
+
+
+class Department(Base):
+    __tablename__ = "department"
+
+    id: Mapped[int_pk]
+    name: Mapped[required_unique_string]
+
+    employees: Mapped[List["Employee"]] = relationship(back_populates="department")
+
+    def __repr__(self):
+        return f'id: {self.id}, name: {self.name}'
+
+
+class Employee(Base):
+    __tablename__ = "employee"
+
+    id: Mapped[int_pk]
+    dep_id: Mapped[int] = mapped_column(ForeignKey("department.id"))
+    name: Mapped[required_unique_string]
+    birthday: Mapped[timestamp_not_null]
+
+    department: Mapped[Department] = relationship(back_populates="employees")
+
+    def __repr__(self):
+        return f'id: {self.id}, dep_id: {self.dep_id}, name: {self.name}, birthday: {self.birthday}'
+
+
+Base.metadata.create_all(engine)
+Session = sessionmaker(bind=engine)
diff --git a/13_orm_query/query.py b/13_orm_query/query.py
@@ -0,0 +1,61 @@
+from sqlalchemy import select, insert, update, bindparam, delete
+from sqlalchemy.orm import outerjoin, aliased
+
+from db_init import Session, Department, Employee
+
+
+def execute_query(query):
+    result = session.execute(query)
+    for row in result:
+        print(row)
+
+
+def select_single_target():
+    query = select(Department).order_by(Department.name)
+    execute_query(query)
+
+
+def select_multiple():
+    query = select(Employee, Department).join(Employee.department, isouter=True)
+    execute_query(query)
+
+
+def select_with_alias():
+    emp_cls = aliased(Employee, name="emp")
+    dep_cls = aliased(Department, name="dep")
+    query = select(emp_cls, dep_cls).join(emp_cls.department.of_type(dep_cls))
+    execute_query(query)
+
+
+def select_fields():
+    query = select(Employee.name.label('emp_name'), Department.name.label('dep_name')).join_from(Employee, Department)
+    execute_query(query)
+
+
+def select_fields_outer():
+    query = select(Employee.name.label('emp_name'), Department.name.label('dep_name'))\
+        .select_from(outerjoin(Employee, Department))
+    execute_query(query)
+
+
+def where_obj():
+    dep = session.get(Department, 1)
+    # query = select(Employee).where(Employee.department == dep)
+    query = select(Employee).where(Employee.dep_id != dep.id)
+    execute_query(query)
+
+
+def select_contains():
+    emp = session.get(Employee, 1)
+    query = select(Department).where(Department.employees.contains(emp))
+    execute_query(query)
+
+
+session = Session()
+# select_single_target()
+# select_multiple()
+# select_with_alias()
+# select_fields()
+# select_fields_outer()
+# where_obj()
+select_contains()
