Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/dishka/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
"alias",
"collect",
"decorate",
"exec_type_checking",
"from_context",
"make_async_container",
"make_container",
Expand All @@ -28,6 +29,7 @@
]

from .async_container import AsyncContainer, make_async_container
from .code_tools.type_tools import exec_type_checking
from .container import Container, make_container
from .entities.component import DEFAULT_COMPONENT, Component
from .entities.depends_marker import FromDishka
Expand Down
70 changes: 70 additions & 0 deletions src/dishka/code_tools/type_tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
import ast
import inspect
from collections.abc import Callable, Sequence
from types import ModuleType


def _make_fragments_collector(*, typing_modules: Sequence[str]) \

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

-> Callable[[ast.Module], list[ast.stmt]]:
def check_condition(expr: ast.expr) -> bool:
# searches for `TYPE_CHECKING`
if (
isinstance(expr, ast.Name)
and isinstance(expr.ctx, ast.Load)
and expr.id == "TYPE_CHECKING"
):
return True

# searches for `typing.TYPE_CHECKING`
if (
isinstance(expr, ast.Attribute)
and expr.attr == "TYPE_CHECKING"
and isinstance(expr.ctx, ast.Load)
and isinstance(expr.value, ast.Name)
and expr.value.id in typing_modules
and isinstance(expr.value.ctx, ast.Load)
):
return True
return False

def collect_type_checking_only_fragments(module: ast.Module) \
-> list[ast.stmt]:
fragments = []
for stmt in module.body:
if isinstance(stmt, ast.If) \
and not stmt.orelse and check_condition(stmt.test):
fragments.extend(stmt.body)

return fragments

return collect_type_checking_only_fragments


default_collector = _make_fragments_collector(typing_modules=["typing"])


def exec_type_checking(
module: ModuleType,
*,
collector: Callable[[ast.Module], list[ast.stmt]] = default_collector,
) -> None:
"""This function scans module source code,
collects fragments under ``if TYPE_CHECKING``
and ``if typing.TYPE_CHECKING`` and executes them in the context of module.
After these, all imports and type definitions became
available at runtime for analysis.

By default, it ignores ``if`` with ``else`` branch.

:param module: A module for processing
:param collector: A function collecting code fragments to execute
"""
source = inspect.getsource(module)
fragments = collector(ast.parse(source))
code = compile(ast.Module(fragments, type_ignores=[]),
f"<exec_type_checking of {module}>", "exec")
namespace = module.__dict__.copy()
exec(code, namespace) # noqa: S102
for k, v in namespace.items():
if not hasattr(module, k):
setattr(module, k, v)
Loading