1. Michael Bayer
  2. sqlalchemy
  3. Issues

Issues

Issue #2438 resolved

Class.somerelation.using(some_alias) ? similar to of_type but with more control

Michael Bayer
repo owner created an issue

see the example of:

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base= declarative_base()

class Job(Base):
    __tablename__ = "job"

    id = Column(Integer, primary_key=True)
    container_id = Column(Integer, ForeignKey('data_container.id'))

class DataContainer(Base):
    __tablename__ = "data_container"

    id = Column(Integer, primary_key=True)
    jobs = relationship(Job)

query:

SELECT * FROM job JOIN 
datacontainer ON datacontainer.id = job.data_container_id 
WHERE NOT (EXISTS (SELECT 1 FROM job AS other_job 
WHERE other_job.data_container_id = datacontainer.id 
AND other_job.id < job.id))

using any(). The feature would allow us to say:

Job_A = aliased(Job)
DataContainer.jobs.using(Job_A).any(Job_A.id < Job.id)

naturally it would work elsewhere, though this wouldn't be the main usage:

query(DataContainer).join(DataContainer.jobs.using(Job_A))

Comments (12)

  1. Michael Bayer reporter

    OK here's the test for the above:

    from sqlalchemy import *
    from sqlalchemy.orm import *
    from sqlalchemy.ext.declarative import declarative_base
    
    Base= declarative_base()
    
    class Job(Base):
        __tablename__ = "job"
    
        id = Column(Integer, primary_key=True)
        container_id = Column(Integer, ForeignKey('data_container.id'))
    
    class DataContainer(Base):
        __tablename__ = "data_container"
    
        id = Column(Integer, primary_key=True)
        jobs = relationship(Job)
    
    Job_A = aliased(Job)
    s = Session()
    print s.query(Job).join(DataContainer.jobs).filter(DataContainer.jobs.of_type(Job_A).any(Job_A.id < Job.id))
    
  2. Michael Bayer reporter

    then patch attached, which works with the above test, and seems to pass the "hard" tests in orm so far. but this would need a crapload more tests. note we're just expanding out of_type() itself. we should also see if a with_polymorphic() can go in there.

  3. Michael Bayer reporter

    join seems to just work so far. this is weird huh. these all seem to do the right thing. let me jinx this totally, can it be that the aliasing architecture is finally approaching enough consistency that new usages just work with almost no changes ?

    from sqlalchemy import *
    from sqlalchemy.orm import *
    from sqlalchemy.ext.declarative import declarative_base
    
    Base= declarative_base()
    
    class Job(Base):
        __tablename__ = "job"
    
        id = Column(Integer, primary_key=True)
        container_id = Column(Integer, ForeignKey('data_container.id'))
    
    class SubJob(Job):
        __tablename__ = 'subjob'
        id = Column(Integer, ForeignKey('job.id'), primary_key=True)
    
    class DataContainer(Base):
        __tablename__ = "data_container"
    
        id = Column(Integer, primary_key=True)
        jobs = relationship(Job)
    
    s = Session()
    
    Job_A = aliased(Job)
    Job_P = with_polymorphic(Job, SubJob, aliased=True)
    
    print s.query(Job).join(DataContainer.jobs).filter(DataContainer.jobs.of_type(Job_A).any(Job_A.id < Job.id)), "\n"
    
    print s.query(Job).join(DataContainer.jobs).filter(DataContainer.jobs.of_type(Job_P).any(Job_P.id < Job.id)), "\n"
    
    print s.query(DataContainer).join(DataContainer.jobs.of_type(Job_A)), "\n"
    
    print s.query(DataContainer).join(DataContainer.jobs.of_type(with_polymorphic(Job, SubJob))), "\n"
    
    print s.query(DataContainer).join(DataContainer.jobs.of_type(Job_P)), "\n"
    print s.query(DataContainer).join(with_polymorphic(Job, SubJob, aliased=True), DataContainer.jobs), "\n"
    
  4. Michael Bayer reporter

    more tests needed, this is #1106

    e = create_engine("sqlite://", echo=True)
    Base.metadata.create_all(e)
    s = Session(e)
    
    s.add_all([   DataContainer(jobs=[
            SubJob(attr="s1"),
            SubJob(attr="s2")
        ](
    ))
    ])
    s.commit()
    
    Job_P = with_polymorphic(Job, SubJob, aliased=True)
    
    for dc in s.query(DataContainer).\
                join(DataContainer.jobs.of_type(Job_P)).\
                    options(contains_eager(DataContainer.jobs.of_type(Job_P))):
        for job in dc.jobs:
            print job.attr
    s.close()
    
    for dc in s.query(DataContainer).\
                    options(joinedload(DataContainer.jobs.of_type(Job_P))):
        for job in dc.jobs:
            print job.attr
    
  5. Log in to comment