Coverage for src/lexigram/graphql/core/validation.py: 94%

1"""GraphQL query validation. 

2 

3This module provides validation utilities for GraphQL queries, 

4delegating depth and alias checks to the canonical security validators 

5in ``lexigram.graphql.security``. 

6""" 

7 

8from __future__ import annotations 

9 

10from dataclasses import dataclass, field 

11from typing import TYPE_CHECKING, Any 

12 

13from lexigram.graphql import constants as const 

14from lexigram.logging import get_logger 

15 

16if TYPE_CHECKING: 

17 from graphql.language.ast import DocumentNode 

18 

19 

20logger = get_logger(__name__) 

21 

22 

23@dataclass 

24class ValidationResult: 

25 """Result of query validation. 

26 

27 Attributes: 

28 is_valid: Whether the query is valid. 

29 errors: List of validation errors. 

30 depth: Maximum query depth. 

31 field_count: Total number of fields. 

32 warnings: Non-fatal warnings. 

33 """ 

34 

35 is_valid: bool = True 

36 errors: list[str] = field(default_factory=list) 

37 depth: int = 0 

38 field_count: int = 0 

39 warnings: list[str] = field(default_factory=list) 

40 

41 def add_error(self, error: str) -> None: 

42 """Add a validation error.""" 

43 self.errors.append(error) 

44 self.is_valid = False 

45 

46 def add_warning(self, warning: str) -> None: 

47 """Add a validation warning.""" 

48 self.warnings.append(warning) 

49 

50 

51class SchemaValidator: 

52 """Validate GraphQL schemas and queries. 

53 

54 Provides comprehensive validation of GraphQL queries against the schema, 

55 delegating depth enforcement to :class:`~lexigram.graphql.security.depth.DepthLimitValidator` 

56 and alias enforcement to :class:`~lexigram.graphql.security.alias.AliasLimitValidator`. 

57 

58 Example: 

59 ```python 

60 validator = SchemaValidator(schema) 

61 result = validator.validate_query(query) 

62 

63 if not result.is_valid: 

64 for error in result.errors: 

65 logger.info("Validation error: %s", error) 

66 ``` 

67 """ 

68 

69 def __init__( 

70 self, 

71 schema: Any, 

72 max_depth: int = 10, 

73 max_aliases: int = 10, 

74 max_complexity: int = const.DEFAULT_MAX_COMPLEXITY, 

75 ) -> None: 

76 """Initialize the validator. 

77 

78 Args: 

79 schema: GraphQL schema. 

80 max_depth: Maximum query depth (delegated to DepthLimitValidator). 

81 max_aliases: Maximum number of aliases (delegated to AliasLimitValidator). 

82 max_complexity: Maximum query complexity (delegated to ComplexityAnalyzer). 

83 """ 

84 from lexigram.graphql.security.alias import AliasLimitValidator 

85 from lexigram.graphql.security.complexity import ComplexityAnalyzer 

86 from lexigram.graphql.security.depth import DepthLimitValidator 

87 

88 self._schema = schema 

89 self._depth_validator = DepthLimitValidator(max_depth=max_depth) 

90 self._alias_validator = AliasLimitValidator(max_aliases=max_aliases) 

91 self._complexity_analyzer = ComplexityAnalyzer( 

92 max_complexity=max_complexity, 

93 ) 

94 

95 def validate_query( 

96 self, 

97 query: str, 

98 variables: dict[str, Any] | None = None, 

99 ) -> ValidationResult: 

100 """Validate a GraphQL query. 

101 

102 Args: 

103 query: GraphQL query string. 

104 variables: Query variables. 

105 

106 Returns: 

107 Validation result. 

108 """ 

109 from graphql import parse 

110 from graphql import validate as gql_validate 

111 

112 result = ValidationResult() 

113 

114 try: 

115 document = parse(query) 

116 

117 # Validate against schema — unwrap Strawberry wrappers if needed 

118 schema = self._schema 

119 if hasattr(schema, "__dict__"): 

120 for attr in ("_schema", "schema", "graphql_schema"): 

121 val = getattr(schema, attr, None) 

122 if val and val.__class__.__name__ == "GraphQLSchema": 

123 schema = val 

124 break 

125 

126 for error in gql_validate(schema, document): 

127 result.add_error(str(error)) 

128 

129 # Depth — delegate to DepthLimitValidator (single source of truth) 

130 depth = self._depth_validator.get_depth(document) 

131 result.depth = depth 

132 if depth > self._depth_validator.max_depth: 

133 result.add_error( 

134 f"Query depth {depth} exceeds maximum allowed depth of " 

135 f"{self._depth_validator.max_depth}", 

136 ) 

137 

138 # Alias count — delegate to AliasLimitValidator (single source of truth) 

139 alias_count = self._alias_validator.count_aliases(document) 

140 if alias_count > self._alias_validator.max_aliases: 

141 result.add_error( 

142 f"Query has {alias_count} aliases, exceeding limit of " 

143 f"{self._alias_validator.max_aliases}", 

144 ) 

145 

146 # Complexity — delegate to ComplexityAnalyzer (single source of truth) 

147 complexity_ok, complexity_result = self._complexity_analyzer.check( 

148 document, schema 

149 ) 

150 if not complexity_ok: 

151 result.add_error(complexity_result.warnings[0]) 

152 

153 # Field count (informational only — no dedicated security validator needed) 

154 result.field_count = self._count_fields(document) 

155 

156 except (ValueError, SyntaxError, TypeError) as e: 

157 result.add_error(f"Query parse error: {e}") 

158 except Exception as e: # noqa: BLE001 — duck-type check for GraphQLSyntaxError which cannot be imported directly 

159 if e.__class__.__name__ == "GraphQLSyntaxError": 

160 result.add_error(f"Query parse error: {e}") 

161 else: 

162 raise 

163 

164 return result 

165 

166 def _count_fields(self, document: DocumentNode) -> int: 

167 """Count the total number of fields in a document. 

168 

169 Args: 

170 document: Parsed GraphQL document. 

171 

172 Returns: 

173 Total field count. 

174 """ 

175 count = 0 

176 

177 def _recurse(selection_set: Any) -> None: 

178 nonlocal count 

179 if selection_set is None: 

180 return 

181 for selection in selection_set.selections: 

182 if hasattr(selection, "name"): 

183 count += 1 

184 if hasattr(selection, "selection_set"): 

185 _recurse(selection.selection_set) 

186 

187 for definition in document.definitions: 

188 if hasattr(definition, "selection_set"): 

189 _recurse(definition.selection_set) 

190 

191 return count 

192 

193 

194def validate_query( 

195 schema: Any, 

196 query: str, 

197 variables: dict[str, Any] | None = None, 

198 max_depth: int = 10, 

199) -> ValidationResult: 

200 """Validate a GraphQL query (convenience function). 

201 

202 Args: 

203 schema: GraphQL schema. 

204 query: GraphQL query string. 

205 variables: Query variables. 

206 max_depth: Maximum allowed depth. 

207 

208 Returns: 

209 Validation result. 

210 """ 

211 validator = SchemaValidator(schema, max_depth=max_depth) 

212 return validator.validate_query(query, variables) 

213 

214 

215__all__ = [ 

216 "SchemaValidator", 

217 "ValidationResult", 

218 "validate_query", 

219]