`orangeqs.juice.schemas.tasks`#

Task schemas for services.

Module Contents#

Classes#

`Ping`	Ping a service to check if it is responsive.
`SleepSync`	Sleep synchronously (block) for a given duration.
`SleepAsync`	Sleep asynchronously for a given duration.
`IPythonTask`	Base class for executing code in an IPython kernel.
`RawIPython`	Run raw code in an IPython kernel.
`AsyncSleepIPython`	Sleep asynchronously in an IPython kernel.
`RebuildEnvironmentDryRun`	Dry-run the rebuild of the environment of a service.
`TaskServerConfig`	Configuration for the task servers of OrangeQS Juice services.
`TaskServerConfigs`	Configurations for the task servers of OrangeQS Juice services.

API#

class orangeqs.juice.schemas.tasks.Ping(/, **data: Any)#

Bases: orangeqs.juice.task.Task

Ping a service to check if it is responsive.

Returns the same message back.

message: str#: ‘pong’

parallel: ClassVar[bool]#: True

class orangeqs.juice.schemas.tasks.SleepSync(/, **data: Any)#

Bases: orangeqs.juice.task.Task

Sleep synchronously (block) for a given duration.

duration: float#: 1.0

parallel: ClassVar[bool]#: True

class orangeqs.juice.schemas.tasks.SleepAsync(/, **data: Any)#

Bases: orangeqs.juice.task.Task

Sleep asynchronously for a given duration.

duration: float#: 1.0

parallel: ClassVar[bool]#: True

class orangeqs.juice.schemas.tasks.IPythonTask(/, **data: Any)#

Bases: orangeqs.juice.task.Task

Base class for executing code in an IPython kernel.

Must be subclassed with the code property implemented.

Returns the result of the executed code cell or an error message if execution fails. If the code executes successfully, returns {"status": "ok", "result": result}. If the result is not JSON serializable, uses repr(result) instead. If an error before or during execution of the cell, returns {"status": "error", "ename": "<Exception class>", "evalue": "<Exception value>"} instead.

Examples#

from pydantic import computed_field

class HelloFromIPython(IPythonTask):
    name: str

    @computed_field
    @property
    def code(self) -> str:
        return f"print('Hello, {self.name}!')"

Note that the order of the decorators matters here!

property hidden_fields: set[str]#: Return a set of field names to exclude from the display name.

abstract property code: str#

Return the code to execute based on the task payload.

This should be a computed property implemented by subclasses. It must be decorated with computed_field() and property(), in that order. See IPythonTask for an example.

classmethod type() → str#

Return the unique name of the task type.

Always returns "IPythonTask" to ensure all IPython tasks are routed to the same handler.

property display_name: str#

Return a human-readable name for the task.

Defaults to the class name and all fields except those in hidden_fields. For example: MyIPythonTask: param1=42, param2='foo'.

Can be overridden to provide a more descriptive name.

parallel: ClassVar[bool]#: False

class orangeqs.juice.schemas.tasks.RawIPython(/, **data: Any)#

Bases: orangeqs.juice.schemas.tasks.IPythonTask

Run raw code in an IPython kernel.

This task should not be subclassed, see IPythonTask instead.

raw_code: str#: None

property code: str#: Return the raw code to execute.

class orangeqs.juice.schemas.tasks.AsyncSleepIPython(/, **data: Any)#

Bases: orangeqs.juice.schemas.tasks.IPythonTask

Sleep asynchronously in an IPython kernel.

duration: float#: 1.0

property code: str#: Return the code to execute.

class orangeqs.juice.schemas.tasks.RebuildEnvironmentDryRun(/, **data: Any)#

Bases: orangeqs.juice.task.Task

Dry-run the rebuild of the environment of a service.

This runs a dry-run of the uv sync --dry-run command to show what changes would be made to the environment without actually applying them.

Returns a tuple[bool, list[str]] where the boolean indicates whether the dry-run was successful, and the list of strings contains the output lines from the dry-run.

class orangeqs.juice.schemas.tasks.TaskServerConfig(/, **data: Any)#