Coverage for src / lexigram / ai / relay / gateway / web / contributor.py: 85%

1"""Web contributor mounting the relay gateway's raw Starlette routes. 

2 

3The gateway package cannot import lexigram-web controller classes, so 

4``get_controllers`` is empty and the inbound relay routes are mounted 

5directly on the host application in ``mount_to_app``. The gateway 

6implementation is resolved from the request-scoped DI container at 

7request time. 

8""" 

9 

10from __future__ import annotations 

11 

12from typing import Any 

13 

14from lexigram.ai.relay.gateway.passthrough import PassthroughService 

15from lexigram.ai.relay.gateway.web.routes import build_routes 

16from lexigram.contracts.ai.relay import RelayGatewayProtocol 

17 

18__all__ = ["RelayGatewayWebContributor"] 

19 

20 

21class RelayGatewayWebContributor: 

22 """Web contributor for the relay gateway. 

23 

24 Implements the ``WebContributorProtocol`` surface duck-typed: the 

25 contributor id is ``"relay-gateway"``, no controllers or middleware 

26 are contributed (controller classes are lexigram-web types), and the 

27 four inbound relay routes are mounted on the Starlette application. 

28 Repeated mounts are idempotent: a path already present in 

29 ``app.routes`` is skipped. 

30 """ 

31 

32 @property 

33 def contributor_id(self) -> str: 

34 """Return the unique contributor identifier. 

35 

36 Returns: 

37 The string ``"relay-gateway"``. 

38 """ 

39 return "relay-gateway" 

40 

41 def get_controllers(self) -> list[type[Any]]: 

42 """Return the controller classes contributed by the gateway. 

43 

44 Controller classes are lexigram-web types, which the gateway 

45 package cannot import; raw Starlette routes are mounted in 

46 ``mount_to_app`` instead. 

47 

48 Returns: 

49 An empty list. 

50 """ 

51 return [] 

52 

53 def get_middleware(self) -> list[type[Any]]: 

54 """Return the middleware classes contributed by the gateway. 

55 

56 Middleware is contributed by the host application, not the 

57 gateway. 

58 

59 Returns: 

60 An empty list. 

61 """ 

62 return [] 

63 

64 async def mount_to_app(self, app: Any, container: object) -> None: 

65 """Mount the four inbound relay routes on *app*. 

66 

67 The gateway is resolved per request: the request-scoped 

68 container from ``request.state.container`` is preferred, falling 

69 back to the mount-time container. Routes already present in 

70 ``app.routes`` are skipped so repeated mounts never duplicate 

71 paths. 

72 

73 Args: 

74 app: The ASGI application (typically Starlette) to mount 

75 routes on. 

76 container: The DI container used as fallback resolution. 

77 """ 

78 

79 async def _resolve(request: Any) -> RelayGatewayProtocol: 

80 request_container: Any = ( 

81 getattr(request.state, "container", None) or container 

82 ) 

83 return await request_container.resolve(RelayGatewayProtocol) 

84 

85 async def _resolve_passthrough(request: Any) -> PassthroughService: 

86 request_container: Any = ( 

87 getattr(request.state, "container", None) or container 

88 ) 

89 return await request_container.resolve(PassthroughService) 

90 

91 routes = build_routes(_resolve, resolve_passthrough=_resolve_passthrough) 

92 for route in routes: 

93 path = route.path 

94 if any( 

95 getattr(existing, "path", None) == path 

96 for existing in getattr(app, "routes", []) 

97 ): 

98 continue 

99 app.add_route(path, route.endpoint, methods=["POST"])