from typing import TYPE_CHECKING, TypedDict
if TYPE_CHECKING:
from render import Render
try:
from .error import RenderError
except ImportError:
from error import RenderError
class TasksConfig(TypedDict):
schedule: str
command: str
class CronConfig(TypedDict):
tasks: list[TasksConfig]
SUPPORTED_REPOS = ["ixsystems/cron"]
class CronContainer:
def __init__(self, render_instance: "Render", name: str, image: str, config: CronConfig):
self._render_instance = render_instance
self._name = name
self._config = config
c = self._render_instance.add_container(name, image)
user, group = 568, 568
run_as = self._render_instance.values.get("run_as")
if run_as:
user = run_as["user"] or user # Avoids running as root
group = run_as["group"] or group # Avoids running as root
c.set_user(user, group)
c.deploy.resources.set_profile("low")
c.environment.add_env("METRICS_PORT", 8080)
c.healthcheck.set_test("curl", {"port": 8080, "path": "/health"})
c.remove_devices()
c.set_grace_period(60)
if not self._config.get("tasks"):
raise RenderError("Cron tasks not defined")
crontasks = []
for task in self._config["tasks"]:
sched = task["schedule"].split(" ")
if len(sched) != 5:
raise RenderError("Invalid cron schedule")
crontasks.append(f"{task['schedule']} {task['command']}")
c.configs.add("crontasks", "\n".join(crontasks), "/crontasks")
self._get_repo(image)
# Store container for further configuration
# For example: c.depends.add_dependency("other_container", "service_started")
self._container = c
@property
def container(self):
return self._container
def _get_repo(self, image):
images = self._render_instance.values["images"]
if image not in images:
raise RenderError(f"Image [{image}] not found in values. Available images: [{', '.join(images.keys())}]")
repo = images[image].get("repository")
if not repo:
raise RenderError("Could not determine repo")
if repo not in SUPPORTED_REPOS:
raise RenderError(f"Unsupported repo [{repo}] for cron. Supported repos: {', '.join(SUPPORTED_REPOS)}")
return repo