82 lines
2.4 KiB
Python
82 lines
2.4 KiB
Python
"""aios-cell command implementation for V2R1 Cell lifecycle."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
|
|
from .cell_manager import CellManager
|
|
from .ingress_service import serve
|
|
from .tenant_context import V2R1Error, ValidationError
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="aios-cell")
|
|
parser.add_argument("registry")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
create = sub.add_parser("create")
|
|
create.add_argument("tenant_id")
|
|
create.add_argument("cell_id")
|
|
create.add_argument("version")
|
|
create.add_argument("digest")
|
|
|
|
status = sub.add_parser("status")
|
|
status.add_argument("tenant_id")
|
|
|
|
transition = sub.add_parser("transition")
|
|
transition.add_argument("tenant_id")
|
|
transition.add_argument("state")
|
|
return parser
|
|
|
|
|
|
def build_serve_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="aios-cell serve")
|
|
parser.add_argument("--host", default="127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=18792)
|
|
return parser
|
|
|
|
|
|
def _domain_failure(error: V2R1Error) -> int:
|
|
code = "VALIDATION_ERROR" if isinstance(error, ValidationError) else "RUNTIME_ERROR"
|
|
print(
|
|
json.dumps({"code": code, "error": str(error)}, ensure_ascii=False, sort_keys=True),
|
|
file=sys.stderr,
|
|
)
|
|
return 3
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
values = list(sys.argv[1:] if argv is None else argv)
|
|
service_mode = (
|
|
bool(values) and values[0] == "serve"
|
|
and (len(values) == 1 or values[1].startswith("-"))
|
|
)
|
|
if service_mode:
|
|
serve_args = build_serve_parser().parse_args(values[1:])
|
|
serve(serve_args.host, serve_args.port)
|
|
return 0
|
|
|
|
args = build_parser().parse_args(values)
|
|
try:
|
|
manager = CellManager(args.registry)
|
|
if args.command == "create":
|
|
result = manager.create(
|
|
args.tenant_id, args.cell_id, args.version, args.digest
|
|
)
|
|
elif args.command == "status":
|
|
result = manager.status(args.tenant_id)
|
|
elif args.command == "transition":
|
|
result = manager.transition(args.tenant_id, args.state)
|
|
else:
|
|
return 2
|
|
except V2R1Error as error:
|
|
return _domain_failure(error)
|
|
|
|
print(json.dumps(result.__dict__, ensure_ascii=False, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|