labrec_bandBenjamin Labrecque
1933b906de
Deploy / deploy (push) Successful in 8s
Add option to use the GUI of an existing service when creating new services --------- Co-authored-by: Benjamin Labrecque <labrecque.benji@gmail.com> Reviewed-on: #12
50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
from pathlib import Path
|
|
|
|
import yaml
|
|
from pydantic import BaseModel
|
|
|
|
from core.enums import ServiceStatus
|
|
from models.service import Service
|
|
|
|
|
|
class ServiceRegistry(BaseModel):
|
|
next_available_ioc_port: int
|
|
services: list[dict]
|
|
|
|
@classmethod
|
|
def read_from_file(cls, path: Path) -> "ServiceRegistry":
|
|
with open(path) as f:
|
|
d = yaml.safe_load(f)
|
|
|
|
registry = ServiceRegistry.model_validate(d)
|
|
return registry
|
|
|
|
def get_service(self, name: str) -> Service:
|
|
return self.assert_service_exists(name=name)
|
|
|
|
def get_services(self) -> list[Service]:
|
|
return [Service(**svc_data) for svc_data in self.services]
|
|
|
|
def assert_service_exists(self, name: str) -> Service:
|
|
for svc_data in self.services:
|
|
service = Service(**svc_data)
|
|
if name == service.name_camel:
|
|
return service
|
|
|
|
raise ValueError(f"Could not find service with name: {name}")
|
|
|
|
def add_service(self, name: str) -> Service:
|
|
svc = Service(
|
|
name=name,
|
|
ioc_port=self.next_available_ioc_port,
|
|
status=ServiceStatus.ACTIVE,
|
|
)
|
|
self.services.append(svc.to_dict())
|
|
self.next_available_ioc_port += 1
|
|
|
|
return svc
|
|
|
|
def write_to_file(self, path: Path) -> None:
|
|
with open(path, "w") as f:
|
|
yaml.safe_dump(self.model_dump(), f, sort_keys=False)
|