Query.count method to act like Query.select method when joining tables

Incorrect count results returned with the Query.count method when custom joins are passed to the count method. The count api should perform the same as the recent Query.select makeover.

Here is a test that compares the count to select results. I've included a function that is performing the same way the select function tackles the problem. I don't think this is the correct solution for everyone though but it works for this test case.

from sqlalchemy import *

engine = create_engine ( 'postgres://localhost/test' )

dbmeta = BoundMetaData(engine)


owners = Table ( 'owners', dbmeta ,
        Column ( 'id', Integer, primary_key=True, nullable=False ) )

class Owner(object):
        pass

mapper(Owner,owners)

categories=Table( 'categories', dbmeta,
        Column ( 'id', Integer,primary_key=True, nullable=False ),
        Column ( 'name', VARCHAR(20), index=True ) )

class Category(object):
        pass
mapper(Category,categories)

tests = Table ( 'tests', dbmeta ,
        Column ( 'id', Integer, primary_key=True, nullable=False ),
        Column ( 'owner_id',Integer, ForeignKey('owners.id'), nullable=False,index=True ),
        Column ( 'category_id', Integer, ForeignKey('categories.id'),nullable=False,index=True ))

class Test(object):
        pass

Test.mapper=mapper(Test,tests,properties={'owner':relation(Owner,backref='tests'),'category':relation(Category)})

options = Table ( 'options', dbmeta ,
        Column ( 'test_id', Integer, ForeignKey ( 'tests.id' ), primary_key=True, nullable=False ),
        Column ( 'owner_id', Integer, ForeignKey ( 'owners.id' ), primary_key=True, nullable=False ),
        Column ( 'someoption', Boolean, PassiveDefault('false'), nullable=False ) )

class Option(object):
        pass
mapper(Option,options,properties={'owner':relation(Owner),'test':relation(Test)})

Test.mapper.add_property ( 'owner_option', relation(Option,primaryjoin=and_(tests.c.id==options.c.test_id,tests.c.owner_id==options.c.owner_id),uselist=False ) )

dbmeta.drop_all()
dbmeta.create_all()


s=create_session()

# an owner
o=Owner()
s.save(o)

# owner a has 3 tests, one of which he has specified options for
c=Category()
c.name='Some Category'
s.save(c)

for i in range(3):
        t=Test()
        t.owner=o
        t.category=c
        s.save(t)
        if i==1:
                op=Option()  
                op.someoption=True
                t.owner_option=op
        if i==2:
                op=Option()  
                t.owner_option=op

s.flush()


# now test
s.clear()


def ctest(self,whereclause=None,params=None,**kwargs):
        #sel=self.compile(whereclause,**kwargs)
        #print "compiled: %s" % str(sel)
        from_obj=kwargs.pop('from_obj',[       alltables = [](])
)
        for l in [for x in from_obj](sql_util.TableFinder(x)):
                alltables += l
        if self.table not in alltables:
                from_obj.append(self.table)
        sel=select(['count(*)']('count(*)'),from_obj=from_obj)
        if whereclause:
                sel.append_whereclause(whereclause)
        return sel.scalar()

# uncomment for correct result ( but perhaps not correct implementation)
#Query.count=ctest

print "Attempting q.count with eagerload"
q=s.query(Test).options(eagerload('category'))

l=q.count( and_(tests.c.owner_id==1,or_(options.c.someoption==None,options.c.someoption==False)),
        from_obj=[tests.outerjoin(options,and_(tests.c.id==options.c.test_id,tests.c.owner_id==options.c.owner_id))](tests.outerjoin(options,and_(tests.c.id==options.c.test_id,tests.c.owner_id==options.c.owner_id))))
print l

Comments (5)