storage.py 文件源码

def filter(self, values: set):
        if not isinstance(values, set):
            raise ValueError
        if values:
            key = "{}:tasks:processed".format(self.name)
            lua = """
                local results = {}
                for _, e in pairs(ARGV) do
                    local x = redis.call('sismember', KEYS[1], e)
                    if x == 1 then
                        table.insert(results, e)
                    end
                end
                return results
            """
            script = self.db.register_script(lua)
            results = script(keys=[key], args={self.codec.dumps(v) for v in values})
            return values - {self.codec.loads(e) for e in results}