Service

Service

Bases: ServiceModel

Service class representing a service in the system.

to_env_dict()

Convert EnvVariable list to dictionary format.

Source code in svs_core/docker/service.py

def to_env_dict(self) -> dict[str, str]:
    """Convert EnvVariable list to dictionary format."""
    return {env.key: env.value for env in self.env_variables}

to_ports_list()

Convert ExposedPort list to list of dictionaries format.

Source code in svs_core/docker/service.py

def to_ports_list(self) -> list[dict[str, Any]]:
    """Convert ExposedPort list to list of dictionaries format."""
    return [
        {"container": port.container_port, "host": port.host_port}
        for port in self.port_mappings
    ]

to_volumes_list()

Convert Volume list to list of dictionaries format.

Source code in svs_core/docker/service.py

def to_volumes_list(self) -> list[dict[str, Any]]:
    """Convert Volume list to list of dictionaries format."""
    return [
        {"container": vol.container_path, "host": vol.host_path}
        for vol in self.volume_mappings
    ]

to_labels_dict()

Convert Label list to dictionary format.

Source code in svs_core/docker/service.py

def to_labels_dict(self) -> dict[str, str]:
    """Convert Label list to dictionary format."""
    return {label.key: label.value for label in self.label_list}

to_healthcheck_dict()

Convert Healthcheck object to dictionary format.

Source code in svs_core/docker/service.py

def to_healthcheck_dict(self) -> dict[str, Any] | None:
    """Convert Healthcheck object to dictionary format."""
    if not self.healthcheck_config:
        return None

    result: dict[str, Any] = {"test": self.healthcheck_config.test}
    if self.healthcheck_config.interval:
        result["interval"] = self.healthcheck_config.interval
    if self.healthcheck_config.timeout:
        result["timeout"] = self.healthcheck_config.timeout
    if self.healthcheck_config.retries:
        result["retries"] = self.healthcheck_config.retries
    if self.healthcheck_config.start_period:
        result["start_period"] = self.healthcheck_config.start_period
    return result

create_from_template(name, template_id, user, domain=None, override_env=None, override_ports=None, override_volumes=None, override_command=None, override_healthcheck=None, override_labels=None, override_args=None, networks=None) classmethod

Creates a service from an existing template with overrides.

Parameters:

Name	Type	Description	Default
name	str	The name of the service.	required
user	User	The user who owns this service.	required
domain	str	The domain for this service.	None
override_env	dict	Environment variables to override template defaults.	None
override_ports	list[dict]	Exposed ports to override template defaults.	None
override_volumes	list[dict]	Volume mappings to override template defaults.	None
override_command	str	Command to override template default.	None
override_healthcheck	dict	Healthcheck configuration to override template default.	None
override_labels	dict	Container labels to override template defaults.	None
override_args	list	Command arguments to override template defaults.	None
networks	list	Networks to connect to.	None

Returns:

Name	Type	Description
Service	Service	The created service instance.

Raises:

Type	Description
ValueError	If name is empty or template_id doesn't correspond to an existing template.

Source code in svs_core/docker/service.py

@classmethod
def create_from_template(
    cls,
    name: str,
    template_id: int,
    user: User,
    domain: str | None = None,
    override_env: dict[str, str] | None = None,
    override_ports: list[dict[str, Any]] | None = None,
    override_volumes: list[dict[str, Any]] | None = None,
    override_command: str | None = None,
    override_healthcheck: dict[str, Any] | None = None,
    override_labels: dict[str, str] | None = None,
    override_args: list[str] | None = None,
    networks: list[str] | None = None,
) -> "Service":
    """Creates a service from an existing template with overrides.

    Args:
        name (str): The name of the service.
        user (User): The user who owns this service.
        domain (str, optional): The domain for this service.
        override_env (dict, optional): Environment variables to override template defaults.
        override_ports (list[dict], optional): Exposed ports to override template defaults.
        override_volumes (list[dict], optional): Volume mappings to override template defaults.
        override_command (str, optional): Command to override template default.
        override_healthcheck (dict, optional): Healthcheck configuration to override template default.
        override_labels (dict, optional): Container labels to override template defaults.
        override_args (list, optional): Command arguments to override template defaults.
        networks (list, optional): Networks to connect to.

    Returns:
        Service: The created service instance.

    Raises:
        ValueError: If name is empty or template_id doesn't correspond to an existing template.
    """
    try:
        template = Template.objects.get(id=template_id)
    except Template.DoesNotExist:
        raise ValueError(f"Template with ID {template_id} does not exist")

    env = {var.key: var.value for var in template.env_variables}
    if override_env:
        env.update(override_env)

    exposed_ports = [
        {"container": port.container_port, "host": port.host_port}
        for port in template.exposed_ports
    ]
    for override in override_ports or []:
        existing = next(
            (
                p
                for p in exposed_ports
                if p["container"] == override.get("container")
            ),
            None,
        )
        if existing:
            existing.update(override)
        else:
            exposed_ports.append(override)

    volumes = [
        {"container": vol.container_path, "host": vol.host_path}
        for vol in template.volumes
    ]
    for override in override_volumes or []:
        # TODO: fix later ;)
        existing = next(
            (v for v in volumes if v["container"] == override.get("container")),
            None,
        )  # type: ignore
        if existing:
            existing.update(override)
        else:
            volumes.append(override)

    labels = {label.key: label.value for label in template.label_list}
    if override_labels:
        labels.update(override_labels)

    healthcheck: dict[str, Any] | None = None
    if template.healthcheck_config:
        healthcheck = {"test": template.healthcheck_config.test}
        if template.healthcheck_config.interval:
            healthcheck["interval"] = template.healthcheck_config.interval
        if template.healthcheck_config.timeout:
            healthcheck["timeout"] = template.healthcheck_config.timeout
        if template.healthcheck_config.retries:
            healthcheck["retries"] = template.healthcheck_config.retries
        if template.healthcheck_config.start_period:
            healthcheck["start_period"] = template.healthcheck_config.start_period

    # TODO: allow partial overrides / merge
    if override_healthcheck:
        healthcheck = override_healthcheck

    command_to_use = (
        override_command if override_command is not None else template.start_cmd
    )
    args_to_use = override_args

    if args_to_use is None and template.args:
        args_to_use = template.args.copy()

    if args_to_use is not None:
        for i, arg in enumerate(args_to_use):
            if not isinstance(arg, str):
                args_to_use[i] = str(arg)

    return cls.create(
        name=name,
        template_id=template.id,
        user=user,
        domain=domain,
        image=template.image,
        exposed_ports=exposed_ports,
        env=env,
        volumes=volumes,
        command=command_to_use,
        healthcheck=healthcheck,
        labels=labels | {"svs_user": user.name},
        args=args_to_use,
        networks=networks,
    )

create(name, template_id, user, domain=None, container_id=None, image=None, exposed_ports=None, env=None, volumes=None, command=None, healthcheck=None, labels=None, args=None, networks=None, status='created', exit_code=None) classmethod

Creates a new service with all supported attributes.

Values not explicitly provided will be inherited from the template where applicable.

Parameters:

Name	Type	Description	Default
name	str	The name of the service.	required
template_id	int	The ID of the template to use.	required
user	User	The user who owns this service.	required
domain	str	The domain for this service.	None
container_id	str	The ID of an existing container.	None
image	str	Docker image to use, defaults to template.image if not provided.	None
exposed_ports	list[dict]	Exposed ports, defaults to template.default_ports if not provided.	None
env	dict	Environment variables, defaults to template.default_env if not provided.	None
volumes	list[dict]	Volume mappings, defaults to template.default_volumes if not provided.	None
command	str	Command to run in the container, defaults to template.start_cmd if not provided.	None
healthcheck	dict	Healthcheck configuration, defaults to template.healthcheck if not provided.	None
labels	dict	Container labels, defaults to template.labels if not provided.	None
args	list	Command arguments, defaults to template.args if not provided.	None
networks	list	Networks to connect to.	None
status	str	Initial service status, defaults to "created".	'created'
exit_code	int	Container exit code.	None

Returns:

Name	Type	Description
Service	Service	The created service instance.

Raises:

Type	Description
ValueError	If name is empty or template_id doesn't correspond to an existing template.

Source code in svs_core/docker/service.py

@classmethod
def create(
    cls,
    name: str,
    template_id: int,
    user: User,
    domain: str | None = None,
    container_id: str | None = None,
    image: str | None = None,
    exposed_ports: list[dict[str, Any]] | None = None,
    env: dict[str, str] | None = None,
    volumes: list[dict[str, Any]] | None = None,
    command: str | None = None,
    healthcheck: dict[str, Any] | None = None,
    labels: dict[str, str] | None = None,
    args: list[str] | None = None,
    networks: list[str] | None = None,
    status: str | None = "created",
    exit_code: int | None = None,
) -> "Service":
    """Creates a new service with all supported attributes.

    Values not explicitly provided will be inherited from the template where
    applicable.

    Args:
        name (str): The name of the service.
        template_id (int): The ID of the template to use.
        user (User): The user who owns this service.
        domain (str, optional): The domain for this service.
        container_id (str, optional): The ID of an existing container.
        image (str, optional): Docker image to use, defaults to template.image if not provided.
        exposed_ports (list[dict], optional): Exposed ports, defaults to template.default_ports if not provided.
        env (dict, optional): Environment variables, defaults to template.default_env if not provided.
        volumes (list[dict], optional): Volume mappings, defaults to template.default_volumes if not provided.
        command (str, optional): Command to run in the container, defaults to template.start_cmd if not provided.
        healthcheck (dict, optional): Healthcheck configuration, defaults to template.healthcheck if not provided.
        labels (dict, optional): Container labels, defaults to template.labels if not provided.
        args (list, optional): Command arguments, defaults to template.args if not provided.
        networks (list, optional): Networks to connect to.
        status (str, optional): Initial service status, defaults to "created".
        exit_code (int, optional): Container exit code.

    Returns:
        Service: The created service instance.

    Raises:
        ValueError: If name is empty or template_id doesn't correspond to an existing template.
    """
    if not name:
        raise ValueError("Service name cannot be empty")

    try:
        template = Template.objects.get(id=template_id)
    except Template.DoesNotExist:
        raise ValueError(f"Template with ID {template_id} does not exist")

    if image is None:
        image = template.image

    if exposed_ports is None:
        template_ports = template.exposed_ports
        exposed_ports = [
            {"container": port.container_port, "host": port.host_port}
            for port in template_ports
        ]
    else:
        for port in exposed_ports:
            if not isinstance(port, dict) or "container" not in port:
                raise ValueError(f"Invalid port specification: {port}")
            if "container" in port and port["container"] is not None:
                try:
                    port["container"] = int(port["container"])
                except (ValueError, TypeError):
                    raise ValueError(f"Container port must be an integer: {port}")
            if "host" in port and port["host"] is not None:
                try:
                    port["host"] = int(port["host"])
                except (ValueError, TypeError):
                    raise ValueError(f"Host port must be an integer: {port}")

    if env is None:
        template_env = template.env_variables
        env = {var.key: var.value for var in template_env}
    else:
        if not isinstance(env, dict):
            raise ValueError(f"Environment variables must be a dictionary: {env}")
        for key, value in env.items():
            if not isinstance(key, str) or not isinstance(value, str):
                raise ValueError(
                    f"Environment variable key and value must be strings: {key}={value}"
                )

    if volumes is None:
        template_volumes = template.volumes
        volumes = [
            {"container": vol.container_path, "host": vol.host_path}
            for vol in template_volumes
        ]
    else:
        for volume in volumes:
            if not isinstance(volume, dict) or "container" not in volume:
                raise ValueError(f"Invalid volume specification: {volume}")
            if "container" in volume and volume["container"] is not None:
                if not isinstance(volume["container"], str):
                    raise ValueError(f"Container path must be a string: {volume}")
            if "host" in volume and volume["host"] is not None:
                if not isinstance(volume["host"], str):
                    raise ValueError(f"Host path must be a string: {volume}")

    if command is None:
        command = template.start_cmd

    if healthcheck is None and template.healthcheck_config:
        healthcheck_obj = template.healthcheck_config
        healthcheck = {"test": healthcheck_obj.test}

        if healthcheck_obj.interval:
            healthcheck["interval"] = healthcheck_obj.interval
        if healthcheck_obj.timeout:
            healthcheck["timeout"] = healthcheck_obj.timeout
        if healthcheck_obj.retries:
            healthcheck["retries"] = healthcheck_obj.retries
        if healthcheck_obj.start_period:
            healthcheck["start_period"] = healthcheck_obj.start_period

    elif healthcheck is not None:
        if not isinstance(healthcheck, dict):
            raise ValueError(f"Healthcheck must be a dictionary: {healthcheck}")
        if "test" not in healthcheck:
            raise ValueError("Healthcheck must contain a 'test' field")
        if not isinstance(healthcheck["test"], list):
            raise ValueError(
                f"Healthcheck test must be a list of strings: {healthcheck['test']}"
            )

    if labels is None:
        template_labels = template.label_list
        labels = {label.key: label.value for label in template_labels}
    else:
        if not isinstance(labels, dict):
            raise ValueError(f"Labels must be a dictionary: {labels}")
        for key, value in labels.items():
            if not isinstance(key, str) or not isinstance(value, str):
                raise ValueError(
                    f"Label key and value must be strings: {key}={value}"
                )

    if args is None:
        args = template.args

    # Generate ports and volumes
    for port in exposed_ports:
        if "host" not in port or port["host"] is None:
            port["host"] = SystemPortManager.find_free_port()

    for volume in volumes:
        if "host" not in volume or volume["host"] is None:
            volume["host"] = SystemVolumeManager.generate_free_volume(
                user
            ).as_posix()

    service_instance = cls.objects.create(
        name=name,
        template_id=template_id,
        user_id=user.id,
        domain=domain,
        container_id=container_id,
        image=image,
        exposed_ports=exposed_ports,
        env=env,
        volumes=volumes,
        command=command,
        healthcheck=healthcheck,
        labels=labels,
        args=args,
        networks=networks,
        status=status,
        exit_code=exit_code,
    )

    system_labels = [Label(key="service_id", value=str(service_instance.id))]

    if service_instance.domain:
        system_labels.append(Label(key="caddy", value=service_instance.domain))

        if service_instance.exposed_ports:
            http_ports = [port for port in service_instance.port_mappings]

            if http_ports:
                upstreams = ", ".join(
                    f"{{upstreams {port.container_port}}}" for port in http_ports
                )
                if upstreams:
                    system_labels.append(Label(key="upstreams", value=upstreams))

    model_labels = service_instance.label_list

    all_labels = system_labels + model_labels

    if not service_instance.image:
        raise ValueError("Service must have an image specified")

    args_to_use = None
    if service_instance.args:
        args_to_use = []
        for arg in service_instance.args:
            if not isinstance(arg, str):
                args_to_use.append(str(arg))
            else:
                args_to_use.append(arg)

    get_logger(__name__).info(f"Creating service '{name}'")

    container = DockerContainerManager.create_container(
        name=name,
        image=service_instance.image,
        command=service_instance.command,
        args=args_to_use,
        labels=all_labels,
        ports={
            port["container"]: port["host"]
            for port in service_instance.to_ports_list()
        },
    )

    service_instance.container_id = container.id
    service_instance.save()

    return cast(Service, service_instance)

start()

Start the service's Docker container.

Source code in svs_core/docker/service.py

def start(self) -> None:
    """Start the service's Docker container."""
    if not self.container_id:
        raise ValueError("Service does not have a container ID")

    container = DockerContainerManager.get_container(self.container_id)
    if not container:
        raise ValueError(f"Container with ID {self.container_id} not found")

    container.start()
    self.status = ServiceStatus.RUNNING
    self.save()

stop()

Stop the service's Docker container.

Source code in svs_core/docker/service.py

def stop(self) -> None:
    """Stop the service's Docker container."""
    if not self.container_id:
        raise ValueError("Service does not have a container ID")

    container = DockerContainerManager.get_container(self.container_id)
    if not container:
        raise ValueError(f"Container with ID {self.container_id} not found")

    container.stop()
    self.status = ServiceStatus.STOPPED
    self.save()