subquery load options

subquery_vs_no_subquery.py

from sqlalchemy import (
    create_engine,
    inspect,
    Column,
    ForeignKey,
    Integer,
    MetaData,
    String,
    Table,
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import (
    defer,
    joinedload,
    mapper,
    relation,
    sessionmaker,
    subqueryload,
)

metadata = MetaData()


class ToolShedRepository:
    def __repr__(self):
        return f'ts-repo-{self.id}'


class ToolDependency:
    def __repr__(self):
        return f't-dep-{self.id}'


ToolShedRepository.table = Table("tool_shed_repository", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(255)),
    Column("metadata", String))

ToolDependency.table = Table("tool_dependency", metadata,
    Column("id", Integer, primary_key=True),
    Column("tool_shed_repository_id", Integer, ForeignKey("tool_shed_repository.id")),
    Column("name", String(255)))

mapper(ToolShedRepository, ToolShedRepository.table,
    properties=dict(
        tool_dependencies=relation(ToolDependency,
            primaryjoin=(ToolShedRepository.table.c.id == ToolDependency.table.c.tool_shed_repository_id),
            backref='tool_shed_repository'),
))

mapper(ToolDependency, ToolDependency.table)


def load():
    session = Session()
    # create 2 tool-repos w/3 tool-dependencies each
    for i in range(2):
        r = ToolShedRepository()
        session.add(r)
        for j in range(3):
            d = ToolDependency()
            r.tool_dependencies.append(d)
            session.add(d)
    session.commit()
    session.close()


def run_subquery():
    session = Session()
    result = session.query(ToolShedRepository).options(
        defer(ToolShedRepository.metadata),
        joinedload('tool_dependencies').subqueryload('tool_shed_repository').options(
            defer(ToolShedRepository.metadata)
        ),
    ).all()
    session.close()
    verify(result)


def run_no_subquery():
    session = Session()
    result = session.query(ToolShedRepository).options(
        defer(ToolShedRepository.metadata), joinedload('tool_dependencies')
    ).all()
    session.close()
    verify(result)


def verify(result):
    assert len(result) == 2
    for tr in result:
        print(tr)
        for td in tr.tool_dependencies:
            assert len(tr.tool_dependencies) == 3
            print('',td)
    
    insp = inspect(result[0])
    print('callables:', insp.callables)
    assert 'metadata' in insp.callables, 'deferred `ToolShedRepository.metadata` should be available as a callable'


engine = create_engine('sqlite:///:memory:', echo=True)
metadata.create_all(engine)
Session = sessionmaker(bind=engine)
load()
run_subquery()
run_no_subquery()


# called for both
# ---------------
# SELECT
#     tool_shed_repository.id AS tool_shed_repository_id,
#     tool_shed_repository.name AS tool_shed_repository_name,
#     tool_dependency_1.id AS tool_dependency_1_id,
#     tool_dependency_1.tool_shed_repository_id AS tool_dependency_1_tool_shed_repository_id,
#     tool_dependency_1.name AS tool_dependency_1_name 
# FROM tool_shed_repository
# LEFT OUTER JOIN tool_dependency AS tool_dependency_1 ON tool_shed_repository.id = tool_dependency_1.tool_shed_repository_id


# called for run_subquery() only
# ------------------------------
# SELECT
#     tool_shed_repository.id AS tool_shed_repository_id,
#     tool_shed_repository.name AS tool_shed_repository_name,
#     tool_dependency_1.tool_shed_repository_id AS tool_dependency_1_tool_shed_repository_id 
# FROM 
#     (SELECT tool_shed_repository.id AS tool_shed_repository_id 
#      FROM tool_shed_repository
#     ) AS anon_1
# JOIN tool_dependency AS tool_dependency_1 ON anon_1.tool_shed_repository_id = tool_dependency_1.tool_shed_repository_id
# JOIN tool_shed_repository ON tool_shed_repository.id = tool_dependency_1.tool_shed_repository_id