Skip to content

Commit

Permalink
feat(stubs): support reuse of StubSource instances, improvements.
Browse files Browse the repository at this point in the history
Signed-off-by: Braden Mars <bradenmars@bradenmars.me>
  • Loading branch information
BradenM committed Dec 12, 2022
1 parent 24ef2fa commit b873a62
Showing 1 changed file with 28 additions and 19 deletions.
47 changes: 28 additions & 19 deletions micropy/stubs/source.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,40 +11,55 @@
import abc
import shutil
import tempfile
from contextlib import contextmanager
from functools import partial
from contextlib import ExitStack, contextmanager
from functools import partial, reduce
from pathlib import Path
from typing import Any, Callable, ContextManager, Iterable, Optional, Union
from typing import TYPE_CHECKING, Any, Callable, ContextManager, Optional, Union, cast

import attrs
from micropy import utils
from micropy.logger import Log
from micropy.utils.types import PathStr
from typing_extensions import Protocol

if TYPE_CHECKING:
pass


class LocateStrategy(Protocol):
@abc.abstractmethod
def prepare(self, location: PathStr) -> Union[PathStr, tuple[PathStr, Callable[..., Any]]]:
...


logger = Log.add_logger(__name__)
logger = Log.add_logger(__name__, show_title=False)


@attrs.define
class StubSource:
"""Handles sourcing stubs."""

location: PathStr = attrs.field()
locators: list[LocateStrategy] = attrs.field()
location: Optional[PathStr] = attrs.field(default=None)

@locators.default
def _default_locators(self: StubSource) -> list[LocateStrategy]:
return [RemoteStubLocator(), StubInfoSpecLocator()]

def _do_locate(self, stack: ExitStack, path: PathStr, locator: LocateStrategy) -> PathStr:
logger.debug(f"running (strategy:{locator}) @ (location:{path})")
response = locator.prepare(path)
parts = iter(response if isinstance(response, tuple) else (response,))
path = next(parts, path)
teardown = next(parts, None)
if teardown:
logger.debug(f"adding teardown callback for: {locator}")
stack.callback(teardown)
logger.debug(f"results of (strategy:{locator}) -> (location:{path})")
return path

@contextmanager
def ready(self) -> ContextManager[PathStr]:
def ready(self, location: Optional[PathStr] = None) -> ContextManager[PathStr]:
"""Yields prepared Stub Source.
Allows StubSource subclasses to have a preparation
Expand All @@ -54,18 +69,12 @@ def ready(self) -> ContextManager[PathStr]:
Resolved PathLike object to stub source
"""
path = self.location
teardown = lambda: None
for locator in self.locators:
logger.debug(f"running (strategy:{locator}) @ (location:{path})")
response = locator.prepare(path)
parts = iter(response if isinstance(response, Iterable) else (response,))
path = next(parts, path)
teardown = next(parts, teardown)
logger.debug(f"results of (strategy:{locator}) -> (location:{path})")
yield path
if teardown:
teardown()
with ExitStack() as stack:
reducer = cast(
Callable[[PathStr, LocateStrategy], PathStr], partial(self._do_locate, stack)
)
path = reduce(reducer, self.locators, location or self.location)
yield path


@attrs.define
Expand Down Expand Up @@ -128,4 +137,4 @@ def get_source(location, **kwargs):
obj: Either Local or Remote StubSource Instance
"""
return StubSource(location, **kwargs)
return StubSource(**kwargs, location=location)

0 comments on commit b873a62

Please sign in to comment.