Initialize the app. Args: *args: Args to initialize the app with. **kwargs: Kwargs to initialize the app with. Raises: ValueError: If the event namespace is not provided in the config. Also, if there are multiple client su
(self, *args, **kwargs)
| 143 | theme: Optional[Component] = themes.theme(accent_color="blue") |
| 144 | |
| 145 | def __init__(self, *args, **kwargs): |
| 146 | """Initialize the app. |
| 147 | |
| 148 | Args: |
| 149 | *args: Args to initialize the app with. |
| 150 | **kwargs: Kwargs to initialize the app with. |
| 151 | |
| 152 | Raises: |
| 153 | ValueError: If the event namespace is not provided in the config. |
| 154 | Also, if there are multiple client subclasses of xt.State(Subclasses of xt.State should consist |
| 155 | of the DefaultState and the client app state). |
| 156 | """ |
| 157 | if "connect_error_component" in kwargs: |
| 158 | raise ValueError( |
| 159 | "`connect_error_component` is deprecated, use `overlay_component` instead" |
| 160 | ) |
| 161 | super().__init__(*args, **kwargs) |
| 162 | state_subclasses = BaseState.__subclasses__() |
| 163 | is_testing_env = constants.PYTEST_CURRENT_TEST in os.environ |
| 164 | |
| 165 | # Special case to allow test cases have multiple subclasses of xt.BaseState. |
| 166 | if not is_testing_env: |
| 167 | # Only one Base State class is allowed. |
| 168 | if len(state_subclasses) > 1: |
| 169 | raise ValueError( |
| 170 | "xt.BaseState cannot be subclassed multiple times. use xt.State instead" |
| 171 | ) |
| 172 | |
| 173 | self.state = State |
| 174 | # Get the config |
| 175 | config = get_config() |
| 176 | |
| 177 | # Add middleware. |
| 178 | self.middleware.append(HydrateMiddleware()) |
| 179 | |
| 180 | # Set up the API. |
| 181 | self.api = FastAPI() |
| 182 | self.add_cors() |
| 183 | self.add_default_endpoints() |
| 184 | |
| 185 | if self.state: |
| 186 | # Set up the state manager. |
| 187 | self._state_manager = StateManager.create(state=self.state) |
| 188 | |
| 189 | # Set up the Socket.IO AsyncServer. |
| 190 | self.sio = AsyncServer( |
| 191 | async_mode="asgi", |
| 192 | cors_allowed_origins="*" |
| 193 | if config.cors_allowed_origins == ["*"] |
| 194 | else config.cors_allowed_origins, |
| 195 | cors_credentials=True, |
| 196 | max_http_buffer_size=constants.POLLING_MAX_HTTP_BUFFER_SIZE, |
| 197 | ping_interval=constants.Ping.INTERVAL, |
| 198 | ping_timeout=constants.Ping.TIMEOUT, |
| 199 | ) |
| 200 | |
| 201 | # Create the socket app. Note event endpoint constant replaces the default 'socket.io' path. |
| 202 | self.socket_app = ASGIApp(self.sio, socketio_path="") |
no test coverage detected