(
*,
request: Request | WebSocket,
dependant: Dependant,
body: dict[str, Any] | FormData | bytes | None = None,
background_tasks: StarletteBackgroundTasks | None = None,
response: Response | None = None,
dependency_overrides_provider: Any | None = None,
dependency_cache: dict[DependencyCacheKey, Any] | None = None,
# TODO: remove this parameter later, no longer used, not removing it yet as some
# people might be monkey patching this function (although that's not supported)
async_exit_stack: AsyncExitStack,
embed_body_fields: bool,
)
| 596 | |
| 597 | |
| 598 | async def solve_dependencies( |
| 599 | *, |
| 600 | request: Request | WebSocket, |
| 601 | dependant: Dependant, |
| 602 | body: dict[str, Any] | FormData | bytes | None = None, |
| 603 | background_tasks: StarletteBackgroundTasks | None = None, |
| 604 | response: Response | None = None, |
| 605 | dependency_overrides_provider: Any | None = None, |
| 606 | dependency_cache: dict[DependencyCacheKey, Any] | None = None, |
| 607 | # TODO: remove this parameter later, no longer used, not removing it yet as some |
| 608 | # people might be monkey patching this function (although that's not supported) |
| 609 | async_exit_stack: AsyncExitStack, |
| 610 | embed_body_fields: bool, |
| 611 | ) -> SolvedDependency: |
| 612 | request_astack = request.scope.get("fastapi_inner_astack") |
| 613 | assert isinstance(request_astack, AsyncExitStack), ( |
| 614 | "fastapi_inner_astack not found in request scope" |
| 615 | ) |
| 616 | function_astack = request.scope.get("fastapi_function_astack") |
| 617 | assert isinstance(function_astack, AsyncExitStack), ( |
| 618 | "fastapi_function_astack not found in request scope" |
| 619 | ) |
| 620 | values: dict[str, Any] = {} |
| 621 | errors: list[Any] = [] |
| 622 | if response is None: |
| 623 | response = Response() |
| 624 | del response.headers["content-length"] |
| 625 | response.status_code = None # type: ignore |
| 626 | if dependency_cache is None: |
| 627 | dependency_cache = {} |
| 628 | for sub_dependant in dependant.dependencies: |
| 629 | sub_dependant.call = cast(Callable[..., Any], sub_dependant.call) |
| 630 | call = sub_dependant.call |
| 631 | use_sub_dependant = sub_dependant |
| 632 | if ( |
| 633 | dependency_overrides_provider |
| 634 | and dependency_overrides_provider.dependency_overrides |
| 635 | ): |
| 636 | original_call = sub_dependant.call |
| 637 | call = getattr( |
| 638 | dependency_overrides_provider, "dependency_overrides", {} |
| 639 | ).get(original_call, original_call) |
| 640 | use_path: str = sub_dependant.path # type: ignore |
| 641 | use_sub_dependant = get_dependant( |
| 642 | path=use_path, |
| 643 | call=call, |
| 644 | name=sub_dependant.name, |
| 645 | parent_oauth_scopes=sub_dependant.oauth_scopes, |
| 646 | scope=sub_dependant.scope, |
| 647 | ) |
| 648 | |
| 649 | solved_result = await solve_dependencies( |
| 650 | request=request, |
| 651 | dependant=use_sub_dependant, |
| 652 | body=body, |
| 653 | background_tasks=background_tasks, |
| 654 | response=response, |
| 655 | dependency_overrides_provider=dependency_overrides_provider, |
no test coverage detected
searching dependent graphs…