diff --git a/routemaster/feeds.py b/routemaster/feeds.py index fad8d29..cb035b3 100644 --- a/routemaster/feeds.py +++ b/routemaster/feeds.py @@ -1,16 +1,28 @@ """Creation and fetching of feed data.""" import threading -from typing import Any, Dict, Union, Callable, Optional, Sequence +from typing import ( + TYPE_CHECKING, + Any, + Dict, + Union, + Callable, + Optional, + Sequence, +) from dataclasses import dataclass import requests from requests.sessions import Session from routemaster.utils import get_path, template_url -from routemaster.config import StateMachine +if TYPE_CHECKING: + from routemaster.config import StateMachine -def feeds_for_state_machine(state_machine: StateMachine) -> Dict[str, 'Feed']: + +def feeds_for_state_machine( + state_machine: 'StateMachine', +) -> Dict[str, 'Feed']: """Get a mapping of feed prefixes to unfetched feeds.""" return { x.name: Feed(x.url, state_machine.name) diff --git a/routemaster/tests/test_layering.py b/routemaster/tests/test_layering.py index 6cf9d36..3b15ced 100644 --- a/routemaster/tests/test_layering.py +++ b/routemaster/tests/test_layering.py @@ -1,6 +1,7 @@ import re import dis import pathlib +from typing import Iterable, Iterator import pytest @@ -23,7 +24,6 @@ ('cli', 'validation'), ('cli', 'middleware'), - ('exit_conditions', 'context'), ('exit_conditions', 'utils'), ('context', 'utils'), @@ -112,6 +112,29 @@ def test_layers_are_acyclic(): RE_ROUTEMASTER_MODULE = re.compile('^routemaster.([a-zA-Z0-9_]+)') +def _skip_type_checking_blocks( + instructions: Iterable[dis.Instruction], +) -> Iterator[dis.Instruction]: + state = None + for instruction in instructions: + if ( + instruction.opname == 'LOAD_NAME' and + instruction.argval == 'TYPE_CHECKING' + ): + state = 'TYPE_CHECKING' + continue + elif ( + state == 'TYPE_CHECKING' and + instruction.opname == 'POP_JUMP_IF_FALSE' + ): + state = 'JUMP' + elif state == 'JUMP' and not instruction.is_jump_target: + continue + else: + state = None + yield instruction + + @pytest.mark.skipif(networkx is None, reason="networkx is not installed") def test_layers(): root = pathlib.Path(routemaster.__file__).parent @@ -144,7 +167,7 @@ def test_layers(): last_import_source = None - for instruction in dis.get_instructions(code): + for instruction in _skip_type_checking_blocks(dis.get_instructions(code)): if instruction.opname == 'IMPORT_NAME': import_target = instruction.argval diff --git a/routemaster/utils.py b/routemaster/utils.py index 478ecf5..2c9bbd7 100644 --- a/routemaster/utils.py +++ b/routemaster/utils.py @@ -1,8 +1,9 @@ """Shared utilities.""" import contextlib -from typing import Any, Dict, Iterator, Sequence +from typing import TYPE_CHECKING, Any, Dict, Iterator, Sequence -from routemaster.logging import BaseLogger +if TYPE_CHECKING: + from routemaster.logging import BaseLogger def dict_merge(d1: Dict[str, Any], d2: Dict[str, Any]) -> Dict[str, Any]: @@ -38,7 +39,7 @@ def get_path(path: Sequence[str], d: Dict[str, Any]) -> Any: @contextlib.contextmanager -def suppress_exceptions(logger: BaseLogger) -> Iterator[None]: +def suppress_exceptions(logger: 'BaseLogger') -> Iterator[None]: """Catch all exceptions and log to a provided logger.""" try: yield