Reorganize

Programming Notes/python/sqlalchemy/orm.py

@@ -0,0 +1,129 @@
+from sqlalchemy import Column, Integer, String
+from sqlalchemy import create_engine
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Session
+from sqlalchemy.orm import registry
+from sqlalchemy.orm import relationship
+
+mapper_registry = registry()
+Base = mapper_registry.generate_base()
+
+# from sqlalchemy.orm import declarative_base
+# Base = declarative_base()
+
+
+
+class User(Base):
+    __tablename__ = 'user_account'
+
+    id = Column(Integer, primary_key=True)
+    name = Column(String(30))
+    fullname = Column(String)
+
+    addresses = relationship("Address", back_populates="user")
+
+    def __repr__(self):
+        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"
+
+
+
+class Address(Base):
+    __tablename__ = 'address'
+
+    id = Column(Integer, primary_key=True)
+    email_address = Column(String, nullable=False)
+    user_id = Column(Integer, ForeignKey('user_account.id'))
+
+    user = relationship("User", back_populates="addresses")
+
+    def __repr__(self):
+        return f"Address(id={self.id!r}, email_address={self.email_address!r})"
+
+
+engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
+Base.metadata.create_all(engine)
+
+
+session = Session(engine)
+
+
+# INSERT
+print('-----insert-----')
+squidward = User(name="squidward", fullname="Squidward Tentacles")
+krabs = User(name="ehkrabs", fullname="Eugene H. Krabs")
+spongebob = User(name='spongebob', fullname='Spongebob Squarepants')
+
+session.add(squidward)
+session.add(krabs)
+session.add(spongebob)
+print('new: ', session.new)
+
+saved_krabs = session.get(User, 2)
+print('krabs: ', saved_krabs)
+print('krabs is krabs: ', saved_krabs is krabs)
+session.commit()  # All changes are applied only after commit
+
+
+
+
+# SELECT
+print('-----select-----')
+from sqlalchemy import select
+
+stmt = select(User)
+first = session.execute(stmt).first()
+print(firs)
+
+_all = session.execute(stmt).scalars().all()
+print(_all)
+
+
+
+# UPDATE
+print('-----update-----')
+from sqlalchemy import update
+
+# Unit of work
+squid = session.execute(select(User).filter_by(name='squidward')).scalar_one()
+print('fetched squidward: ', squid)
+squid.fullname = 'Squidward Octopus'
+print('squid modified: ', squid in session.dirty)
+print('new squid status: ', session.execute(select(User).where(User.name == 'squidward')).first())
+
+# ORM enabled
+session.execute(
+    update(User).
+    where(User.name == "squidward").
+    values(fullname="Sandy Octopus Extraordinaire")
+)
+print('Squid after bulk update: ', squid.fullname)
+
+
+
+# DELETE
+print('-----delete-----')
+from sqlalchemy import delete
+
+krabs = session.get(User, 2)
+session.delete(krabs)
+session.execute(select(User).where(User.name == "ehkrabs")).first()
+print('Krabs present: ', krabs in session)
+
+# orm-enabled
+squidward = session.get(User, 1)
+session.execute(delete(User).where(User.name == "squidward"))
+print('Squidward present: ', squidward in session)
+
+
+
+# ROLLBACK
+print('-----rollback-----')
+print(squid.__dict__)
+session.rollback()
+print(squid.__dict__)
+squid.fullname
+print(squid.__dict__)
+print('Krabs present: ', krabs in session)
+
+
+session.close()