diff --git a/metrics/__main__.py b/metrics/__main__.py index e6e09f1..c46ec6c 100644 --- a/metrics/__main__.py +++ b/metrics/__main__.py @@ -9,8 +9,7 @@ orchestrator = Orchestrator() def main() -> None: app.include_router(router) app.set_description_message_pattern( - lambda command, - description: f"[bold cyan]▸[/bold cyan] [bold white]{command}[/bold white] [dim]│[/dim] [yellow italic]{description}[/yellow italic]" + lambda command, description: f"[bold cyan]▸[/bold cyan] [bold white]{command}[/bold white] [dim]│[/dim] [yellow italic]{description}[/yellow italic]" ) orchestrator.run_repl(app) diff --git a/src/argenta/_cli/__main__.py b/src/argenta/_cli/__main__.py index 5016f60..ec05f2c 100644 --- a/src/argenta/_cli/__main__.py +++ b/src/argenta/_cli/__main__.py @@ -5,7 +5,7 @@ from .commands import run_handler, init_handler, new_handler def main() -> None: app = Typer() - app.command("run", help='Command to start the orchestrator repl; the path to the orchestrator is required')(run_handler) + app.command("run", help='Command to start the orchestrator repl; the path to the callable object is required')(run_handler) app.command("init", help="Creates a flat/src boilerplate architecture in an existing project")(init_handler) app.command("new", help="Creates a project and in it flat/src boilerplate architecture")(new_handler) app() diff --git a/src/argenta/_cli/commands/run.py b/src/argenta/_cli/commands/run.py index 220638e..2c46a2b 100644 --- a/src/argenta/_cli/commands/run.py +++ b/src/argenta/_cli/commands/run.py @@ -1,47 +1,29 @@ __all__ = ["run_handler"] -import importlib import os -import sys from pathlib import Path -from typing import Any +import sys + +from ..infrastructure.entrypoint_resolver.entity import ( + CallableEntryPoint, + EntrypointResolver, + ResolveFromStringError, +) -class ImportFromStringError(Exception): - pass - - -def import_from_string(import_str: str) -> Any: - module_str, _, attrs_str = import_str.partition(":") - if not module_str or not attrs_str: - raise ImportFromStringError( - f'Import string "{import_str}" must be in format ":".' - ) - - try: - module = importlib.import_module(module_str) - except ModuleNotFoundError as exc: - raise ImportFromStringError(f'Could not import module "{module_str}".') from exc - - instance = module - try: - for attr_str in attrs_str.split("."): - instance = getattr(instance, attr_str) - except AttributeError: - raise ImportFromStringError(f'Attribute "{attrs_str}" not found in module "{module_str}".') - - return instance - - -def run_handler(entry_point: str) -> None: +def run_handler(entrypoint_path: str) -> None: os.environ["RUN_FROM_ARGENTA_RUNNER"] = "1" - + entrypoint_path, _, entrypoint_callable_name = entrypoint_path.partition(":") + if not entrypoint_callable_name: + raise ResolveFromStringError( + "Path to callable object that run orchestrator repl must be in the format :" + ) + if str(Path.cwd()) not in sys.path: sys.path.insert(0, str(Path.cwd())) - runner = import_from_string(entry_point) + runner = EntrypointResolver(entrypoint_path).parse_entrypoint_with_type( + entrypoint_callable_name, CallableEntryPoint + ) - if not callable(runner): - raise TypeError(f'"{entry_point}" is not callable') - - runner() + runner.instance_object() diff --git a/src/argenta/_cli/infrastructure/entrypoint_resolver/__init__.py b/src/argenta/_cli/infrastructure/entrypoint_resolver/__init__.py new file mode 100644 index 0000000..2d9c198 --- /dev/null +++ b/src/argenta/_cli/infrastructure/entrypoint_resolver/__init__.py @@ -0,0 +1,7 @@ +from .entity import CallableEntryPoint as CallableEntryPoint +from .entity import EntryPointAsApp as EntryPointAsApp +from .entity import EntrypointResolver as EntrypointResolver +from .exceptions import EntrypointNotCallableError as EntrypointNotCallableError +from .exceptions import ResolveFromStringError as ResolveFromStringError +from .exceptions import CallableEntrypointNotMatchRequiredSignatureError as CallableEntrypointNotMatchRequiredSignatureError +from .exceptions import EntrypointNotAppInstanceError as EntrypointNotAppInstanceError \ No newline at end of file diff --git a/src/argenta/_cli/infrastructure/entrypoint_resolver.py b/src/argenta/_cli/infrastructure/entrypoint_resolver/entity.py similarity index 73% rename from src/argenta/_cli/infrastructure/entrypoint_resolver.py rename to src/argenta/_cli/infrastructure/entrypoint_resolver/entity.py index 42d2f67..ec91500 100644 --- a/src/argenta/_cli/infrastructure/entrypoint_resolver.py +++ b/src/argenta/_cli/infrastructure/entrypoint_resolver/entity.py @@ -1,30 +1,19 @@ -from dataclasses import dataclass -import inspect +__all__ = ['EntrypointResolver', 'EntryPointAsApp', 'CallableEntryPoint'] + import importlib.util +import inspect +from dataclasses import dataclass from pathlib import Path from typing import Callable, Protocol, cast, overload -from argenta import App +from argenta.app.models import App - -class ResolverError(Exception): - def __init__(self, entrypoint_as_repr: str) -> None: - self.entrypoint_as_repr = entrypoint_as_repr - -class ResolveFromStringError(ResolverError): - pass - -class EntrypointNotCallableError(ResolverError): - def __str__(self): - return f'Entrypoint {self.entrypoint_as_repr} is not callable' - -class CallableEntrypointNotMatchRequiredSignatureError(ResolverError): - def __str__(self) -> str: - return f'Callable entrypoint {self.entrypoint_as_repr} not match with required signature Callable[[], ...]' - -class EntrypointNotAppInstanceError(ResolverError): - def __str__(self): - return f'Entrypoint {self.entrypoint_as_repr} is not instance of App' +from .exceptions import ( + CallableEntrypointNotMatchRequiredSignatureError, + EntrypointNotAppInstanceError, + EntrypointNotCallableError, + ResolveFromStringError, +) class EntryPoint[T](Protocol): @@ -52,21 +41,17 @@ class EntrypointResolver: @overload def parse_entrypoint_with_type( - self, - entrypoint_object_name: str, - entrypoint_type: type[CallableEntryPoint] + self, entrypoint_object_name: str, entrypoint_type: type[CallableEntryPoint] ) -> EntryPoint[Callable[[], None]]: ... @overload def parse_entrypoint_with_type( - self, - entrypoint_object_name: str, - entrypoint_type: type[EntryPointAsApp] + self, entrypoint_object_name: str, entrypoint_type: type[EntryPointAsApp] ) -> EntryPoint[App]: ... def parse_entrypoint_with_type( - self, + self, entrypoint_object_name: str, - entrypoint_type: type[CallableEntryPoint] | type[EntryPointAsApp] + entrypoint_type: type[CallableEntryPoint] | type[EntryPointAsApp], ) -> EntryPoint[Callable[[], None]] | EntryPoint[App]: if entrypoint_type is CallableEntryPoint: return self._parse_callable_entrypoint(entrypoint_object_name) @@ -81,10 +66,10 @@ class EntrypointResolver: raise EntrypointNotCallableError(repr(instance_object)) instance_object_signature = inspect.signature(instance_object) required_params = instance_object_signature.parameters - + if required_params: raise CallableEntrypointNotMatchRequiredSignatureError(repr(instance_object)) - + instance_object = cast(Callable[[], None], instance_object) return CallableEntryPoint(raw_path=resolved_entrypoint[0], instance_object=instance_object) @@ -93,32 +78,27 @@ class EntrypointResolver: instance_object = resolved_entrypoint[1] if not isinstance(instance_object, App): raise EntrypointNotAppInstanceError(repr(instance_object)) - + return EntryPointAsApp(raw_path=resolved_entrypoint[0], instance_object=instance_object) - + def _resolve_from_string(self, entrypoint_object_name: str) -> tuple[str, object]: file_path: str = self._path_to_entrypoint attr_name: str = entrypoint_object_name - - if not file_path or not attr_name: - raise ResolveFromStringError( - f'"{self._path_to_entrypoint}" must be in format ""' - ) - + path = Path(file_path).resolve() if not path.exists(): raise ResolveFromStringError(f'File "{file_path}" not found') - + spec = importlib.util.spec_from_file_location(path.stem, path) if spec is None or spec.loader is None: raise ResolveFromStringError(f'Cannot load module from "{file_path}"') - + module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - + try: instance = getattr(module, attr_name) except AttributeError: raise ResolveFromStringError(f'"{attr_name}" not found in "{file_path}"') - + return file_path, instance diff --git a/src/argenta/_cli/infrastructure/entrypoint_resolver/exceptions.py b/src/argenta/_cli/infrastructure/entrypoint_resolver/exceptions.py new file mode 100644 index 0000000..7707dda --- /dev/null +++ b/src/argenta/_cli/infrastructure/entrypoint_resolver/exceptions.py @@ -0,0 +1,26 @@ +class ResolverError(Exception): + pass + + +class ResolveFromStringError(ResolverError): + pass + + +class EntrypointError(Exception): + def __init__(self, entrypoint_as_repr: str) -> None: + self.entrypoint_as_repr = entrypoint_as_repr + + +class EntrypointNotCallableError(EntrypointError): + def __str__(self): + return f"Entrypoint {self.entrypoint_as_repr} is not callable" + + +class CallableEntrypointNotMatchRequiredSignatureError(EntrypointError): + def __str__(self) -> str: + return f"Callable entrypoint {self.entrypoint_as_repr} not match with required signature Callable[[], ...]" + + +class EntrypointNotAppInstanceError(EntrypointError): + def __str__(self): + return f"Entrypoint {self.entrypoint_as_repr} is not instance of App"