Coverage for fastagency/app.py: 69%

1__all__ = ["FastAgency"] 9 ctx1afghbicde

2 

3import os 9 ctx1afghbicde

4from collections.abc import Awaitable, Generator 9 ctx1afghbicde

5from contextlib import contextmanager 9 ctx1afghbicde

6from typing import Any, Callable, Optional, Union 9 ctx1afghbicde

7 

8from .base import ( 9 ctx1afghbicde

9 ASGIProtocol, 

10 ProviderProtocol, 

11 Runnable, 

12 UIBase, 

13 WSGIProtocol, 

14) 

15from .exceptions import ( 9 ctx1afghbicde

16 FastAgencyASGINotImplementedError, 

17 FastAgencyWSGINotImplementedError, 

18) 

19from .logging import get_logger 9 ctx1afghbicde

20 

21logger = get_logger(__name__) 9 ctx1afghbicde

22 

23 

24class FastAgency: # Runnable 9 ctx1afghbicde

25 def __init__( 9 ctx1afghbicde

26 self, 

27 provider: ProviderProtocol, 

28 ui: UIBase, 

29 *, 

30 title: Optional[str] = None, 

31 description: Optional[str] = None, 

32 ) -> None: 

33 """Initialize the FastAgency object. 

34 

35 Args: 

36 provider (ProviderProtocol): The provider object to use 

37 ui (UI): The UI object to use 

38 title (Optional[str], optional): The title of the FastAgency. If None, the default string will be used. Defaults to None. 

39 description (Optional[str], optional): The description of the FastAgency. If None, the default string will be used. Defaults to None. 

40 """ 

41 # check if we need to start coverage 

42 logger.info("Checking if coverage is needed.") 5 ctx1abcde

43 coverage_process_start = os.environ.get("COVERAGE_PROCESS_START") 5 ctx1abcde

44 if coverage_process_start: 5 ctx1abcde

45 logger.info("Coverage process start detected") 

46 logger.info(f"Coverage configuration file: {coverage_process_start}") 

47 logger.info( 

48 "To ensure coverage is written out, terminate this program with SIGTERM" 

49 ) 

50 import coverage 

51 

52 coverage.process_startup() 

53 _self: Runnable = self 5 ctx1abcde

54 self._title = title or "FastAgency application" 5 ctx1abcde

55 self._description = description or "FastAgency application" 5 ctx1abcde

56 

57 logger.info( 5 ctx1abcde

58 f"Initializing FastAgency {self} with workflows: {provider} and UI: {ui}" 

59 ) 

60 self._provider = provider 5 ctx1abcde

61 self._ui = ui 5 ctx1abcde

62 logger.info(f"Initialized FastAgency: {self}") 5 ctx1abcde

63 

64 @property 9 ctx1afghbicde

65 def title(self) -> str: 9 ctx1afghbicde

66 """Return the title of the FastAgency.""" 

67 return self._title 1 ctx1j

68 

69 @property 9 ctx1afghbicde

70 def description(self) -> str: 9 ctx1afghbicde

71 """Return the description of the FastAgency.""" 

72 return self._description 

73 

74 def __str__(self) -> str: 9 ctx1afghbicde

75 """Return the string representation of the FastAgency.""" 

76 return f"<FastAgency title={self._title}>" 6 ctx1abcdej

77 

78 @property 9 ctx1afghbicde

79 def provider(self) -> ProviderProtocol: 9 ctx1afghbicde

80 """Return the provider object.""" 

81 return self._provider 6 ctx1abcdej

82 

83 @property 9 ctx1afghbicde

84 def ui(self) -> UIBase: 9 ctx1afghbicde

85 """Return the UI object.""" 

86 return self._ui 6 ctx1abcdej

87 

88 @contextmanager 9 ctx1afghbicde

89 def create(self, import_string: str) -> Generator[None, None, None]: 9 ctx1afghbicde

90 """Create the FastAgency.""" 

91 with self._ui.create(app=self, import_string=import_string): 5 ctx1abcde

92 yield 5 ctx1abcde

93 

94 def start( 9 ctx1afghbicde

95 self, 

96 *, 

97 import_string: str, 

98 name: Optional[str] = None, 

99 params: dict[str, Any], 

100 single_run: bool = False, 

101 ) -> None: 

102 """Start the FastAgency.""" 

103 self.ui.start( 5 ctx1abcde

104 app=self, 

105 import_string=import_string, 

106 name=name, 

107 params=params, 

108 single_run=single_run, 

109 ) 

110 

111 def __call__(self, *args: Any) -> Union[Awaitable[None], list[bytes]]: 9 ctx1afghbicde

112 if len(args) == 2 and callable(args[1]): 112 ↛ 116line 112 didn't jump to line 116 because the condition on line 112 was always true1 ctx1j

113 # WSGI interface 

114 environ, start_response = args 1 ctx1j

115 return self.handle_wsgi(environ, start_response) 1 ctx1j

116 elif len(args) == 3 and callable(args[1]) and callable(args[2]): 

117 # ASGI interface 

118 scope, receive, send = args 

119 scope_type = scope.get("type") 

120 if scope_type == "http": 

121 return self.handle_asgi(scope, receive, send) 

122 else: 

123 raise NotImplementedError( 

124 f"ASGI scope type '{scope_type}' not supported." 

125 ) 

126 else: 

127 raise TypeError(f"Invalid arguments for __call__: {args}") 

128 

129 def handle_wsgi( 9 ctx1afghbicde

130 self, environ: dict[str, Any], start_response: Callable[..., Any] 9 ctx1afghbicde

131 ) -> list[bytes]: 9 ctx1afghbicde

132 logger.debug(f"Handling WSGI request: {environ}") 1 ctx1j

133 if isinstance(self.ui, WSGIProtocol): 1 ctx1j

134 return self.ui.handle_wsgi(self, environ, start_response) 1 ctx1j

135 else: 

136 raise FastAgencyWSGINotImplementedError( 

137 f"WSGI interface not supported for UI: {self.ui.__class__.__name__}. Try running with 'fastapi run' or with a ASGI server like 'uvicorn'." 

138 ) 

139 

140 async def handle_asgi( 9 ctx1afghbicde

141 self, 

142 scope: dict[str, Any], 

143 receive: Callable[[dict[str, Any]], Awaitable[None]], 

144 send: Callable[[dict[str, Any]], Awaitable[None]], 

145 ) -> None: 

146 if isinstance(self.ui, ASGIProtocol): 

147 return await self.ui.handle_asgi(self, scope, receive, send) 

148 else: 

149 raise FastAgencyASGINotImplementedError( 

150 f"ASGI interface not supported for UI: {self.ui.__class__.__name__}. Try running with 'fastapi run' or with a WSGI server like 'gunicorn'/'waitress'." 

151 )