Merge pull request #9 from strollby/input-transform

feat: add support for input transformation

README.md

@@ -154,6 +154,15 @@ from graphql import (
 from graphene_directives import CustomDirective, DirectiveLocation, Schema, build_schema, directive_decorator
 
 
+def input_transform(inputs: dict, _schema: Schema) -> dict:
+    """
+    def input_transform (inputs: Any, schema: Schema) -> dict,
+    """
+    if inputs.get("max_age") > 200:
+        inputs["swr"] = 30
+    return inputs
+
+
 def validate_non_field_input(_type: Any, inputs: dict, _schema: Schema) -> bool:
     """
     def validator (type_: graphene type, inputs: Any, schema: Schema) -> bool,
@@ -194,6 +203,7 @@ CacheDirective = CustomDirective(
     description="Caching directive to control cache behavior of fields or fragments.",
     non_field_validator=validate_non_field_input,
     field_validator=validate_field_input,
+    input_transform=input_transform,
 )
 
 # This returns a partial of directive function

example/complex_uses.graphql

@@ -30,7 +30,7 @@ directive @repeatable_directive(
   serviceName: String!
 ) repeatable on OBJECT | FIELD_DEFINITION
 
-union SearchResult @cache(maxAge: 500) @authenticated(required: true) = Human | Droid | Starship
+union SearchResult @cache(maxAge: 500, swr: 30) @authenticated(required: true) = Human | Droid | Starship
 
 type Human @cache(maxAge: 60) {
   name: String
@@ -39,7 +39,7 @@ type Human @cache(maxAge: 60) {
 
 type Droid  @cache(maxAge: 200) {
   """Test Description"""
-  name: String @deprecated(reason: "Deprecated use born in") @cache(maxAge: 300)
+  name: String @deprecated(reason: "Deprecated use born in") @cache(maxAge: 300, swr: 30)
   primaryFunction: String
 }
 
@@ -61,15 +61,15 @@ type Admin @internal @key {
 input HumanInput  @cache(maxAge: 60) @authenticated(required: true) {
   bornIn: String
   """Test Description"""
-  name: String @deprecated(reason: "Deprecated use born in") @cache(maxAge: 300)
+  name: String @deprecated(reason: "Deprecated use born in") @cache(maxAge: 300, swr: 30)
 }
 
 enum TruthEnum  @cache(maxAge: 100) @authenticated(required: true) {
   A @authenticated(required: true)
   B
 }
 
-scalar DateNewScalar @cache(maxAge: 500) @authenticated(required: true)
+scalar DateNewScalar @cache(maxAge: 500, swr: 30) @authenticated(required: true)
 
 type User  @authenticated(required: true) {
   name: String

example/complex_uses.py

@@ -22,6 +22,15 @@
 curr_dir = os.path.dirname(os.path.realpath(__file__))
 
 
+def input_transform(inputs: dict, _schema: Schema) -> dict:
+    """
+    def input_transform (inputs: Any, schema: Schema) -> dict,
+    """
+    if inputs.get("max_age") > 200:
+        inputs["swr"] = 30
+    return inputs
+
+
 def validate_non_field_input(_type: Any, inputs: dict, _schema: Schema) -> bool:
     """
     def validator (type_: graphene type, inputs: Any, schema: Schema) -> bool,
@@ -71,6 +80,7 @@ def validator (parent_type_: graphene_type, field_type_: graphene type, inputs:
     description="Caching directive to control cache behavior of fields or fragments.",
     non_field_validator=validate_non_field_input,
     field_validator=validate_field_input,
+    input_transform=input_transform,
 )
 
 

graphene_directives/data_models/custom_directive_meta.py

@@ -13,6 +13,9 @@ class CustomDirectiveMeta:
     non_field_types: set[GrapheneDirectiveLocation]
     supports_field_types: bool
     supports_non_field_types: bool
+    input_transform: Union[
+        Callable[[dict[str, Any], Any], dict[str, Any]], None
+    ]  # (args, schema) -> args
     non_field_validator: Union[
         Callable[[Any, dict[str, Any], Any], bool], None
     ]  # (type, args, schema) -> valid

graphene_directives/directive.py

@@ -29,6 +29,7 @@ def CustomDirective(  # noqa
     add_definition_to_schema: bool = True,
     non_field_validator: Callable[[Any, dict[str, Any], Any], bool] = None,
     field_validator: Callable[[Any, Any, dict[str, Any], Any], bool] = None,
+    input_transform: Callable[[dict[str, Any], Any], dict[str, Any]] = None,
 ) -> GraphQLDirective:
     """
     Creates a GraphQLDirective
@@ -43,13 +44,15 @@ def CustomDirective(  # noqa
     :param locations: list[DirectiveLocation], if need to use unsupported locations, set allow_all_directive_locations True
     :param allow_all_directive_locations: Allow other DirectiveLocation other than the ones supported by library
     :param add_definition_to_schema: If false, the @directive definition is not added to the graphql schema
-    :param non_field_validator: a validator function
+    :param non_field_validator: a non field validator function
                 def validator (type_: graphene type, inputs: Any, schema: Schema) -> bool,
                     if validator returns False, library raises DirectiveCustomValidationError
-    :param field_validator: a validator function
+    :param field_validator: a field validator function
                 def validator (parent_type_: graphene_type, field_type_: graphene type, inputs: Any, schema: Schema) -> bool,
                     if validator returns False, library raises DirectiveCustomValidationError
 
+    :param input_transform: a function to transform the input arg's values before usage
+                def input_transform (inputs: dict[str, Any], schema: Schema) -> dict[str, Any]
 
     """
 
@@ -115,6 +118,7 @@ def validator (parent_type_: graphene_type, field_type_: graphene type, inputs:
         supports_non_field_types=supports_non_field_types,
         non_field_validator=non_field_validator,
         field_validator=field_validator,
+        input_transform=input_transform,
     )
 
     # Check if target_directive.locations have accepted types

graphene_directives/parsers.py

@@ -150,3 +150,11 @@ def on_input_value_error(
         raise DirectiveInvalidArgValueTypeError(errors=errors)
 
     return coerced_values
+
+
+def arg_camel_case(inputs: dict) -> dict:
+    return {to_camel_case(k): v for k, v in inputs.items()}
+
+
+def arg_snake_case(inputs: dict) -> dict:
+    return {to_snake_case(k): v for k, v in inputs.items()}

graphene_directives/schema.py

@@ -5,7 +5,7 @@
 from graphene import Schema as GrapheneSchema
 from graphene.types.scalars import ScalarOptions
 from graphene.types.union import UnionOptions
-from graphene.utils.str_converters import to_camel_case, to_snake_case
+from graphene.utils.str_converters import to_camel_case
 from graphql import (
     DirectiveLocation,
     GraphQLArgument,
@@ -32,6 +32,8 @@
 from .directive import CustomDirectiveMeta
 from .exceptions import DirectiveCustomValidationError, DirectiveValidationError
 from .parsers import (
+    arg_camel_case,
+    arg_snake_case,
     decorator_string,
     entity_type_to_fields_string,
     enum_type_to_fields_string,
@@ -125,6 +127,10 @@ def add_argument_decorators(
             for directive in self.directives:
                 if has_field_attribute(arg, directive):
                     directive_values = get_field_attribute_value(arg, directive)
+                    meta_data: CustomDirectiveMeta = getattr(
+                        directive, "_graphene_directive"
+                    )
+
                     if required_directive_field_types in set(directive.locations):
                         raise DirectiveValidationError(
                             ", ".join(
@@ -135,7 +141,15 @@ def add_argument_decorators(
                                 ]
                             )
                         )
+
                     for directive_value in directive_values:
+                        if meta_data.input_transform is not None:
+                            directive_value = arg_camel_case(
+                                meta_data.input_transform(
+                                    arg_snake_case(directive_value), self
+                                )
+                            )
+
                         directive_str = decorator_string(directive, **directive_value)
                         directives.append(directive_str)
 
@@ -251,7 +265,6 @@ def add_field_decorators(self, graphene_types: set, string_schema: str) -> str:
                     meta_data: CustomDirectiveMeta = getattr(
                         directive, "_graphene_directive"
                     )
-                    field_validator = meta_data.field_validator
 
                     if required_directive_field_types in set(directive.locations):
                         raise DirectiveValidationError(
@@ -264,11 +277,14 @@ def add_field_decorators(self, graphene_types: set, string_schema: str) -> str:
                             )
                         )
                     for directive_value in directive_values:
-                        if field_validator is not None and not field_validator(
-                            entity_type,
-                            field,
-                            {to_snake_case(k): v for k, v in directive_value.items()},
-                            self,
+                        if (
+                            meta_data.field_validator is not None
+                            and not meta_data.field_validator(
+                                entity_type,
+                                field,
+                                arg_snake_case(directive_value),
+                                self,
+                            )
                         ):
                             raise DirectiveCustomValidationError(
                                 ", ".join(
@@ -279,6 +295,13 @@ def add_field_decorators(self, graphene_types: set, string_schema: str) -> str:
                                 )
                             )
 
+                        if meta_data.input_transform is not None:
+                            directive_value = arg_camel_case(
+                                meta_data.input_transform(
+                                    arg_snake_case(directive_value), self
+                                )
+                            )
+
                         str_field += (
                             f" {decorator_string(directive, **directive_value)}"
                         )
@@ -347,12 +370,7 @@ def add_non_field_decorators(
                         if (
                             meta_data.non_field_validator is not None
                             and not meta_data.non_field_validator(
-                                non_field,
-                                {
-                                    to_snake_case(k): v
-                                    for k, v in directive_value.items()
-                                },
-                                self,
+                                non_field, arg_snake_case(directive_value), self
                             )
                         ):
                             raise DirectiveCustomValidationError(
@@ -363,6 +381,13 @@ def add_non_field_decorators(
                                     ]
                                 )
                             )
+                        if meta_data.input_transform is not None:
+                            directive_value = arg_camel_case(
+                                meta_data.input_transform(
+                                    arg_snake_case(directive_value), self
+                                )
+                            )
+
                         directive_annotations.append(
                             f"{decorator_string(directive, **directive_value)}"
                         )

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "graphene-directives"
-version = "0.4.1"
+version = "0.4.2"
 packages = [{include = "graphene_directives"}]
 description = "Schema Directives implementation for graphene"
 authors = ["Strollby <[email protected]>"]

tests/schema_files/test_directive_input_transform.graphql

@@ -0,0 +1,37 @@
+"""Caching directive to control cache behavior of fields or fragments."""
+directive @cache(
+  """Specifies the maximum age for cache in seconds."""
+  maxAge: Int!
+
+  """Stale-while-revalidate value in seconds. Optional."""
+  swr: Int
+
+  """Scope of the cache. Optional."""
+  scope: String
+) on FIELD_DEFINITION | OBJECT | UNION
+
+union SearchResult @cache(maxAge: 500, swr: 30) = Human | Droid | Starship
+
+type Human @cache(maxAge: 60) {
+  name: String
+  bornIn: String
+}
+
+type Droid  @cache(maxAge: 200) {
+  name: String @cache(maxAge: 300, swr: 30)
+  primaryFunction: String
+}
+
+type Starship  @cache(maxAge: 200) {
+  name: String
+  length: Int @deprecated(reason: "Koo") @cache(maxAge: 60)
+}
+
+type Query {
+  position: Position @deprecated(reason: "Koo")
+}
+
+type Position  @cache(maxAge: 500, swr: 30) {
+  x: Int!
+  y: Int! @cache(maxAge: 60)
+}

tests/test_directive_input_transform.py

@@ -0,0 +1,92 @@
+from pathlib import Path
+
+import graphene
+from graphql import GraphQLArgument, GraphQLInt, GraphQLNonNull, GraphQLString
+
+from graphene_directives import (
+    CustomDirective,
+    DirectiveLocation,
+    Schema,
+    build_schema,
+    directive_decorator,
+)
+
+curr_dir = Path(__file__).parent
+
+
+def input_transform(inputs: dict, _schema: Schema) -> dict:
+    """
+    def input_transform (inputs: Any, schema: Schema) -> dict,
+    """
+    if inputs.get("max_age") > 200:
+        inputs["swr"] = 30
+    return inputs
+
+
+CacheDirective = CustomDirective(
+    name="cache",
+    locations=[
+        DirectiveLocation.FIELD_DEFINITION,
+        DirectiveLocation.OBJECT,
+        DirectiveLocation.UNION,
+    ],
+    args={
+        "max_age": GraphQLArgument(
+            GraphQLNonNull(GraphQLInt),
+            description="Specifies the maximum age for cache in seconds.",
+        ),
+        "swr": GraphQLArgument(
+            GraphQLInt, description="Stale-while-revalidate value in seconds. Optional."
+        ),
+        "scope": GraphQLArgument(
+            GraphQLString, description="Scope of the cache. Optional."
+        ),
+    },
+    description="Caching directive to control cache behavior of fields or fragments.",
+    input_transform=input_transform,
+)
+
+
+cache = directive_decorator(target_directive=CacheDirective)
+
+
+@cache(max_age=500)
+class Position(graphene.ObjectType):
+    x = graphene.Int(required=True)
+    y = cache(field=graphene.Int(required=True), max_age=60)
+
+
+@cache(max_age=60)
+class Human(graphene.ObjectType):
+    name = graphene.String()
+    born_in = graphene.String()
+
+
+@cache(max_age=200)
+class Droid(graphene.ObjectType):
+    name = cache(field=graphene.String(), max_age=300)
+    primary_function = graphene.String()
+
+
+@cache(max_age=200)
+class Starship(graphene.ObjectType):
+    name = graphene.String()
+    length = cache(field=graphene.Int(deprecation_reason="Koo"), max_age=60)
+
+
+@cache(max_age=500)
+class SearchResult(graphene.Union):
+    class Meta:
+        types = (Human, Droid, Starship)
+
+
+class Query(graphene.ObjectType):
+    position = graphene.Field(Position, deprecation_reason="Koo")
+
+
+schema = build_schema(query=Query, types=(SearchResult,), directives=[CacheDirective])
+
+
+def test_generate_schema() -> None:
+    with open(f"{curr_dir}/schema_files/test_directive_input_transform.graphql") as f:
+        assert str(schema) == f.read()