Consultas en SQLAlchemy 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()

Selección

from sqlalchemy import select

print(select(User))

SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account

row = session.execute(select(User)).first()
row

(User(id=1, name='spongebob', fullname='Spongebob Squarepants'),)

row[0]

User(id=1, name='spongebob', fullname='Spongebob Squarepants')

user = session.scalars(select(User)).first()
user

User(id=1, name='spongebob', fullname='Spongebob Squarepants')

print(select(User.name, User.fullname))

SELECT user_account.name, user_account.fullname
FROM user_account

row = session.execute(select(User.name, User.fullname)).first()
row

('spongebob', 'Spongebob Squarepants')

session.execute(
    select(User.name, Address).where(User.id == Address.user_id).order_by(Address.id)
).all()

[('spongebob', Address(id=1, email_address='spongebob@sqlalchemy.org')),
 ('sandy', Address(id=2, email_address='sandy@sqlalchemy.org')),
 ('sandy', Address(id=3, email_address='sandy@squirrelpower.org'))]

WHERE

from sqlalchemy import and_, or_

print(
    select(Address.email_address).where(
        and_(
            or_(User.name == "squidward", User.name == "sandy"),
            Address.user_id == User.id,
        )
    )
)

SELECT address.email_address
FROM address, user_account
WHERE (user_account.name = :name_1 OR user_account.name = :name_2) AND address.user_id = user_account.id

print(select(User).filter_by(name="spongebob", fullname="Spongebob Squarepants"))

SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = :name_1 AND user_account.fullname = :fullname_1

from sqlalchemy import func

with engine.connect() as conn:

    result = conn.execute(
        select(User.name, func.count(Address.id).label("count"))
        .join(Address)
        .group_by(User.name)
        .having(func.count(Address.id) > 1)
    )

    print(result.all())

[('sandy', 2)]

from sqlalchemy import desc, func

stmt = (
    select(Address.user_id, func.count(Address.id).label("num_addresses"))
    .group_by("user_id")
    .order_by("user_id", desc("num_addresses"))
)

print(stmt)

SELECT address.user_id, count(address.id) AS num_addresses
FROM address GROUP BY address.user_id ORDER BY address.user_id, num_addresses DESC

ALIAS

from sqlalchemy.orm import aliased

address_alias_1 = aliased(Address)
address_alias_2 = aliased(Address)

print(
    select(User)
    .join_from(User, address_alias_1)
    .where(address_alias_1.email_address == "patrick@aol.com")
    .join_from(User, address_alias_2)
    .where(address_alias_2.email_address == "patrick@gmail.com")
)

SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account JOIN address AS address_1 ON user_account.id = address_1.user_id JOIN address AS address_2 ON user_account.id = address_2.user_id
WHERE address_1.email_address = :email_address_1 AND address_2.email_address = :email_address_2

Subconsultas

subq = select(Address).where(~Address.email_address.like("%@aol.com")).subquery()

address_subq = aliased(Address, subq)

stmt = (
    select(User, address_subq)
    .join_from(User, address_subq)
    .order_by(User.id, address_subq.id)
)

with Session(engine) as session:
    for user, address in session.execute(stmt):
        print(f"{user} {address}")

User(id=1, name='spongebob', fullname='Spongebob Squarepants') Address(id=1, email_address='spongebob@sqlalchemy.org')
User(id=2, name='sandy', fullname='Sandy Cheeks') Address(id=2, email_address='sandy@sqlalchemy.org')
User(id=2, name='sandy', fullname='Sandy Cheeks') Address(id=3, email_address='sandy@squirrelpower.org')

cte_obj = select(Address).where(~Address.email_address.like("%@aol.com")).cte()

address_cte = aliased(Address, cte_obj)

stmt = (
    select(User, address_cte)
    .join_from(User, address_cte)
    .order_by(User.id, address_cte.id)
)

with Session(engine) as session:
    for user, address in session.execute(stmt):
        print(f"{user} {address}")

User(id=1, name='spongebob', fullname='Spongebob Squarepants') Address(id=1, email_address='spongebob@sqlalchemy.org')
User(id=2, name='sandy', fullname='Sandy Cheeks') Address(id=2, email_address='sandy@sqlalchemy.org')
User(id=2, name='sandy', fullname='Sandy Cheeks') Address(id=3, email_address='sandy@squirrelpower.org')

UNION

from sqlalchemy import union_all

stmt1 = select(User).where(User.name == "sandy")
stmt2 = select(User).where(User.name == "spongebob")
u = union_all(stmt1, stmt2)

orm_stmt = select(User).from_statement(u)

with Session(engine) as session:
    for obj in session.execute(orm_stmt).scalars():
        print(obj)

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

user_alias = aliased(User, u.subquery())
orm_stmt = select(user_alias).order_by(user_alias.id)

with Session(engine) as session:
    for obj in session.execute(orm_stmt).scalars():
        print(obj)

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