Manipulación de datos con el ORM

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

Preparación

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})"

from sqlalchemy import create_engine

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

Base.metadata.create_all(engine)

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()

Adición de objetos a una sesión

squidward = User(name="squidward", fullname="Squidward Tentacles")
krabs = User(name="ehkrabs", fullname="Eugene H. Krabs")

squidward

User(id=None, name='squidward', fullname='Squidward Tentacles')

krabs

User(id=None, name='ehkrabs', fullname='Eugene H. Krabs')

session = Session(engine)
session.add(squidward)
session.add(krabs)

#
# Los objetos pendientes por insertar se ven en session.new
#
session.new

IdentitySet([User(id=None, name='squidward', fullname='Squidward Tentacles'), User(id=None, name='ehkrabs', fullname='Eugene H. Krabs')])

#
# Cuando se emiten los cambios a la BD se hace un flush.
#
session.flush()

#
# No hay cambios pendientes
#
session.new

IdentitySet([])

Claves primarias autogeneradas

squidward.id

4

krabs.id

5

Obtención de objetos por la clave primaria

some_squidward = session.get(User, 4)
some_squidward

User(id=4, name='squidward', fullname='Squidward Tentacles')

some_squidward is squidward

True

Commiting

session.commit()

Actualización de objetos

from sqlalchemy import select

sandy = session.execute(select(User).filter_by(name="sandy")).scalar_one()
sandy

User(id=2, name='sandy', fullname='Sandy Cheeks')

#
# Se altera el atributo de un objeto
#
sandy.fullname = "Sandy Squirrel"

#
# El objeto pertenece a una colección "dirty"
#
sandy in session.dirty

True

sandy_fullname = session.execute(select(User.fullname).where(User.id == 2)).scalar_one()

print(sandy_fullname)

Sandy Squirrel

sandy in session.dirty

False

UPDATE en el ORM

from sqlalchemy import update

session.execute(
    update(User)
    .where(User.name == "sandy")
    .values(fullname="Sandy Squirrel Extraordinaire")
)

<sqlalchemy.engine.cursor.CursorResult at 0x110b6a4c0>

sandy.fullname

'Sandy Squirrel Extraordinaire'

Borrado de objetos ORM

patrick = session.get(User, 3)

session.delete(patrick)

session.execute(select(User).where(User.name == "patrick")).first()

patrick in session

False

from sqlalchemy import delete

squidward = session.get(User, 4)

session.execute(delete(User).where(User.name == "squidward"))

<sqlalchemy.engine.cursor.CursorResult at 0x110b88ac0>

squidward in session

False

Rolling back

session.rollback()

sandy.__dict__

{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState at 0x110b464c0>}

sandy.fullname

'Sandy Cheeks'

sandy.__dict__

{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState at 0x110b464c0>,
 'id': 2,
 'name': 'sandy',
 'fullname': 'Sandy Cheeks'}

patrick in session

True

session.execute(select(User).where(User.name == "patrick")).scalar_one() is patrick

True

Cerrado de una sesión

session.close()