0

我正在尝试抽象我的 sqlalchemy 查询调用，完整的未修改调用如下所示：

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

我需要控制的动态部分是User，redis1和最后一次调用all() 我想调用类似上面的代码，并且可以在参数中更改上面粗体的内容。

我的第一次尝试结果是这样的：

# 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')

但我显然只想要一个功能而不是两个，这样我就可以控制动作和模型。这样做的原因是我想将.one()或.all()调用包装在一个 try-except 块中，以防查询失败，该块将通过 FromCache 选项中的不同缓存服务器。但是每次我写查询时都这样做会很烦人。

我正在寻找这样的东西（代码不起作用，显然）：

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')

有什么想法还是我完全偏离了基础？还是有更聪明的方法来做到这一点？

4

1 回答 1

1

唯一需要发生有趣事情的部分是“all()”。其余部分似乎您会坚持使用直接查询。

所以这是一种方式模式：

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

这是我们可以做到的一种方法：

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

但这似乎有点蹩脚。由于您使用的是缓存配方和子类Query化，因此您也可以将您的 redis 功能添加到其中，有点像：

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__()

你这样称呼的那个：

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

只是一些想法。

于 2013-01-31T00:42:46.960 回答