|
| 1 | +from typing import Dict, Optional, Union |
| 2 | + |
| 3 | +from ...error import GraphQLError |
| 4 | +from ...language import ( |
| 5 | + OperationTypeDefinitionNode, |
| 6 | + OperationType, |
| 7 | + SchemaDefinitionNode, |
| 8 | + SchemaExtensionNode, |
| 9 | +) |
| 10 | +from ...type import GraphQLObjectType |
| 11 | +from . import SDLValidationContext, SDLValidationRule |
| 12 | + |
| 13 | +__all__ = [ |
| 14 | + "UniqueOperationTypesRule", |
| 15 | + "duplicate_operation_type_message", |
| 16 | + "existed_operation_type_message", |
| 17 | +] |
| 18 | + |
| 19 | + |
| 20 | +def duplicate_operation_type_message(operation: str) -> str: |
| 21 | + return f"There can be only one '{operation}' type in schema." |
| 22 | + |
| 23 | + |
| 24 | +def existed_operation_type_message(operation: str) -> str: |
| 25 | + return ( |
| 26 | + f"Type for '{operation}' already defined in the schema." |
| 27 | + " It cannot be redefined." |
| 28 | + ) |
| 29 | + |
| 30 | + |
| 31 | +class UniqueOperationTypesRule(SDLValidationRule): |
| 32 | + """Unique operation types |
| 33 | +
|
| 34 | + A GraphQL document is only valid if it has only one type per operation. |
| 35 | + """ |
| 36 | + |
| 37 | + def __init__(self, context: SDLValidationContext) -> None: |
| 38 | + super().__init__(context) |
| 39 | + schema = context.schema |
| 40 | + self.defined_operation_types: Dict[ |
| 41 | + OperationType, OperationTypeDefinitionNode |
| 42 | + ] = {} |
| 43 | + self.existing_operation_types: Dict[ |
| 44 | + OperationType, Optional[GraphQLObjectType] |
| 45 | + ] = ( |
| 46 | + { |
| 47 | + OperationType.QUERY: schema.query_type, |
| 48 | + OperationType.MUTATION: schema.mutation_type, |
| 49 | + OperationType.SUBSCRIPTION: schema.subscription_type, |
| 50 | + } |
| 51 | + if schema |
| 52 | + else {} |
| 53 | + ) |
| 54 | + self.schema = schema |
| 55 | + |
| 56 | + def check_operation_types( |
| 57 | + self, node: Union[SchemaDefinitionNode, SchemaExtensionNode], *_args |
| 58 | + ): |
| 59 | + for operation_type in node.operation_types or []: |
| 60 | + operation = operation_type.operation |
| 61 | + already_defined_operation_type = self.defined_operation_types.get(operation) |
| 62 | + |
| 63 | + if self.existing_operation_types.get(operation): |
| 64 | + self.report_error( |
| 65 | + GraphQLError( |
| 66 | + existed_operation_type_message(operation.value), operation_type |
| 67 | + ) |
| 68 | + ) |
| 69 | + elif already_defined_operation_type: |
| 70 | + self.report_error( |
| 71 | + GraphQLError( |
| 72 | + duplicate_operation_type_message(operation.value), |
| 73 | + [already_defined_operation_type, operation_type], |
| 74 | + ) |
| 75 | + ) |
| 76 | + else: |
| 77 | + self.defined_operation_types[operation] = operation_type |
| 78 | + return self.SKIP |
| 79 | + |
| 80 | + enter_schema_definition = enter_schema_extension = check_operation_types |
0 commit comments