mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
new command routes
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
from typer import Typer
|
||||
|
||||
from .commands import init_handler, new_handler, run_handler
|
||||
from .commands import init_handler, new_handler, routes_handler, run_handler
|
||||
|
||||
|
||||
def main() -> None:
|
||||
@@ -25,6 +25,13 @@ def main() -> None:
|
||||
short_help="Create a new project with boilerplate",
|
||||
epilog="This will create a new directory with the project structure.",
|
||||
)(new_handler)
|
||||
|
||||
app.command(
|
||||
"routes",
|
||||
help="Creates a project and in it flat/src boilerplate architecture",
|
||||
short_help="Create a new project with boilerplate",
|
||||
epilog="This will create a new directory with the project structure.",
|
||||
)(routes_handler)
|
||||
|
||||
app()
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from .run import run_handler as run_handler
|
||||
from .init import init_handler as init_handler
|
||||
from .new import new_handler as new_handler
|
||||
from .new import new_handler as new_handler
|
||||
from .routes import routes_handler as routes_handler
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
__all__ = ["routes_handler"]
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.tree import Tree
|
||||
|
||||
from ..infrastructure.entrypoint_resolver.entity import (
|
||||
EntryPointAsApp,
|
||||
EntrypointResolver,
|
||||
ResolveFromStringError,
|
||||
)
|
||||
|
||||
|
||||
def routes_handler(entrypoint_path: str) -> None:
|
||||
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>"
|
||||
)
|
||||
|
||||
app_instance = EntrypointResolver[EntryPointAsApp](entrypoint_path).parse_entrypoint_with_type(
|
||||
entrypoint_callable_name
|
||||
)
|
||||
|
||||
app = app_instance.instance_object
|
||||
routers = app.registered_routers
|
||||
|
||||
console = Console()
|
||||
|
||||
stats: dict[str, int] = defaultdict(int)
|
||||
|
||||
tree = Tree(f"📦 [bold blue]App object:[/bold blue] {app!r}")
|
||||
|
||||
for router in routers:
|
||||
stats["routers"] += 1
|
||||
router_node = tree.add(f"📁 [bold green]Router:[/bold green] {router.title}")
|
||||
|
||||
for command in router.command_handlers:
|
||||
stats["commands"] += 1
|
||||
trigger = command.handled_command.trigger
|
||||
description = command.handled_command.description
|
||||
router_node.add(f"⚡ [cyan]Command:[/cyan] [bold]{trigger}[/bold] | [cyan]description:[/cyan] [bold]{description}[/bold] ")
|
||||
|
||||
stats_text = (
|
||||
f"📁 [bold]Total Routers:[/bold] {stats['routers']}\n"
|
||||
f"⚡ [bold]Total Commands:[/bold] {stats['commands']}"
|
||||
)
|
||||
|
||||
console.print(
|
||||
Panel(
|
||||
stats_text,
|
||||
title="[bold blue]App Stats[/bold blue]",
|
||||
expand=False,
|
||||
border_style="blue",
|
||||
)
|
||||
)
|
||||
console.print()
|
||||
console.print(tree)
|
||||
|
||||
@@ -14,7 +14,7 @@ def run_handler(entrypoint_path: str) -> None:
|
||||
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>"
|
||||
"Path to callable object that run orchestrator repl must be in the format <path/to/file.py>:<object_name> or <path.to.module>:<object_name>"
|
||||
)
|
||||
|
||||
runner = EntrypointResolver[CallableEntryPoint](entrypoint_path).parse_entrypoint_with_type(
|
||||
|
||||
@@ -73,11 +73,11 @@ class EntrypointResolver[T: (CallableEntryPoint, EntryPointAsApp)]:
|
||||
|
||||
def _parse_entrypoint_as_app(self, entrypoint_object_name: str) -> EntryPointAsApp:
|
||||
resolved_entrypoint = self._resolve_from_string(entrypoint_object_name)
|
||||
instance_object = resolved_entrypoint[1]
|
||||
instance_object = resolved_entrypoint.instance
|
||||
if not isinstance(instance_object, App):
|
||||
raise EntrypointNotAppInstanceError(repr(instance_object))
|
||||
|
||||
return EntryPointAsApp(raw_path=resolved_entrypoint[0], instance_object=instance_object)
|
||||
return EntryPointAsApp(raw_path=resolved_entrypoint.resolved_source_path, instance_object=instance_object)
|
||||
|
||||
def _resolve_from_string(self, entrypoint_object_name: str) -> ResolvedEntrypoint:
|
||||
raw_path = self._path_to_entrypoint
|
||||
|
||||
Reference in New Issue
Block a user