Abstracting away boiler plate sqlalchemy call to allow different options in a try/except block

Question

I'm trying to abstract my sqlalchemy query call, the complete unmodified call looks like this:

Session.query(User).options(FromCache('redis1')).filter(User.id == user_id).all()

The dynamic parts I need control over is User, redis1 and the last call all() I want to call something like above line with the possibility to change the things in bold above in the arguments.

My first try turned out like this:

# Abstracts the final chained method
def q(obj, action):
    return getattr(obj, action)()

# Removes a lot of boiler plate code
# but doesn't give me control over the cache argument
# which I need in the same method as the action above
def query(model):
    return Session.query(model).options(FromCache('redis1'))

# Called like this
result = q(query(User).filter(User.id == user_id), 'all')

But I'd obviously only want one function instead of two so I can control both the action and the model. The reason for this is that I'd like to wrap the .one() or .all() call in a try-except block that will go through different cache servers in the FromCache option in case the query fails. But doing this every time I write a query would be annoying.

I'm looking for something like this (code doesn't work, obviously):

def q(model, filters, action):
    data = None
    servers = ['redis1', 'redis2']
    for server in servers:
        try:
            call = Session.query(model).options(FromCache(redis)).filters_here_somehow()
            data = getattr(call, obj)() # Final call from action added
        except:
            continue
        break;
    return data

# and called something like this.. (or some other way)
result = q(User, the filter/between/limit/whatnot here, 'all')

Any ideas or am I totally way off base? Or is there a smarter way to do this?

Answer 1

the only part of this where something interesting has to happen is the "all()". The rest of it seems like you'd stick with a straight Query.

So here's one way pattern:

q = query(User).filter(User.id = user_id).limit(whatever)
results = from_redis(q, Query.all)

and here's a way we can do that:

def from_redis(query, meth):
    for server in ['r1', 'r2']:
        try:
            return meth(q.options(FromCache(server)))
        except Exception, e:
            continue
    else:
        raise e

but that might seem a little lame. Since you're using the caching recipe and subclassing Query, you could add your redis feature to it as well, sort of like:

class RedisQuery(CachingQuery):
    redis_servers = False

    @_generative(self):
    def using_redis(self):
        self.redis_servers = True

    def __iter__(self):
        if self.redis_servers:
            for server in ['r1', 'r2']:
                try:
                    return super(RedisQuery, self).options(FromCache(server)).__iter__()
                except Exception, e:
                    continue
            else:
                raise e
        else:
            return super(RedisQuery, self).__iter__()

that one you call like this:

q = query(User).filter(User.id = user_id).limit(whatever)
results = q.using_redis().all()

just some ideas.