Merge pull request #9 from lirsacc/repeatable-directives

Repeatable directives

Author: lirsacc

CHANGES.md

@@ -20,6 +20,12 @@ Unreleased
 - Add `py_gql.exts.scalars.Base64String` scalar type.
 - Add support for schema description (see: [graphql/graphql-spec/pull/466](https://github.com/graphql/graphql-spec/pull/466)).
 - Add `py_gql.exc.UnknownDirective` for cleaner errors in `py_gql.ResolveInfo.get_directive_arguments`.
+- Add support for repeatable directives (see: [graphql/graphql-spec/pull/472](https://github.com/graphql/graphql-spec/pull/472)).
+
+  - Language support (parser, ast, SDL and schema definitions)
+  - Schema directives: repeatable directives applied multiple times when calling `build_schema()` will be called multiple times in order.
+  - `ResolveInfo.get_directive_arguments` has not been modified to not break exising code. It returns the first set of arguments for repeated directives.
+  - `ResolveInfo.get_all_directive_arguments` has been added to handle repeated directives.
 
 [0.6.1](https://github.com/lirsacc/py-gql/releases/tag/0.6.1) - 2020-04-01
 --------------------------------------------------------------------------

src/py_gql/execution/wrappers.py

@@ -22,9 +22,9 @@
     TYPE_NAME_INTROSPECTION_FIELD,
 )
 from ..utilities import (
+    all_directive_arguments,
     coerce_argument_values,
     collect_fields,
-    directive_arguments,
     selected_fields,
 )
 from .runtime import Runtime
@@ -192,6 +192,10 @@ class ResolveInfo:
 
     This is the 3rd positional argument provided to resolver functions and is
     constructed internally during query execution.
+
+    Warning:
+        This class assumes that the document and schema have been validated for
+        execution and may break unexectedly if used outside of such a context.
     """
 
     __slots__ = (
@@ -225,9 +229,7 @@ def __init__(
         self.runtime = runtime
 
         self._context = context
-        self._directive_arguments = (
-            {}
-        )  # type: Dict[str, Optional[Dict[str, Any]]]
+        self._directive_arguments = {}  # type: Dict[str, List[Dict[str, Any]]]
 
     @property
     def schema(self) -> Schema:  # noqa: D401
@@ -254,9 +256,7 @@ def get_directive_arguments(self, name: str) -> Optional[Dict[str, Any]]:
         """
         Extract arguments for a given directive on the current field.
 
-        Warning:
-            This method assumes the document has been validated and the
-            definition exists and is valid at this position.
+        This has the same semantics as `py_gql.utilities.directive_arguments`.
 
         Args:
             name: The name of the directive to extract.
@@ -265,6 +265,21 @@ def get_directive_arguments(self, name: str) -> Optional[Dict[str, Any]]:
             ``None`` if the directive is not present on the current field and a
             dictionary of coerced arguments otherwise.
         """
+        args = self.get_all_directive_arguments(name)
+        return args[0] if args else None
+
+    def get_all_directive_arguments(self, name: str) -> List[Dict[str, Any]]:
+        """
+        Extract arguments for a given directive on the current field.
+
+        This has the same semantics as `py_gql.utilities.all_directive_arguments`.
+
+        Args:
+            name: The name of the directive to extract.
+
+        Returns:
+            List of directive arguments in order of occurrence.
+        """
         try:
             return self._directive_arguments[name]
         except KeyError:
@@ -273,7 +288,7 @@ def get_directive_arguments(self, name: str) -> Optional[Dict[str, Any]]:
             except KeyError:
                 raise UnknownDirective(name) from None
 
-            args = self._directive_arguments[name] = directive_arguments(
+            args = self._directive_arguments[name] = all_directive_arguments(
                 directive_def, self.nodes[0], self._context.variables,
             )
             return args

src/py_gql/lang/ast.py

@@ -944,20 +944,23 @@ class DirectiveDefinition(SupportDescription, TypeSystemDefinition):
         "description",
         "name",
         "arguments",
+        "repeatable",
         "locations",
     )
 
     def __init__(
         self,
         name: Name,
         arguments: Optional[List[InputValueDefinition]] = None,
+        repeatable: bool = False,
         locations: Optional[List[Name]] = None,
         source: Optional[str] = None,
         loc: Optional[Tuple[int, int]] = None,
         description: Optional[StringValue] = None,
     ):
         self.name = name
         self.arguments = arguments or []  # type: List[InputValueDefinition]
+        self.repeatable = repeatable
         self.locations = locations or []  # type: List[Name]
         self.source = source
         self.loc = loc

src/py_gql/lang/parser.py

@@ -394,6 +394,25 @@ def skip(self, kind: Kind) -> bool:
             return True
         return False
 
+    def skip_keyword(self, keyword: str) -> bool:
+        """
+        Conditionally advance the parser asserting over a given keyword.
+
+        Args:
+            keyword (str): Expected keyword
+
+        Returns:
+            ``True`` if the next token was the given keyword and we've advanced
+            the parser, ``False`` otherwise.
+
+        """
+        next_token = self.peek()
+        if next_token.__class__ is Name and next_token.value == keyword:
+            self.advance()
+            return True
+
+        return False
+
     def many(
         self, open_kind: Kind, parse_fn: Callable[[], N], close_kind: Kind
     ) -> List[N]:
@@ -1430,19 +1449,22 @@ def parse_input_object_type_extension(
     def parse_directive_definition(self) -> _ast.DirectiveDefinition:
         """
         DirectiveDefinition :
-            Description? directive @ Name ArgumentsDefinition? on DirectiveLocations
+            Description? directive @ Name ArgumentsDefinition? repeatable?
+            on DirectiveLocations
         """
         start = self.peek()
         desc = self.parse_description()
         self.expect_keyword("directive")
         self.expect(At)
         name = self.parse_name()
         args = self.parse_argument_definitions()
+        repeatable = self.skip_keyword("repeatable")
         self.expect_keyword("on")
         return _ast.DirectiveDefinition(
             description=desc,
             name=name,
             arguments=args,
+            repeatable=repeatable,
             locations=self.parse_directive_locations(),
             loc=self._loc(start),
             source=self._source,

src/py_gql/lang/printer.py

@@ -484,6 +484,7 @@ def print_directive_definition(self, node: _ast.DirectiveDefinition) -> str:
                     "directive @",
                     node.name.value,
                     self.print_argument_definitions(node),
+                    " repeatable" if node.repeatable else "",
                     " on ",
                     _join(map(self, node.locations), " | "),
                 ]

src/py_gql/schema/schema_visitor.py

@@ -178,6 +178,7 @@ def on_directive(self, directive: Directive) -> Optional[Directive]:
                 directive.name,
                 directive.locations,
                 args=updated_args,
+                repeatable=directive.repeatable,
                 description=directive.description,
                 node=directive.node,
             )

src/py_gql/schema/types.py

@@ -1007,30 +1007,42 @@ class Directive:
     Type system creators will usually not create these directly.
 
     Args:
-        name: Directive name
+        name: Directive name.
 
-        locations: Possible locations for that directive
+        locations: Possible locations for that directive.
 
-        args: Argument definitions
+        args: Argument definitions.
 
-        description: Directive description
+        repeatable: Specify that the directive can be applied multiple times to
+            the same target. Repeatable directives are often useful when the
+            same directive should be used with different arguments at a single
+            location, especially in cases where additional information needs to
+            be provided to a type or schema extension via a directive.
+
+        description: Directive description.
 
         node: Source node used when building type from the SDL.
 
     Attributes:
-        name (str): Directive name
+        name (str): Directive name.
 
-        description (Optional[str]): Directive description
+        description (Optional[str]): Directive description.
 
-        locations (List[str]): Possible locations for that directive
+        locations (List[str]): Possible locations for that directive.
 
         arguments (List[py_gql.schema.Argument]): Directive arguments.
 
         argument_map (Dict[str, py_gql.schema.Argument]):
             Directive arguments in indexed form.
 
+        repeatable: Specify that the directive can be applied multiple times to
+            the same target. Repeatable directives are often useful when the
+            same directive should be used with different arguments at a single
+            location, especially in cases where additional information needs to
+            be provided to a type or schema extension via a directive.
+
         node (Optional[py_gql.lang.ast.DirectiveDefinition]):
-            Source node used when building type from the SDL
+            Source node used when building type from the SDL.
     """
 
     ALL_LOCATIONS = DIRECTIVE_LOCATIONS
@@ -1042,6 +1054,7 @@ def __init__(
         name: str,
         locations: Sequence[str],
         args: Optional[List[Argument]] = None,
+        repeatable: bool = False,
         description: Optional[str] = None,
         node: Optional[_ast.DirectiveDefinition] = None,
     ):
@@ -1057,6 +1070,7 @@ def __init__(
         self.name = name
         self.description = description
         self.locations = locations
+        self.repeatable = repeatable
         self.arguments = args if args is not None else []
         self.argument_map = {arg.name: arg for arg in self.arguments}
         self.node = node

src/py_gql/sdl/ast_schema_printer.py

@@ -342,10 +342,11 @@ def print_input_object_type(self, type_: InputObjectType) -> str:
         )
 
     def print_directive_definition(self, directive: Directive) -> str:
-        return "%sdirective @%s%s on %s" % (
+        return "%sdirective @%s%s%son %s" % (
             self.print_description(directive),
             directive.name,
             self.print_arguments(directive.arguments, 0),
+            " repeatable " if directive.repeatable else " ",
             " | ".join(directive.locations),
         )
 

src/py_gql/sdl/ast_type_builder.py

@@ -154,6 +154,7 @@ def build_directive(
         return Directive(
             name=directive_def.name.value,
             description=_desc(directive_def),
+            repeatable=directive_def.repeatable,
             locations=[loc.value for loc in directive_def.locations],
             args=(
                 [self._build_argument(arg) for arg in directive_def.arguments]
@@ -203,6 +204,7 @@ def extend_directive(self, directive: Directive) -> Directive:
         return Directive(
             directive.name,
             description=directive.description,
+            repeatable=directive.repeatable,
             locations=directive.locations,
             args=[self._extend_argument(a) for a in directive.arguments],
             node=directive.node,

src/py_gql/sdl/schema_directives.py

@@ -177,7 +177,7 @@ def _collect_schema_directives(
                     [node],
                 )
 
-            if name in applied:
+            if name in applied and not directive_def.repeatable:
                 raise SDLError('Directive "@%s" already applied' % name, [node])
 
             args = coerce_argument_values(directive_def, node)

src/py_gql/utilities/__init__.py

@@ -8,6 +8,7 @@
 
 from .ast_node_from_value import ast_node_from_value
 from .coerce_value import (
+    all_directive_arguments,
     coerce_argument_values,
     coerce_value,
     coerce_variable_values,
@@ -33,6 +34,7 @@
     "collect_fields_untyped",
     "selected_fields",
     "directive_arguments",
+    "all_directive_arguments",
     "introspection_query",
     "untyped_value_from_ast",
     "value_from_ast",

src/py_gql/utilities/coerce_value.py

@@ -264,13 +264,48 @@ def coerce_argument_values(
     return coerced_values
 
 
+def all_directive_arguments(
+    definition: Directive,
+    node: _ast.SupportDirectives,
+    variables: Optional[Mapping[str, Any]] = None,
+) -> List[Dict[str, Any]]:
+    """
+    Extract all directive arguments given node and a directive definition.
+
+    Args:
+        definition: Directive definition from which to extract arguments
+        node: Parse node
+        variables: Coerced variable values
+
+    Returns:
+        List of coerced directive arguments for all occurrences of the directive.
+        If the directive is not present the list is empty, otherwise this returns
+        one or more (for repeatable directives) dictionnaries of arguments in
+        the order they appear on the node.
+
+    Raises:
+        CoercionError: If any argument fails to coerce, is missing, etc.
+
+    """
+    return [
+        coerce_argument_values(definition, directive, variables)
+        for directive in node.directives
+        if directive.name.value == definition.name
+    ]
+
+
 def directive_arguments(
     definition: Directive,
     node: _ast.SupportDirectives,
     variables: Optional[Mapping[str, Any]] = None,
 ) -> Optional[Dict[str, Any]]:
     """
-    Extract directive argument given node and a directive definition.
+    Extract first directive arguments given node and a directive definition.
+
+    Warning:
+        This extracts at most a single set of arguments which may not be
+        suitable for repeatable directives. In that case `py_gql.utilities.
+        all_directive_arguments`. should be preferred.
 
     Args:
         definition: Directive definition from which to extract arguments
@@ -283,7 +318,6 @@ def directive_arguments(
 
     Raises:
         CoercionError: If any argument fails to coerce, is missing, etc.
-
     """
     directive = find_one(
         node.directives, lambda d: d.name.value == definition.name

src/py_gql/validation/rules/__init__.py

@@ -731,7 +731,7 @@ def _validate_unique_directive_names(self, node):
             name = directive.name.value
             directive_def = self.schema.directives.get(name)
 
-            if directive_def is None:
+            if directive_def is None or directive_def.repeatable:
                 continue
 
             if name in seen:

tests/fixtures/schema-kitchen-sink.graphql

@@ -125,6 +125,10 @@ directive @include2(if: Boolean!) on
   | FRAGMENT_SPREAD
   | INLINE_FRAGMENT
 
+directive @myRepeatableDir(name: String!) repeatable on
+  | OBJECT
+  | INTERFACE
+
 extend schema @onSchema
 
 extend schema @onSchema {

tests/fixtures/schema-kitchen-sink.printed.graphql

@@ -108,6 +108,8 @@ directive @include(if: Boolean!) on FIELD | FRAGMENT_SPREAD | INLINE_FRAGMENT
 
 directive @include2(if: Boolean!) on FIELD | FRAGMENT_SPREAD | INLINE_FRAGMENT
 
+directive @myRepeatableDir(name: String!) repeatable on OBJECT | INTERFACE
+
 extend schema @onSchema
 
 extend schema @onSchema {