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

しかし、アクションとモデルの両方を制御できるように、関数を 2 つではなく 1 つだけにしたいのは明らかです。この理由は、クエリが失敗した場合に FromCache オプションで異なるキャッシュ サーバーを通過する try-except ブロックで.one()または.all()呼び出しをラップしたいからです。しかし、クエリを書くたびにこれを行うのは面倒です。

私はこのようなものを探しています（コードは明らかに機能しません）：

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 に答える