Bring test coverage up to scratch, and clean up other things in the meantime

Author: Rami Chowdhury

.pre-commit-config.yaml

@@ -14,7 +14,7 @@ repos:
       exclude: README.md
     - id: flake8
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.701
+    rev: v0.720
     hooks:
       - id: mypy
         args: [--ignore-missing-imports, --no-strict-optional]

examples/departments.py

@@ -113,6 +113,7 @@ def resolve_list_departments(self, info):
               id
               name
               hiredOn
+              salary { rating }
             }
             ...on Manager {
               name

graphene_pydantic/converters.py

@@ -1,38 +1,30 @@
-from collections import OrderedDict
-from collections.abc import Mapping, Sequence
+import collections
+from collections import abc
 import typing as T
 import uuid
 import datetime
 import decimal
 import enum
-import inspect
-
-from graphene import (
-    Field,
-    Boolean,
-    Dynamic,
-    Enum,
-    Float,
-    Int,
-    List,
-    String,
-    UUID,
-    Union,
-)
+
+from graphene import Field, Boolean, Enum, Float, Int, List, String, UUID, Union
 from graphene.types.base import BaseType
 
 try:
     from graphene.types.decimal import Decimal as GrapheneDecimal
 
     DECIMAL_SUPPORTED = True
-except ImportError:
+except ImportError:  # pragma: no cover
     # graphene 2.1.5+ is required for Decimals
     DECIMAL_SUPPORTED = False
 
 from graphene.types.datetime import Date, Time, DateTime
 from pydantic import fields
 
 from .registry import Registry
+from .util import construct_union_class_name
+
+
+NONE_TYPE = None.__class__  # need to do this because mypy complains about type(None)
 
 
 class ConversionError(TypeError):
@@ -73,11 +65,36 @@ def convert_pydantic_field(
     return Field(resolver=get_attr_resolver(field.name), **field_kwargs)
 
 
-def to_graphene_type(
+def convert_pydantic_type(
     type_: T.Type, field: fields.Field, registry: Registry = None
 ) -> BaseType:  # noqa: C901
     """
-    Map a native Python type to a Graphene-supported Field type, where possible.
+    Convert a Pydantic type to a Graphene Field type, including not just the
+    native Python type but any additional metadata (e.g. shape) that Pydantic
+    knows about.
+    """
+    graphene_type = find_graphene_type(type_, field, registry)
+    if field.shape == fields.Shape.SINGLETON:
+        return graphene_type
+    elif field.shape in (
+        fields.Shape.LIST,
+        fields.Shape.TUPLE,
+        fields.Shape.TUPLE_ELLIPS,
+        fields.Shape.SEQUENCE,
+        fields.Shape.SET,
+    ):
+        # TODO: _should_ Sets remain here?
+        return List(graphene_type)
+    elif field.shape == fields.Shape.MAPPING:
+        raise ConversionError(f"Don't know how to handle mappings in Graphene.")
+
+
+def find_graphene_type(
+    type_: T.Type, field: fields.Field, registry: Registry = None
+) -> BaseType:  # noqa: C901
+    """
+    Map a native Python type to a Graphene-supported Field type, where possible,
+    throwing an error if we don't know what to map it to.
     """
     if type_ == uuid.UUID:
         return UUID
@@ -97,98 +114,89 @@ def to_graphene_type(
         return GrapheneDecimal if DECIMAL_SUPPORTED else Float
     elif type_ == int:
         return Int
+    # NOTE: this has to come before any `issubclass()` checks, because annotated
+    # generic types aren't valid arguments to `issubclass`
+    elif hasattr(type_, "__origin__"):
+        return convert_generic_python_type(type_, field, registry)
     elif type_ in (tuple, list, set):
         # TODO: do Sets really belong here?
         return List
-    elif hasattr(type_, "__origin__"):
-        return convert_generic_type(type_, field, registry)
     elif issubclass(type_, enum.Enum):
         return Enum.from_enum(type_)
     elif registry and registry.get_type_for_model(type_):
         return registry.get_type_for_model(type_)
-    elif inspect.isfunction(type_):
-        # TODO: this may result in false positives?
-        return Dynamic(type_)
     else:
-        raise Exception(
+        raise ConversionError(
             f"Don't know how to convert the Pydantic field {field!r} ({field.type_})"
         )
 
 
-def convert_pydantic_type(
+def convert_generic_python_type(
     type_: T.Type, field: fields.Field, registry: Registry = None
 ) -> BaseType:  # noqa: C901
-    """
-    Convert a Pydantic type to a Graphene Field type, including not just the
-    native Python type but any additional metadata (e.g. shape) that Pydantic
-    knows about.
-    """
-    graphene_type = to_graphene_type(type_, field, registry)
-    if field.shape == fields.Shape.SINGLETON:
-        return graphene_type
-    elif field.shape in (
-        fields.Shape.LIST,
-        fields.Shape.TUPLE,
-        fields.Shape.SEQUENCE,
-        fields.Shape.SET,
-    ):
-        # TODO: _should_ Sets remain here?
-        return List(graphene_type)
-    elif field.shape == fields.Shape.MAPPING:
-        raise ConversionError(f"Don't know how to handle mappings in Graphene.")
-
-
-def convert_generic_type(type_, field, registry=None):
     """
     Convert annotated Python generic types into the most appropriate Graphene
     Field type -- e.g. turn `typing.Union` into a Graphene Union.
     """
     origin = type_.__origin__
-    if not origin:
+    if not origin:  # pragma: no cover  # this really should be impossible
         raise ConversionError(f"Don't know how to convert type {type_!r} ({field})")
+
     # NOTE: This is a little clumsy, but working with generic types is; it's hard to
     # decide whether the origin type is a subtype of, say, T.Iterable since typical
     # Python functions like `isinstance()` don't work
     if origin == T.Union:
         return convert_union_type(type_, field, registry)
-    elif origin in (T.Dict, T.OrderedDict, T.Mapping, dict, OrderedDict) or issubclass(
-        origin, Mapping
-    ):
-        raise ConversionError("Don't know how to handle mappings in Graphene")
-    elif origin in (T.List, T.Set, T.Collection, T.Iterable, list, set) or issubclass(
-        origin, Sequence
-    ):
-        wrapped_types = getattr(type_, "__args__", [])
-        if not wrapped_types:
+    elif origin in (
+        T.Tuple,
+        T.List,
+        T.Set,
+        T.Collection,
+        T.Iterable,
+        list,
+        set,
+    ) or issubclass(origin, abc.Sequence):
+        # TODO: find a better way of divining that the origin is sequence-like
+        inner_types = getattr(type_, "__args__", [])
+        if not inner_types:  # pragma: no cover  # this really should be impossible
             raise ConversionError(
                 f"Don't know how to handle {type_} (generic: {origin})"
             )
-        return List(to_graphene_type(wrapped_types[0], field, registry))
+        # Of course, we can only return a homogeneous type here, so we pick the
+        # first of the wrapped types
+        inner_type = inner_types[0]
+        return List(find_graphene_type(inner_type, field, registry))
+    elif origin in (
+        T.Dict,
+        T.OrderedDict,
+        T.Mapping,
+        collections.OrderedDict,
+        dict,
+    ) or issubclass(origin, abc.Mapping):
+        raise ConversionError("Don't know how to handle mappings in Graphene")
     else:
         raise ConversionError(f"Don't know how to handle {type_} (generic: {origin})")
 
 
-def convert_union_type(type_, field, registry=None):
+def convert_union_type(type_: T.Type, field: fields.Field, registry: Registry = None):
     """
     Convert an annotated Python Union type into a Graphene Union.
     """
-    wrapped_types = type_.__args__
-    # NOTE: a typing.Optional decomposes to a Union[None, T], so we can return
-    # the Graphene type for T; Pydantic will have already parsed it as optional
-    if len(wrapped_types) == 2 and type(None) in wrapped_types:
-        native_type = next(x for x in wrapped_types if x != type(None))  # noqa: E721
-        graphene_type = to_graphene_type(native_type, field, registry)
+    inner_types = type_.__args__
+    if len(inner_types) == 2 and NONE_TYPE in inner_types:
+        # This is effectively a typing.Optional[T], which decomposes into a
+        # typing.Union[None, T] -- we can return the Graphene type for T directly
+        # since Pydantic will have already parsed it as optional
+        native_type = next(x for x in inner_types if x != NONE_TYPE)  # noqa: E721
+        graphene_type = find_graphene_type(native_type, field, registry)
         return graphene_type
-    else:
-        # Otherwise, we use a little metaprogramming -- create our own unique
-        # subclass of graphene.Union that knows its constituent Graphene types
-        graphene_types = tuple(
-            to_graphene_type(x, field, registry) for x in wrapped_types
-        )
-        internal_meta = type("Meta", (), {"types": graphene_types})
 
-        union_class_name = "".join(x.__name__ for x in wrapped_types)
-        union_class = type(
-            f"Union_{union_class_name}", (Union,), {"Meta": internal_meta}
-        )
-        return union_class
+    # Otherwise, we use a little metaprogramming -- create our own unique
+    # subclass of graphene.Union that knows its constituent Graphene types
+    parent_types = tuple(find_graphene_type(x, field, registry) for x in inner_types)
+    internal_meta_cls = type("Meta", (), {"types": parent_types})
+
+    union_cls = type(
+        construct_union_class_name(inner_types), (Union,), {"Meta": internal_meta_cls}
+    )
+    return union_cls

graphene_pydantic/registry.py

@@ -1,81 +1,67 @@
+import typing as T
 from collections import defaultdict
-from enum import Enum as PyEnum
 
-from graphene import Enum
+from pydantic import BaseModel
+from pydantic.fields import Field
+
+if T.TYPE_CHECKING:  # pragma: no cover
+    from .types import PydanticObjectType
 
 
-def assert_is_pydantic_object_type(obj_type):
+def assert_is_pydantic_object_type(obj_type: T.Type["PydanticObjectType"]):
+    """An object in this registry must be a PydanticObjectType."""
     from .types import PydanticObjectType
 
     if not isinstance(obj_type, type) or not issubclass(obj_type, PydanticObjectType):
         raise TypeError(f"Expected PydanticObjectType, but got: {obj_type!r}")
 
 
-class Registry(object):
+class Registry:
+    """Hold information about Pydantic models and how they (and their fields) map to Graphene types."""
+
     def __init__(self):
         self._registry = {}
         self._registry_models = {}
-        self._registry_orm_fields = defaultdict(dict)
-        self._registry_composites = {}
-        self._registry_enums = {}
-        self._registry_sort_enums = {}
+        self._registry_object_fields = defaultdict(dict)
 
-    def register(self, obj_type):
+    def register(self, obj_type: T.Type["PydanticObjectType"]):
         assert_is_pydantic_object_type(obj_type)
 
         assert (
             obj_type._meta.registry == self
         ), "Can't register models linked to another Registry"
         self._registry[obj_type._meta.model] = obj_type
 
-    def get_type_for_model(self, model):
+    def get_type_for_model(self, model: T.Type[BaseModel]) -> "PydanticObjectType":
         return self._registry.get(model)
 
-    def register_orm_field(self, obj_type, field_name, orm_field):
+    def register_object_field(
+        self, obj_type: T.Type["PydanticObjectType"], field_name: str, obj_field: Field
+    ):
         assert_is_pydantic_object_type(obj_type)
 
-        if not field_name or not isinstance(field_name, str):
+        if not field_name or not isinstance(field_name, str):  # pragma: no cover
             raise TypeError(f"Expected a field name, but got: {field_name!r}")
-        self._registry_orm_fields[obj_type][field_name] = orm_field
-
-    def get_orm_field_for_graphene_field(self, obj_type, field_name):
-        return self._registry_orm_fields.get(obj_type, {}).get(field_name)
-
-    def register_composite_converter(self, composite, converter):
-        self._registry_composites[composite] = converter
-
-    def get_converter_for_composite(self, composite):
-        return self._registry_composites.get(composite)
-
-    def register_enum(self, py_enum, graphene_enum):
-        if not isinstance(py_enum, PyEnum):
-            raise TypeError(f"Expected Python Enum, but got: {py_enum!r}")
-        if not isinstance(graphene_enum, type(Enum)):
-            raise TypeError(f"Expected Graphene Enum, but got: {graphene_enum!r}")
-
-        self._registry_enums[py_enum] = graphene_enum
-
-    def register_sort_enum(self, obj_type, sort_enum):
-        assert_is_pydantic_object_type(obj_type)
-
-        if not isinstance(sort_enum, type(Enum)):
-            raise TypeError(f"Expected Graphene Enum, but got: {sort_enum!r}")
-        self._registry_sort_enums[obj_type] = sort_enum
+        self._registry_object_fields[obj_type][field_name] = obj_field
 
-    def get_sort_enum_for_object_type(self, obj_type):
-        return self._registry_sort_enums.get(obj_type)
+    def get_object_field_for_graphene_field(
+        self, obj_type: "PydanticObjectType", field_name: str
+    ) -> Field:
+        return self._registry_object_fields.get(obj_type, {}).get(field_name)
 
 
-registry = None
+registry: T.Optional[Registry] = None
 
 
-def get_global_registry():
+def get_global_registry() -> Registry:
+    """Return a global instance of Registry for common use."""
     global registry
     if not registry:
         registry = Registry()
     return registry
 
 
 def reset_global_registry():
+    """Clear the global instance of the registry."""
     global registry
     registry = None