one to one orm

Steven Chen · Steven Chen · commit 660934545c67 · 2023-11-12T01:16:06.000-08:00
diff --git a/12_one_to_one_orm/__init__.py b/12_one_to_one_orm/__init__.py
diff --git a/12_one_to_one_orm/db_init.py b/12_one_to_one_orm/db_init.py
@@ -0,0 +1,46 @@
+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_string = Annotated[str, mapped_column(String(128), unique=True, nullable=False)]
+required_string = Annotated[str, mapped_column(String(128), nullable=False)]
+
+
+class Employee(Base):
+    __tablename__ = "employee"
+
+    id: Mapped[int_pk]
+    name: Mapped[required_unique_string]
+    computer_id: Mapped[int] = mapped_column(ForeignKey("computer.id"), nullable=True)
+
+    computer = relationship("Computer", lazy=False, back_populates="employee")
+
+    def __repr__(self):
+        return f'id: {self.id}, name: {self.name}'
+
+
+class Computer(Base):
+    __tablename__ = "computer"
+
+    id: Mapped[int_pk]
+    model: Mapped[required_string]
+    number: Mapped[required_unique_string]
+
+    employee = relationship("Employee", lazy=True, back_populates="computer")
+
+    def __repr__(self):
+        return f'id: {self.id}, model: {self.model}, number: {self.number}'
+
+
+Base.metadata.create_all(engine)
+Session = sessionmaker(bind=engine)
diff --git a/12_one_to_one_orm/query.py b/12_one_to_one_orm/query.py
@@ -0,0 +1,47 @@
+from db_init import Session, Employee, Computer
+
+
+def insert(s):
+    c1 = Computer(model="Dell", number="1111")
+    c2 = Computer(model="Surface", number="2222")
+    c3 = Computer(model="MacBook Pro", number="3333")
+
+    e1 = Employee(name="Jack", computer=c1)
+    e2 = Employee(name="Mary", computer=c2)
+    e3 = Employee(name="Tome", computer=c3)
+
+    s.add_all([e1, e2, e3])
+
+    s.commit()
+
+
+def select(s):
+    e = s.query(Employee).filter(Employee.id == 1).scalar()
+    if e:
+        print(e)
+        print(e.computer)
+
+    c = s.query(Computer).filter(Computer.id == 2).scalar()
+    if c:
+        print(c)
+        print(c.employee)
+
+
+def update_1(s):
+    s.query(Employee).filter(Employee.id == 3).update({Employee.computer_id: None})
+    s.commit()
+
+
+def update_2(s):
+    c = s.query(Computer).filter(Computer.id == 3).scalar()
+    e = s.query(Employee).filter(Employee.id == 3).scalar()
+    if c and e:
+        e.computer = c
+        s.commit()
+
+
+session = Session()
+# insert(session)
+# select(session)
+# update_1(session)
+update_2(session)
