Merge pull request #1 from PotomacInnovation/update-public

Bring public version up to scratch

Author: necaris

.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]

CONTRIBUTING.md

@@ -33,4 +33,4 @@ this project to learn more.
 
 To sign the CLA, open a pull request as usual. If you haven't signed the CLA
 yet, we cannot merge any pull request until the CLA is signed. You only need to
-sign the CLA once. Please submit a signed CLA via email to [[email protected]](mailto:[email protected]).
+sign the CLA once.

CONTRIBUTOR_LICENSE_AGREEMENT.md

@@ -27,11 +27,3 @@ You accept and agree to the following terms and conditions for Your present and
 7. Should You wish to submit work that is not Your original creation, You may submit it to Upside separately from any Contribution, identifying the complete details of its source and of any license or other restriction (including, but not limited to, related patents, trademarks, and license agreements) of which you are personally aware, and conspicuously marking the work as “Submitted on behalf of a third-party: [named here]”.
 
 8. You agree to notify Upside of any facts or circumstances of which you become aware that would make these representations inaccurate in any respect.
-
-
-
-Please sign: __________________________________                     Date: ________________
-
-
-
-**Please submit a signed copy via email to [email protected].**

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,85 @@ 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.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