130 lines
3.0 KiB
Python
Executable File
130 lines
3.0 KiB
Python
Executable File
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()
|