Introducción a SQLAlchemy ORM

• Ultima modificación: Mar 6, 2024 | YouTube

Declaración de modelos en el ORM

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import declarative_base, relationship

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", cascade="all, delete-orphan"
    )

    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"), nullable=False)
    user = relationship("User", back_populates="addresses")

    def __repr__(self):
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

Creación de una máquina

from sqlalchemy import create_engine

engine = create_engine(
    "sqlite+pysqlite:///:memory:",
    echo=False,
    future=True,
)

Emisión CREATE TABLE DDL

Base.metadata.create_all(engine)

Creación de objetos y persistencia

from sqlalchemy.orm import Session

with Session(engine) as session:

    spongebob = User(
        name="spongebob",
        fullname="Spongebob Squarepants",
        addresses=[Address(email_address="spongebob@sqlalchemy.org")],
    )

    sandy = User(
        name="sandy",
        fullname="Sandy Cheeks",
        addresses=[
            Address(email_address="sandy@sqlalchemy.org"),
            Address(email_address="sandy@squirrelpower.org"),
        ],
    )

    patrick = User(
        name="patrick",
        fullname="Patrick Star",
    )

    session.add_all(
        [
            spongebob,
            sandy,
            patrick,
        ]
    )

    session.commit()

SELECT

from sqlalchemy import select

session = Session(engine)

stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))

for user in session.scalars(stmt):
    print(user)

User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')

SELECT … JOIN

stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "sandy")
    .where(Address.email_address == "sandy@sqlalchemy.org")
)

sandy_address = session.scalars(stmt).one()
sandy_address

Address(id=2, email_address='sandy@sqlalchemy.org')

Actualización o adición de datos

#
# SELECT ... WHERE
#
stmt = select(User).where(User.name == "patrick")

#
# Extrae el registro
#
patrick = session.scalars(stmt).one()

#
# Agrega nueva información
#
patrick.addresses.append(Address(email_address="patrickstar@sqlalchemy.org"))

sandy_address.email_address = "sandy_cheeks@sqlalchemy.org"

session.commit()

Borrado de registros

for user in session.scalars(select(User)):
    print(user)

User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
User(id=3, name='patrick', fullname='Patrick Star')

#
# Borra un campo en particular
#
sandy = session.get(User, 2)
sandy.addresses.remove(sandy_address)

for user in session.scalars(select(User)):
    print(user)

User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
User(id=3, name='patrick', fullname='Patrick Star')

#
# Emite el cambio sin el commit()
#
session.flush()

for user in session.scalars(select(User)):
    print(user)

User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
User(id=3, name='patrick', fullname='Patrick Star')

#
# Borra un registro completo
#
session.delete(patrick)

for user in session.scalars(select(User)):
    print(user)

User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')

#
# Emite los cambios
#
session.commit()