The right way to auto filter SQLAlchemy queries?

Question

I've just introspected a pretty nasty schema from a CRM app with sqlalchemy. All of the tables have a deleted column on them and I wanted to auto filter all those entities and relations flagged as deleted. Here's what I came up with:

---

class CustomizableQuery(Query):
    """An overridden sqlalchemy.orm.query.Query to filter entities

    Filters itself by BinaryExpressions
    found in :attr:`CONDITIONS`
    """

    CONDITIONS = []

    def __init__(self, mapper, session=None):
        super(CustomizableQuery, self).__init__(mapper, session)
        for cond in self.CONDITIONS:
            self._add_criterion(cond)

    def _add_criterion(self, criterion):
        criterion = self._adapt_clause(criterion, False, True)
        if self._criterion is not None:
            self._criterion = self._criterion & criterion
        else:
            self._criterion = criterion

And it's used like this:

class UndeletedContactQuery(CustomizableQuery):
    CONDITIONS = [contacts.c.deleted != True]

    def by_email(self, email_address):
        return EmailInfo.query.by_module_and_address('Contacts', email_address).contact

    def by_username(self, uname):
        return self.filter_by(twod_username_c=uname).one()

class Contact(object):
    query = session.query_property(UndeletedContactQuery)

Contact.query.by_email('[email protected]')

EmailInfo is the class that's mapped to the join table between emails and the other Modules that they're related to.

Here's an example of a mapper:

contacts_map = mapper(Contact, join(contacts, contacts_cstm), {
    '_emails': dynamic_loader(EmailInfo,
                              foreign_keys=[email_join.c.bean_id],
                              primaryjoin=contacts.c.id==email_join.c.bean_id,
                              query_class=EmailInfoQuery),
    })

class EmailInfoQuery(CustomizableQuery):

    CONDITIONS = [email_join.c.deleted != True]
    # More methods here

---

This gives me what I want in that I've filtered out all deleted Contacts. I can also use this as the query_class argument to dynamic_loader in my mappers - However...

1. Is there a better way to do this, I'm not really happy with poking around with the internals of a compicated class like Query as I am.
2. Has anyone solved this in a different way that they can share?

Answer 1

You can map to a select. Like this:

mapper(EmailInfo, select([email_join], email_join.c.deleted == False))

Answer 2

I'd consider seeing if it was possible to create views for these tables that filter out the deleted elements, and then you might be able to map directly to that view instead of the underlying table, at least for querying operations. However I've never tried this myself!