(
plugin_path: str,
request: Request,
username: str,
)
| 187 | |
| 188 | |
| 189 | async def _call_plugin_extension( |
| 190 | plugin_path: str, |
| 191 | request: Request, |
| 192 | username: str, |
| 193 | ): |
| 194 | registered_web_apis = ( |
| 195 | request.app.state.core_lifecycle.star_context.registered_web_apis |
| 196 | ) |
| 197 | matched_api = _match_registered_web_api( |
| 198 | registered_web_apis, |
| 199 | plugin_path, |
| 200 | request.method, |
| 201 | ) |
| 202 | if not matched_api: |
| 203 | return {"status": "error", "message": "未找到该路由", "data": {}} |
| 204 | |
| 205 | view_handler, path_values = matched_api |
| 206 | plugin_name = plugin_path.strip("/").split("/", 1)[0].strip() or None |
| 207 | plugin_request = PluginRequest( |
| 208 | request, |
| 209 | path_params=path_values, |
| 210 | plugin_name=plugin_name, |
| 211 | username=username, |
| 212 | ) |
| 213 | app_adapter = getattr(request.app.state, "dashboard_app_adapter", None) |
| 214 | if app_adapter is None: |
| 215 | with bind_request_context(plugin_request): |
| 216 | return await run_maybe_async(lambda: view_handler(**path_values)) |
| 217 | |
| 218 | g_obj = DashboardRequestState() |
| 219 | g_obj.username = username |
| 220 | with bind_request_context(plugin_request): |
| 221 | return await call_request_view( |
| 222 | request, |
| 223 | app_adapter, |
| 224 | view_handler, |
| 225 | path_values, |
| 226 | g_obj=g_obj, |
| 227 | quart_compat_path=_plugin_extension_legacy_path(plugin_path, request), |
| 228 | ) |
| 229 | |
| 230 | |
| 231 | def _get_request_locale(request: Request, default: str = "zh-CN") -> str: |
no test coverage detected