1from hashlib import sha1 2 3from six import string_types 4from ..type import GraphQLSchema 5 6from .base import GraphQLBackend 7 8# Necessary for static type checking 9if False: # flake8: noqa 10 from typing import Dict, Optional, Hashable 11 from .base import GraphQLDocument 12 13_cached_schemas = {} # type: Dict[GraphQLSchema, str] 14 15_cached_queries = {} # type: Dict[str, str] 16 17 18def get_unique_schema_id(schema): 19 # type: (GraphQLSchema) -> str 20 """Get a unique id given a GraphQLSchema""" 21 assert isinstance(schema, GraphQLSchema), ( 22 "Must receive a GraphQLSchema as schema. Received {}" 23 ).format(repr(schema)) 24 25 if schema not in _cached_schemas: 26 _cached_schemas[schema] = sha1(str(schema).encode("utf-8")).hexdigest() 27 return _cached_schemas[schema] 28 29 30def get_unique_document_id(query_str): 31 # type: (str) -> str 32 """Get a unique id given a query_string""" 33 assert isinstance(query_str, string_types), ( 34 "Must receive a string as query_str. Received {}" 35 ).format(repr(query_str)) 36 37 if query_str not in _cached_queries: 38 _cached_queries[query_str] = sha1(str(query_str).encode("utf-8")).hexdigest() 39 return _cached_queries[query_str] 40 41 42class GraphQLCachedBackend(GraphQLBackend): 43 """GraphQLCachedBackend will cache the document response from the backend 44 given a key for that document""" 45 46 def __init__( 47 self, 48 backend, # type: GraphQLBackend 49 cache_map=None, # type: Optional[Dict[Hashable, GraphQLDocument]] 50 use_consistent_hash=False, # type: bool 51 ): 52 # type: (...) -> None 53 assert isinstance( 54 backend, GraphQLBackend 55 ), "Provided backend must be an instance of GraphQLBackend" 56 if cache_map is None: 57 cache_map = {} 58 self.backend = backend 59 self.cache_map = cache_map 60 self.use_consistent_hash = use_consistent_hash 61 62 def get_key_for_schema_and_document_string(self, schema, request_string): 63 # type: (GraphQLSchema, str) -> int 64 """This method returns a unique key given a schema and a request_string""" 65 if self.use_consistent_hash: 66 schema_id = get_unique_schema_id(schema) 67 document_id = get_unique_document_id(request_string) 68 return hash((schema_id, document_id)) 69 return hash((schema, request_string)) 70 71 def document_from_string(self, schema, request_string): 72 # type: (GraphQLSchema, str) -> Optional[GraphQLDocument] 73 """This method returns a GraphQLQuery (from cache if present)""" 74 key = self.get_key_for_schema_and_document_string(schema, request_string) 75 if key not in self.cache_map: 76 self.cache_map[key] = self.backend.document_from_string( 77 schema, request_string 78 ) 79 80 return self.cache_map[key] 81