Coverage for src/lexigram/graphql/controllers/graphql.py: 96%

1"""GraphQL Controller for lexigram-web integration. 

2 

3This module provides a Controller-based implementation for GraphQL endpoints, 

4following the lexigram-contracts.web.Controller pattern. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from starlette.responses import JSONResponse 

12 

13from lexigram.contracts.web import post 

14from lexigram.contracts.web.controller import ControllerProtocol 

15from lexigram.graphql import constants as const 

16from lexigram.graphql.exceptions import AuthenticationError 

17from lexigram.logging import get_logger 

18 

19if TYPE_CHECKING: 

20 # GraphQLExecutorProtocol imported by providers/tests; not needed here now 

21 from lexigram.contracts.core.di import ContainerProtocol 

22 from lexigram.contracts.graphql import GraphQLRequestProtocol 

23 from lexigram.graphql.di.provider import GraphQLProvider 

24 

25logger = get_logger(__name__) 

26 

27 

28class GraphQLController(ControllerProtocol): 

29 """GraphQL HTTP endpoint controller. 

30 

31 This controller exposes GraphQL queries and mutations via HTTP POST 

32 at the configured path (default: ``/graphql``). 

33 

34 Example: 

35 Registration is handled automatically by the GraphQL provider. 

36 To mount the controller manually, resolve the web application via the 

37 ``HTTPApplicationProtocol`` from contracts — do not import from 

38 ``lexigram.web`` directly:: 

39 

40 from lexigram.contracts.web.protocols import HTTPApplicationProtocol 

41 # app = await container.resolve(HTTPApplicationProtocol) 

42 # app.mount("/graphql", graphql_view) 

43 """ 

44 

45 container: ContainerProtocol | None = None 

46 

47 def __init__(self, provider: GraphQLProvider | None = None) -> None: 

48 """Initialize the GraphQL controller. 

49 

50 Args: 

51 provider: GraphQL provider instance. If None, will be resolved 

52 from the DI container at request time. 

53 """ 

54 self._provider = provider 

55 

56 @classmethod 

57 def collect_routes(cls) -> list[dict[str, Any]]: 

58 """Collect routes from controller methods.""" 

59 routes = [] 

60 seen_handlers = set() 

61 

62 for klass in cls.__mro__: 

63 if klass is object: 

64 continue 

65 

66 for attr_name in dir(klass): 

67 if attr_name.startswith("_") or attr_name in seen_handlers: 

68 continue 

69 

70 attr_value = getattr(klass, attr_name, None) 

71 if attr_value is not None and hasattr(attr_value, "_route_config"): 

72 route_config = attr_value._route_config 

73 routes.append( 

74 { 

75 "method": route_config["method"], 

76 "path": route_config["path"], 

77 "handler_name": attr_name, 

78 "response_model": route_config.get("response_model"), 

79 "request_model": route_config.get("request_model"), 

80 "status_code": route_config.get("status_code", 200), 

81 "summary": route_config.get("summary"), 

82 "description": route_config.get("description"), 

83 "tags": route_config.get("tags"), 

84 "operation_id": route_config.get("operation_id"), 

85 "responses": route_config.get("responses"), 

86 "deprecated": route_config.get("deprecated", False), 

87 } 

88 ) 

89 seen_handlers.add(attr_name) 

90 

91 return routes 

92 

93 @post(const.DEFAULT_GRAPHQL_PATH) 

94 async def execute(self, request: GraphQLRequestProtocol) -> JSONResponse: 

95 """Execute a GraphQL query. 

96 

97 Args: 

98 request: HTTP request containing the GraphQL query. 

99 

100 Returns: 

101 JSON response with query results or errors. 

102 """ 

103 provider = self._get_provider(request) 

104 executor = provider.executor() if provider else None 

105 

106 if not provider or not executor: 

107 return JSONResponse( 

108 {"errors": [{"message": "GraphQL subsystem not configured"}]}, 

109 status_code=503, 

110 ) 

111 

112 try: 

113 body = await request.json() 

114 except (ValueError, UnicodeDecodeError): 

115 return JSONResponse( 

116 {"errors": [{"message": "Invalid JSON body"}]}, 

117 status_code=400, 

118 ) 

119 

120 query = body.get("query") 

121 if not query: 

122 return JSONResponse( 

123 {"errors": [{"message": "No query provided"}]}, 

124 status_code=400, 

125 ) 

126 

127 try: 

128 # Extract user from request state or scope extensions 

129 user = getattr(request.state, "user", None) 

130 if user is None: 

131 user = request.scope.get("extensions", {}).get("user") 

132 

133 context = ( 

134 await provider.context_factory.create_context( 

135 raw_request=request, 

136 user=user, 

137 metadata={"container": getattr(self, "container", None)}, 

138 ) 

139 if provider.context_factory is not None 

140 else None 

141 ) 

142 

143 result = await executor.execute( 

144 query, 

145 variables=body.get("variables"), 

146 operation_name=body.get("operationName"), 

147 context=context, 

148 ) 

149 

150 if result.is_err(): 

151 error = result.unwrap_err() 

152 logger.error("GraphQL setup error: %s", error) 

153 return JSONResponse( 

154 {"errors": [{"message": f"Transport error: {error!s}"}]}, 

155 status_code=500, 

156 ) 

157 

158 gql_response = result.unwrap() 

159 response_data: dict[str, Any] = {"data": gql_response.data} 

160 if gql_response.errors: 

161 response_data["errors"] = [ 

162 {"message": str(e)} for e in gql_response.errors 

163 ] 

164 

165 return JSONResponse(response_data) 

166 

167 except AuthenticationError as exc: 

168 logger.warning("auth_required", error=str(exc)) 

169 return JSONResponse( 

170 { 

171 "data": None, 

172 "errors": [ 

173 { 

174 "message": str(exc), 

175 "extensions": {"code": "AUTH_REQUIRED"}, 

176 } 

177 ], 

178 }, 

179 ) 

180 except Exception as exc: # noqa: BLE001 — controller boundary; all errors become HTTP 500 GraphQL error responses 

181 logger.exception("GraphQL execution error") 

182 return JSONResponse( 

183 {"errors": [{"message": f"Execution error: {exc}"}]}, 

184 status_code=500, 

185 ) 

186 

187 def _get_provider(self, request: GraphQLRequestProtocol) -> GraphQLProvider | None: 

188 """Get GraphQL provider from request app or instance. 

189 

190 Args: 

191 request: Current HTTP request. 

192 

193 Returns: 

194 GraphQL provider or None if not available. 

195 """ 

196 # Use injected provider if available 

197 if self._provider is not None: 

198 return self._provider 

199 

200 # Try to get from request app 

201 app = getattr(request, "app", None) 

202 if app is not None: 

203 provider: GraphQLProvider | None = getattr( 

204 app, "graphql_provider", lambda: None 

205 )() 

206 if provider is not None and not hasattr(provider, "_mock_return_value"): 

207 return provider 

208 

209 return None 

210 

211 

212class GraphQLSubscriptionController(ControllerProtocol): 

213 """GraphQL WebSocket subscription controller. 

214 

215 This controller handles GraphQL subscriptions via WebSocket 

216 at the configured path (default: ``/graphql/subscriptions``). 

217 """ 

218 

219 def __init__(self, provider: GraphQLProvider | None = None) -> None: 

220 """Initialize the subscription controller. 

221 

222 Args: 

223 provider: GraphQL provider instance. If None, will be resolved 

224 from the DI container at request time. 

225 """ 

226 self._provider = provider 

227 

228 @classmethod 

229 def collect_routes(cls) -> list[dict[str, Any]]: 

230 """Collect routes from controller methods.""" 

231 return [] 

232 

233 # Note: WebSocket routes use a different mechanism than HTTP. 

234 # This controller serves as a placeholder for WebSocket handler integration. 

235 # The actual WebSocket handling is done via GraphQLWSHandler in subscriptions. 

236 

237 

238__all__ = ["GraphQLController", "GraphQLSubscriptionController"]