(self)
return schemas
Load all schemas
Load all schemas
Load all schemas
async def load(self) -> Dict[str, Dict]:
        """Load all schemas"""
        schemas = {}
        with await self.connection_manager() as redis:
            # Get & decode the api names
            api_names = list(await redis.smembers(self.schema_set_key()))
            api_names = [api_name.decode("utf8") for api_name in api_names]

            # Convert the api names into redis keys
            keys = [self.schema_key(api_name) for api_name in api_names]

            if not keys:
                return {}

            # Get the schemas from the keys
            encoded_schemas = await redis.mget(*keys)
            for api_name, schema in zip(api_names, encoded_schemas):
                # Schema may have expired
                if schema:
                    schemas[api_name] = json.loads(schema)
        return schemas
async def load ( self ) -> Dict [ str , Dict ] : schemas = { } with await self . connection_manager ( ) as redis : # Get & decode the api names api_names = list ( await redis . smembers ( self . schema_set_key ( ) ) ) api_names = [ api_name . decode ( "utf8" ) for api_name in api_names ] # Convert the api names into redis keys keys = [ self . schema_key ( api_name ) for api_name in api_names ] if not keys : return { } # Get the schemas from the keys encoded_schemas = await redis . mget ( * keys ) for api_name , schema in zip ( api_names , encoded_schemas ) : # Schema may have expired if schema : schemas [ api_name ] = json . loads ( schema ) return schemas