Coverage for src/lexigram/web/transport/__init__.py: 40%

1"""Transport Domain - HTTP Data Movement""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7# Lazy loading to avoid circular imports 

8 

9if TYPE_CHECKING: 

10 from lexigram.web.transport.reactive import sse_from_stream 

11 from lexigram.web.transport.responses import ( # type: ignore[attr-defined] 

12 FastJSONResponse, 

13 FileResponse, 

14 HTMLResponse, 

15 HTMXResponse, 

16 JSONResponse, 

17 RedirectResponse, 

18 Response, 

19 StreamingResponse, 

20 file_response, 

21 html_response, 

22 json_response, 

23 redirect_response, 

24 streaming_response, 

25 ) 

26 from lexigram.web.transport.sse import ( 

27 EventSourceResponse, 

28 ServerSentEvent, 

29 sse_response, 

30 ) 

31 from lexigram.web.transport.websocket_guards import ( 

32 GuardedWebSocket, 

33 execute_websocket_guards, 

34 ) 

35 from lexigram.web.transport.websockets import WebSocket 

36 

37_LAZY_IMPORTS = { 

38 # Response 

39 "Response": ("lexigram.web.transport.responses", "Response"), 

40 "JSONResponse": ("lexigram.web.transport.responses", "JSONResponse"), 

41 "HTMLResponse": ("lexigram.web.transport.responses", "HTMLResponse"), 

42 "HTMXResponse": ("lexigram.web.transport.responses", "HTMXResponse"), 

43 "RedirectResponse": ("lexigram.web.transport.responses", "RedirectResponse"), 

44 "StreamingResponse": ("lexigram.web.transport.responses", "StreamingResponse"), 

45 "FileResponse": ("lexigram.web.transport.responses", "FileResponse"), 

46 "FastJSONResponse": ("lexigram.web.transport.responses", "FastJSONResponse"), 

47 # Response helpers 

48 "json_response": ("lexigram.web.transport.responses", "json_response"), 

49 "html_response": ("lexigram.web.transport.responses", "html_response"), 

50 "redirect_response": ("lexigram.web.transport.responses", "redirect_response"), 

51 "streaming_response": ("lexigram.web.transport.responses", "streaming_response"), 

52 "file_response": ("lexigram.web.transport.responses", "file_response"), 

53 # SSE 

54 "EventSourceResponse": ("lexigram.web.transport.sse", "EventSourceResponse"), 

55 "ServerSentEvent": ("lexigram.web.transport.sse", "ServerSentEvent"), 

56 "sse_response": ("lexigram.web.transport.sse", "sse_response"), 

57 # WebSocket 

58 "WebSocket": ("lexigram.web.transport.websockets", "WebSocket"), 

59 "GuardedWebSocket": ("lexigram.web.transport.websocket_guards", "GuardedWebSocket"), 

60 "execute_websocket_guards": ( 

61 "lexigram.web.transport.websocket_guards", 

62 "execute_websocket_guards", 

63 ), 

64 # Reactive 

65 "sse_from_stream": ("lexigram.web.transport.reactive", "sse_from_stream"), 

66} 

67 

68 

69def __getattr__(name: str) -> Any: 

70 """Lazy load attributes to avoid circular imports.""" 

71 if name in _LAZY_IMPORTS: 

72 import importlib 

73 

74 module_path, attr_name = _LAZY_IMPORTS[name] 

75 module = importlib.import_module(module_path) 

76 value = getattr(module, attr_name) 

77 globals()[name] = value 

78 return value 

79 raise AttributeError(f"module {__name__!r} has no attribute {name!r}") 

80 

81 

82def __dir__() -> list[str]: 

83 """List available attributes for IDE support.""" 

84 return list(__all__) + list(_LAZY_IMPORTS.keys()) 

85 

86 

87__all__ = [ 

88 "EventSourceResponse", 

89 "FastJSONResponse", 

90 "FileResponse", 

91 "GuardedWebSocket", 

92 "HTMLResponse", 

93 "HTMXResponse", 

94 "JSONResponse", 

95 "RedirectResponse", 

96 "Response", 

97 "ServerSentEvent", 

98 "StreamingResponse", 

99 "WebSocket", 

100 "execute_websocket_guards", 

101 "file_response", 

102 "html_response", 

103 "json_response", 

104 "redirect_response", 

105 "sse_response", 

106 "streaming_response", 

107 "sse_from_stream", 

108]