stress_at_work_analysis/config/ORMtutorial.py

# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: percent
#       format_version: '1.3'
#       jupytext_version: 1.11.2
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# %%
import sqlalchemy

print(sqlalchemy.__version__)

# %%
from sqlalchemy import create_engine

engine = create_engine("sqlite:///:memory:", echo=True)

# %%
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

# %%
from sqlalchemy import Column, Integer, String


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    nickname = Column(String)

    def __repr__(self):
        return "<User(name='%s', fullname='%s', nickname='%s')>" % (
            self.name,
            self.fullname,
            self.nickname,
        )


# %%
print(User.__table__)

# %%
Base.metadata.create_all(engine)

# %%
ed_user = User(name="ed", fullname="Ed Jones", nickname="edsnickname")
print(ed_user.name)
print(ed_user.nickname)
print(str(ed_user.id))

# %%
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

# %% [markdown]
# # Adding and Updating Objects

# %%
ed_user = User(name="ed", fullname="Ed Jones", nickname="edsnickname")
session.add(ed_user)

# %%
our_user = session.query(User).filter_by(name="ed").first()
print(our_user)

# %%
print(ed_user is our_user)

# %%
session.add_all(
    [
        User(name="wendy", fullname="Wendy Williams", nickname="windy"),
        User(name="mary", fullname="Mary Contrary", nickname="mary"),
        User(name="fred", fullname="Fred Flintstone", nickname="freddy"),
    ]
)

# %%
ed_user.nickname = "eddie"

# %%
print(session.dirty)

# %%
print(session.new)

# %%
session.commit()

# %%
print(ed_user.id)

# %% [markdown]
# # Rolling back

# %%
ed_user.name = "Edwardo"

# %%
fake_user = User(name="fakeuser", fullname="Invalid", nickname="12345")
session.add(fake_user)

# %%
session.query(User).filter(User.name.in_(["Edwardo", "fakeuser"])).all()

# %%
session.rollback()
print(ed_user.name)
print(fake_user in session)

# %%
session.query(User).filter(User.name.in_(["ed", "fakeuser"])).all()