chore: add cli tool -- init skeleton
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class ServiceStatus(str, Enum):
|
||||
ACTIVE = "active"
|
||||
DISABLED = "disabled"
|
||||
@@ -0,0 +1,48 @@
|
||||
from pathlib import Path
|
||||
from typing import ClassVar
|
||||
|
||||
import yaml
|
||||
from pydantic import BaseModel
|
||||
|
||||
from core.enums import ServiceStatus
|
||||
|
||||
HERE_DIR = Path(__file__).parent
|
||||
CONFIG_DIR = HERE_DIR / ".." / ".." / ".." / "config"
|
||||
SERVICE_REGISTRY_FILENAME = CONFIG_DIR / "services_registry.yml"
|
||||
|
||||
|
||||
class Service(BaseModel):
|
||||
id: int
|
||||
name: str
|
||||
status: ServiceStatus
|
||||
|
||||
def to_dict(self):
|
||||
return {"name": self.name, "status": str(self.status)}
|
||||
|
||||
class Config:
|
||||
use_enum_values = True
|
||||
|
||||
|
||||
class ServiceRegistry(BaseModel):
|
||||
config_path: ClassVar[str] = str(SERVICE_REGISTRY_FILENAME)
|
||||
|
||||
next_available_id: int
|
||||
services: dict[int, dict]
|
||||
|
||||
@classmethod
|
||||
def read_from_config(cls) -> "ServiceRegistry":
|
||||
with open(cls.config_path) as f:
|
||||
d = yaml.safe_load(f)
|
||||
|
||||
registry = ServiceRegistry.model_validate(d)
|
||||
return registry
|
||||
|
||||
def get_services(self) -> list[Service]:
|
||||
return [Service(id=svc_id, **svc_data) for svc_id, svc_data in self.services.items()]
|
||||
|
||||
def add_service(self, name: str) -> None:
|
||||
s = Service(id=self.next_available_id, name=name, status=ServiceStatus.ACTIVE)
|
||||
self.services.update({s.id: s.to_dict()})
|
||||
self.next_available_id += 1
|
||||
with open(self.config_path, "w") as f:
|
||||
yaml.safe_dump(self.model_dump(), f)
|
||||
+12
-3
@@ -1,6 +1,15 @@
|
||||
def main():
|
||||
pass
|
||||
import logging
|
||||
|
||||
import typer
|
||||
|
||||
import service
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
cli = typer.Typer(no_args_is_help=True, add_completion=False)
|
||||
|
||||
cli.add_typer(service.service, name="service")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
cli()
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
import typer
|
||||
|
||||
from core.models import ServiceRegistry
|
||||
|
||||
service = typer.Typer(no_args_is_help=True)
|
||||
|
||||
|
||||
@service.command()
|
||||
def add(
|
||||
name: str = typer.Option(..., "--name", "-n"),
|
||||
):
|
||||
registry = ServiceRegistry.read_from_config()
|
||||
registry.add_service(name)
|
||||
|
||||
|
||||
@service.command()
|
||||
def list():
|
||||
registry = ServiceRegistry.read_from_config()
|
||||
services = registry.get_services()
|
||||
for service in services:
|
||||
print(service)
|
||||
Reference in New Issue
Block a user