Coverage for src/lexigram/web/integrations/graphql.py: 21%

1"""GraphQL integration for WebProvider. 

2 

3This module integrates GraphQL functionality with the web layer. 

4WebSocket subscriptions are registered directly on the Starlette application. 

5HTTP endpoints are handled by GraphQLController via the web contributor pattern. 

6""" 

7 

8from __future__ import annotations 

9 

10import contextlib 

11from typing import TYPE_CHECKING, Any 

12 

13from lexigram.contracts.graphql.protocols import DEFAULT_SUBSCRIPTIONS_PATH 

14from lexigram.logging import get_logger 

15 

16if TYPE_CHECKING: 

17 from starlette.applications import Starlette 

18 

19logger = get_logger(__name__) 

20 

21 

22class GraphQLIntegration: 

23 """Handles GraphQL WebSocket endpoint configuration. 

24 

25 HTTP endpoints are handled by GraphQLController using the standard 

26 Controller pattern. This integration only handles WebSocket subscriptions 

27 which require direct Starlette route registration. 

28 """ 

29 

30 @staticmethod 

31 async def configure(app: Starlette, container: Any | None = None) -> None: 

32 """Configure GraphQL WebSocket endpoints. 

33 

34 Registers a WebSocket route at ``/graphql/ws`` that lazily initialises 

35 the GraphQL transport on the first connection. This avoids boot-order 

36 issues when the GraphQL provider boots in parallel with the web provider. 

37 

38 Falls back to the legacy ``app.graphql_ws_controller_class`` attribute 

39 when the container is not available. 

40 

41 Args: 

42 app: The Starlette application instance. 

43 container: Optional DI container resolver for GraphQLProvider lookup. 

44 """ 

45 # ── Strategy 1: Lazy WS endpoint backed by the container ─────────── 

46 if container is not None: 

47 from starlette.routing import WebSocketRoute 

48 

49 from lexigram.contracts.graphql.protocols import WebSocketTransportProtocol 

50 

51 _transport_holder: dict[str, Any] = {} 

52 

53 async def _ws_endpoint(websocket: Any) -> None: 

54 transport = _transport_holder.get("transport") 

55 if transport is None: 

56 try: 

57 transport = await container.resolve(WebSocketTransportProtocol) 

58 _transport_holder["transport"] = transport 

59 logger.info("GraphQL WS transport initialised") 

60 except (LookupError, RuntimeError, OSError) as exc: 

61 logger.warning( 

62 "GraphQL WS transport init failed", 

63 error=str(exc), 

64 ) 

65 await websocket.close(code=1013) 

66 return 

67 

68 await transport.handle(websocket, app) 

69 

70 ws_path = DEFAULT_SUBSCRIPTIONS_PATH 

71 ws_route = WebSocketRoute(ws_path, _ws_endpoint) 

72 app.routes.insert(0, ws_route) 

73 logger.info("GraphQL WebSocket route registered at %s (lazy init)", ws_path) 

74 return 

75 

76 # ── Strategy 2 (legacy): app.graphql_ws_controller_class ─────────── 

77 graphql_ws_cls = None 

78 with contextlib.suppress(AttributeError, TypeError): 

79 graphql_ws_cls = getattr(app, "graphql_ws_controller_class", lambda: None)() 

80 

81 if graphql_ws_cls is not None and not hasattr( 

82 graphql_ws_cls, 

83 "_mock_return_value", 

84 ): 

85 try: 

86 from starlette.routing import WebSocketRoute 

87 

88 ws_path = getattr( 

89 app, "graphql_ws_path", lambda: DEFAULT_SUBSCRIPTIONS_PATH 

90 )() 

91 ws_route = WebSocketRoute(str(ws_path), graphql_ws_cls) 

92 app.routes.append(ws_route) 

93 logger.info("GraphQL WebSocket registered at %s", ws_path) 

94 except (ImportError, AttributeError, RuntimeError): 

95 logger.exception("Failed to register GraphQL WebSocket route")