mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
cli module creating
This commit is contained in:
@@ -1,6 +1,13 @@
|
||||
from typer import Typer
|
||||
|
||||
from .commands import run_handler
|
||||
|
||||
|
||||
def main():
|
||||
print(f'run from {__name__}')
|
||||
print('hello world')
|
||||
app = Typer()
|
||||
app.command("run")(run_handler)
|
||||
app.command("init")(run_handler)
|
||||
app()
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1 @@
|
||||
from .run import run_handler
|
||||
@@ -0,0 +1,62 @@
|
||||
__all__ = ["run_handler"]
|
||||
|
||||
import importlib
|
||||
from typing import Any
|
||||
|
||||
from argenta import App, Orchestrator
|
||||
|
||||
|
||||
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:
|
||||
if exc.name != module_str:
|
||||
raise exc from None
|
||||
raise ImportFromStringError(f'Could not import module "{module_str}".')
|
||||
|
||||
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(orchestrator_str: str, app_str: str | None = None) -> Any:
|
||||
orchestrator = import_from_string(orchestrator_str)
|
||||
|
||||
if not isinstance(orchestrator, Orchestrator):
|
||||
raise TypeError(f"Not an Orchestrator: {type(orchestrator).__name__}")
|
||||
|
||||
if app_str is not None:
|
||||
app = import_from_string(app_str)
|
||||
else:
|
||||
module_str = orchestrator_str.partition(":")[0]
|
||||
module = importlib.import_module(module_str)
|
||||
|
||||
app = None
|
||||
for attr_name in dir(module):
|
||||
attr = getattr(module, attr_name)
|
||||
if isinstance(attr, App):
|
||||
app = attr
|
||||
break
|
||||
|
||||
if app is None:
|
||||
raise ValueError(f'No App instance found in module "{module_str}"')
|
||||
|
||||
if not isinstance(app, App):
|
||||
raise TypeError(f"Not an App: {type(app).__name__}")
|
||||
|
||||
return orchestrator.start_polling(app)
|
||||
Reference in New Issue
Block a user