summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/_kwargs/dependencies.py
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/litestar/_kwargs/dependencies.py
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/_kwargs/dependencies.py')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_kwargs/dependencies.py119
1 files changed, 0 insertions, 119 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/_kwargs/dependencies.py b/venv/lib/python3.11/site-packages/litestar/_kwargs/dependencies.py
deleted file mode 100644
index 88ffb07..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_kwargs/dependencies.py
+++ /dev/null
@@ -1,119 +0,0 @@
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, Any
-
-from litestar.utils.compat import async_next
-
-__all__ = ("Dependency", "create_dependency_batches", "map_dependencies_recursively", "resolve_dependency")
-
-
-if TYPE_CHECKING:
- from litestar._kwargs.cleanup import DependencyCleanupGroup
- from litestar.connection import ASGIConnection
- from litestar.di import Provide
-
-
-class Dependency:
- """Dependency graph of a given combination of ``Route`` + ``RouteHandler``"""
-
- __slots__ = ("key", "provide", "dependencies")
-
- def __init__(self, key: str, provide: Provide, dependencies: list[Dependency]) -> None:
- """Initialize a dependency.
-
- Args:
- key: The dependency key
- provide: Provider
- dependencies: List of child nodes
- """
- self.key = key
- self.provide = provide
- self.dependencies = dependencies
-
- def __eq__(self, other: Any) -> bool:
- # check if memory address is identical, otherwise compare attributes
- return other is self or (isinstance(other, self.__class__) and other.key == self.key)
-
- def __hash__(self) -> int:
- return hash(self.key)
-
-
-async def resolve_dependency(
- dependency: Dependency,
- connection: ASGIConnection,
- kwargs: dict[str, Any],
- cleanup_group: DependencyCleanupGroup,
-) -> None:
- """Resolve a given instance of :class:`Dependency <litestar._kwargs.Dependency>`.
-
- All required sub dependencies must already
- be resolved into the kwargs. The result of the dependency will be stored in the kwargs.
-
- Args:
- dependency: An instance of :class:`Dependency <litestar._kwargs.Dependency>`
- connection: An instance of :class:`Request <litestar.connection.Request>` or
- :class:`WebSocket <litestar.connection.WebSocket>`.
- kwargs: Any kwargs to pass to the dependency, the result will be stored here as well.
- cleanup_group: DependencyCleanupGroup to which generators returned by ``dependency`` will be added
- """
- signature_model = dependency.provide.signature_model
- dependency_kwargs = (
- signature_model.parse_values_from_connection_kwargs(connection=connection, **kwargs)
- if signature_model._fields
- else {}
- )
- value = await dependency.provide(**dependency_kwargs)
-
- if dependency.provide.has_sync_generator_dependency:
- cleanup_group.add(value)
- value = next(value)
- elif dependency.provide.has_async_generator_dependency:
- cleanup_group.add(value)
- value = await async_next(value)
-
- kwargs[dependency.key] = value
-
-
-def create_dependency_batches(expected_dependencies: set[Dependency]) -> list[set[Dependency]]:
- """Calculate batches for all dependencies, recursively.
-
- Args:
- expected_dependencies: A set of all direct :class:`Dependencies <litestar._kwargs.Dependency>`.
-
- Returns:
- A list of batches.
- """
- dependencies_to: dict[Dependency, set[Dependency]] = {}
- for dependency in expected_dependencies:
- if dependency not in dependencies_to:
- map_dependencies_recursively(dependency, dependencies_to)
-
- batches = []
- while dependencies_to:
- current_batch = {
- dependency
- for dependency, remaining_sub_dependencies in dependencies_to.items()
- if not remaining_sub_dependencies
- }
-
- for dependency in current_batch:
- del dependencies_to[dependency]
- for others_dependencies in dependencies_to.values():
- others_dependencies.discard(dependency)
-
- batches.append(current_batch)
-
- return batches
-
-
-def map_dependencies_recursively(dependency: Dependency, dependencies_to: dict[Dependency, set[Dependency]]) -> None:
- """Recursively map dependencies to their sub dependencies.
-
- Args:
- dependency: The current dependency to map.
- dependencies_to: A map of dependency to its sub dependencies.
- """
- dependencies_to[dependency] = set(dependency.dependencies)
- for sub in dependency.dependencies:
- if sub not in dependencies_to:
- map_dependencies_recursively(sub, dependencies_to)