complete entrypoint resolver refactor

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)
 

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()

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 "<module>:<attribute>".'
-        )
-
-    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 <path/to/file.py>:<object_name>"
+        )
         
     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()

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

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,
         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)
@@ -100,11 +85,6 @@ class EntrypointResolver:
         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/to/file.py>"'
-            )
-    
         path = Path(file_path).resolve()
         if not path.exists():
             raise ResolveFromStringError(f'File "{file_path}" not found')

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"