Added initial version of static typing

Author: syrusakbary

.gitignore

@@ -67,3 +67,7 @@ target/
 
 # OS X
 .DS_Store
+/.mypy_cache
+.pyre
+/.vscode
+/type_info.json

conftest.py

@@ -0,0 +1,30 @@
+# Configuration for pytest to automatically collect types.
+# Thanks to Guilherme Salgado.
+import pytest
+
+try:
+ import pyannotate_runtime
+ PYANOTATE_PRESENT = True
+except ImportError:
+ PYANOTATE_PRESENT = False
+
+if PYANOTATE_PRESENT:
+ def pytest_collection_finish(session):
+ """Handle the pytest collection finish hook: configure pyannotate.
+ Explicitly delay importing `collect_types` until all tests have
+ been collected. This gives gevent a chance to monkey patch the
+ world before importing pyannotate.
+ """
+ from pyannotate_runtime import collect_types
+ collect_types.init_types_collection()
+
+ @pytest.fixture(autouse=True)
+ def collect_types_fixture():
+ from pyannotate_runtime import collect_types
+ collect_types.resume()
+ yield
+ collect_types.pause()
+
+ def pytest_sessionfinish(session, exitstatus):
+ from pyannotate_runtime import collect_types
+ collect_types.dump_stats("type_info.json")

graphql/backend/__init__.py

@@ -9,17 +9,22 @@
 from .decider import GraphQLDeciderBackend
 from .cache import GraphQLCachedBackend
 
+if False:
+ from typing import Union
+
 _default_backend = None
 
 
 def get_default_backend():
+ # type: () -> GraphQLCoreBackend
  global _default_backend
  if _default_backend is None:
  _default_backend = GraphQLCoreBackend()
  return _default_backend
 
 
 def set_default_backend(backend):
+ # type: (GraphQLCoreBackend) -> None
  global _default_backend
  assert isinstance(
  backend, GraphQLBackend

graphql/backend/base.py

@@ -4,6 +4,11 @@
 from abc import ABCMeta, abstractmethod
 import six
 
+if False:
+ from typing import Dict, Optional, Union, Callable
+ from ..language.ast import Document
+ from ..type.schema import GraphQLSchema
+
 
 class GraphQLBackend(six.with_metaclass(ABCMeta)):
  @abstractmethod
@@ -15,35 +20,38 @@ def document_from_string(self, schema, request_string):
 
 class GraphQLDocument(object):
  def __init__(self, schema, document_string, document_ast, execute):
+ # type: (GraphQLSchema, str, Document, Callable) -> None
  self.schema = schema
  self.document_string = document_string
  self.document_ast = document_ast
  self.execute = execute
 
  @cached_property
  def operations_map(self):
- '''
+ # type: () -> Dict[Union[str, None], str]
+ """
  returns a Mapping of operation names and it's associated types.
  E.g. {'myQuery': 'query', 'myMutation': 'mutation'}
- '''
+ """
  document_ast = self.document_ast
- operations = {}
+ operations = {} # type: Dict[Union[str, None], str]
  for definition in document_ast.definitions:
  if isinstance(definition, ast.OperationDefinition):
  if definition.name:
- operation_name = definition.name.value
+ operations[definition.name.value] = definition.operation
  else:
- operation_name = None
- operations[operation_name] = definition.operation
+ operations[None] = definition.operation
+
  return operations
 
  def get_operation_type(self, operation_name):
- '''
+ # type: (Optional[str]) -> Optional[str]
+ """
  Returns the operation type ('query', 'mutation', 'subscription' or None)
  for the given operation_name.
  If no operation_name is provided (and only one operation exists) it will return the
  operation type for that operation
- '''
+ """
  operations_map = self.operations_map
  if not operation_name and len(operations_map) == 1:
  return next(iter(operations_map.values()))

graphql/backend/cache.py

@@ -4,12 +4,17 @@
 
 from .base import GraphQLBackend
 
-_cached_schemas = {}
+if False:
+ from typing import Any, Dict, Optional, Union, Tuple, Hashable
+ from .base import GraphQLDocument
 
-_cached_queries = {}
+_cached_schemas = {} # type: Dict[GraphQLSchema, str]
+
+_cached_queries = {} # type: Dict[str, str]
 
 
 def get_unique_schema_id(schema):
+ # type: (GraphQLSchema) -> str
  """Get a unique id given a GraphQLSchema"""
  assert isinstance(schema, GraphQLSchema), (
  "Must receive a GraphQLSchema as schema. Received {}"
@@ -21,6 +26,7 @@ def get_unique_schema_id(schema):
 
 
 def get_unique_document_id(query_str):
+ # type: (str) -> str
  """Get a unique id given a query_string"""
  assert isinstance(query_str, string_types), (
  "Must receive a string as query_str. Received {}"
@@ -32,7 +38,13 @@ def get_unique_document_id(query_str):
 
 
 class GraphQLCachedBackend(GraphQLBackend):
- def __init__(self, backend, cache_map=None, use_consistent_hash=False):
+ def __init__(
+ self,
+ backend, # type: GraphQLBackend
+ cache_map=None, # type: Optional[Dict[Hashable, GraphQLDocument]]
+ use_consistent_hash=False, # type: bool
+ ):
+ # type: (...) -> None
  assert isinstance(
  backend, GraphQLBackend
  ), "Provided backend must be an instance of GraphQLBackend"
@@ -43,14 +55,16 @@ def __init__(self, backend, cache_map=None, use_consistent_hash=False):
  self.use_consistent_hash = use_consistent_hash
 
  def get_key_for_schema_and_document_string(self, schema, request_string):
+ # type: (GraphQLSchema, str) -> int
  """This method returns a unique key given a schema and a request_string"""
  if self.use_consistent_hash:
  schema_id = get_unique_schema_id(schema)
  document_id = get_unique_document_id(request_string)
- return (schema_id, document_id)
+ return hash((schema_id, document_id))
  return hash((schema, request_string))
 
  def document_from_string(self, schema, request_string):
+ # type: (GraphQLSchema, str) -> Optional[GraphQLDocument]
  """This method returns a GraphQLQuery (from cache if present)"""
  key = self.get_key_for_schema_and_document_string(schema, request_string)
  if key not in self.cache_map:

graphql/backend/compiled.py

@@ -1,39 +1,52 @@
 from six import string_types
 from .base import GraphQLDocument
 
+if False:
+ from ..type.schema import GraphQLSchema
+ from typing import Any, Optional, Dict, Callable, Union
+
 
 class GraphQLCompiledDocument(GraphQLDocument):
  @classmethod
- def from_code(cls, schema, code, uptodate=None, extra_namespace=None):
+ def from_code(
+ cls,
+ schema, # type: GraphQLSchema
+ code, # type: Union[str, Any]
+ uptodate=None, # type: Optional[bool]
+ extra_namespace=None, # type: Optional[Dict[str, Any]]
+ ):
+ # type: (...) -> GraphQLCompiledDocument
  """Creates a GraphQLDocument object from compiled code and the globals. This
  is used by the loaders and schema to create a document object.
  """
  if isinstance(code, string_types):
- filename = '<document>'
- code = compile(code, filename, 'exec')
+ filename = "<document>"
+ code = compile(code, filename, "exec")
  namespace = {"__file__": code.co_filename}
  exec(code, namespace)
  if extra_namespace:
  namespace.update(extra_namespace)
  rv = cls._from_namespace(schema, namespace)
- rv._uptodate = uptodate
+ # rv._uptodate = uptodate
  return rv
 
  @classmethod
  def from_module_dict(cls, schema, module_dict):
+ # type: (GraphQLSchema, Dict[str, Any]) -> GraphQLCompiledDocument
  """Creates a template object from a module. This is used by the
  module loader to create a document object.
  """
  return cls._from_namespace(schema, module_dict)
 
  @classmethod
  def _from_namespace(cls, schema, namespace):
- document_string = namespace.get("document_string", "")
- document_ast = namespace.get("document_ast")
- execute = namespace["execute"]
+ # type: (GraphQLSchema, Dict[str, Any]) -> GraphQLCompiledDocument
+ document_string = namespace.get("document_string", "") # type: str
+ document_ast = namespace.get("document_ast") # type: ignore
+ execute = namespace["execute"] # type: Callable
 
  namespace["schema"] = schema
- return cls(
+ return cls( # type: ignore
  schema=schema,
  document_string=document_string,
  document_ast=document_ast,

graphql/backend/core.py

@@ -8,35 +8,51 @@
 
 from .base import GraphQLBackend, GraphQLDocument
 
-
-def execute_and_validate(schema, document_ast, *args, **kwargs):
- do_validation = kwargs.get('validate', True)
+if False:
+ from typing import Any, Optional, Union
+ from .base import GraphQLDocument
+ from ..language.ast import Document
+ from ..type.schema import GraphQLSchema
+ from ..execution.base import ExecutionResult
+ from rx.core.anonymousobservable import AnonymousObservable
+
+
+def execute_and_validate(
+ schema, # type: GraphQLSchema
+ document_ast, # type: Document
+ *args, # type: Any
+ **kwargs # type: Any
+):
+ # type: (...) -> Union[ExecutionResult, AnonymousObservable]
+ do_validation = kwargs.get("validate", True)
  if do_validation:
  validation_errors = validate(schema, document_ast)
  if validation_errors:
- return ExecutionResult(
- errors=validation_errors,
- invalid=True,
- )
+ return ExecutionResult(errors=validation_errors, invalid=True)
 
  return execute(schema, document_ast, *args, **kwargs)
 
 
 class GraphQLCoreBackend(GraphQLBackend):
- def __init__(self, executor=None, **kwargs):
- super(GraphQLCoreBackend, self).__init__(**kwargs)
+ def __init__(self, executor=None):
+ # type: (Optional[Any]) -> None
  self.execute_params = {"executor": executor}
 
  def document_from_string(self, schema, document_string):
+ # type: (GraphQLSchema, Union[Document, str]) -> GraphQLDocument
  if isinstance(document_string, ast.Document):
  document_ast = document_string
  document_string = print_ast(document_ast)
  else:
- assert isinstance(document_string, string_types), "The query must be a string"
+ assert isinstance(
+ document_string, string_types
+ ), "The query must be a string"
  document_ast = parse(document_string)
  return GraphQLDocument(
  schema=schema,
  document_string=document_string,
  document_ast=document_ast,
- execute=partial(execute_and_validate, schema, document_ast, **self.execute_params),
+ execute=partial(
+ execute_and_validate, schema, document_ast, **self.execute_params
+ ),
  )

graphql/backend/decider.py

@@ -1,8 +1,13 @@
-from .base import GraphQLBackend
+from .base import GraphQLBackend, GraphQLDocument
+
+if False:
+ from typing import List, Union, Any, Optional
+ from ..type.schema import GraphQLSchema
 
 
 class GraphQLDeciderBackend(GraphQLBackend):
- def __init__(self, backends=None):
+ def __init__(self, backends):
+ # type: (List[GraphQLBackend], ) -> None
  if not backends:
  raise Exception("Need to provide backends to decide into.")
  if not isinstance(backends, (list, tuple)):
@@ -11,6 +16,7 @@ def __init__(self, backends=None):
  super(GraphQLDeciderBackend, self).__init__()
 
  def document_from_string(self, schema, request_string):
+ # type: (GraphQLSchema, str) -> GraphQLDocument
  for backend in self.backends:
  try:
  return backend.document_from_string(schema, request_string)

graphql/backend/tests/test_base.py

@@ -3,11 +3,13 @@
 
 
 def test_get_default_backend_returns_core_by_default():
+ # type: () -> None
  backend = get_default_backend()
  assert isinstance(backend, GraphQLCoreBackend)
 
 
 def test_set_default_backend():
+ # type: () -> None
  default_backend = get_default_backend()
  new_backend = GraphQLCoreBackend()
  assert new_backend != default_backend
@@ -16,6 +18,7 @@ def test_set_default_backend():
 
 
 def test_set_default_backend_fails_if_invalid_backend():
+ # type: () -> None
  default_backend = get_default_backend()
  with pytest.raises(Exception) as exc_info:
  set_default_backend(object())

graphql/backend/tests/test_cache.py

@@ -11,13 +11,15 @@
 
 
 def test_cached_backend():
+ # type: () -> None
  cached_backend = GraphQLCachedBackend(GraphQLCoreBackend())
  document1 = cached_backend.document_from_string(schema, "{ hello }")
  document2 = cached_backend.document_from_string(schema, "{ hello }")
  assert document1 == document2
 
 
 def test_cached_backend_with_use_consistent_hash():
+ # type: () -> None
  cached_backend = GraphQLCachedBackend(GraphQLCoreBackend(), use_consistent_hash=True)
  document1 = cached_backend.document_from_string(schema, "{ hello }")
  document2 = cached_backend.document_from_string(schema, "{ hello }")