bartender.web.api.job.interactive_apps#

Module for running interactive applications.

Attributes#

MediaEncoding

Classes#

InteractiveAppResult

Represents the result of running a InteractiveApp.

Functions#

_shell(→ InteractiveAppResult)

Executes a shell command in the specified job directory.

build_command(→ str)

Builds a command string for an interactive application.

encoding_of_prop(→ Union[MediaEncoding, None])

Determines the encoding of a property.

media_in_schema(→ dict[str, MediaEncoding])

Extracts properties from JSON schema that store media.

write_file(→ None)

Stages a file to disk.

get_path_in_payload(→ Union[Any, None])

Get the value at the specified path in the given dictionary.

set_path_in_payload(→ None)

Sets the value of a path in a dictionary.

stage_embedded_file(→ None)

Stage embedded file.

stage_embedded_files(→ AsyncGenerator[None, None])

Stages embedded files in the payload to disk.

run(→ InteractiveAppResult)

Runs an interactive application with the given payload and configuration.

Module Contents#

class bartender.web.api.job.interactive_apps.InteractiveAppResult#

Bases: pydantic.BaseModel

Represents the result of running a InteractiveApp.

returncode#

The return code of the InteractiveApp process.

stderr#

The standard error output of the InteractiveApp process.

stdout#

The standard output of the InteractiveApp process.

returncode: int#
stderr: str#
stdout: str#
async bartender.web.api.job.interactive_apps._shell(job_dir: pathlib.Path, command: str, timeout: float) → InteractiveAppResult#

Executes a shell command in the specified job directory.

Parameters:

  • job_dir (pathlib.Path) – The path to the job directory.

  • command (str) – The shell command to execute.

  • timeout (float) – The maximum time to wait for the command to finish. In seconds.

Returns:

The result of running the shell command.

Return type:

InteractiveAppResult

bartender.web.api.job.interactive_apps.build_command(payload: dict[Any, Any], app: bartender.config.InteractiveApplicationConfiguration) → str#

Builds a command string for an interactive application.

Parameters:
Returns:

A string representing the command to be executed.

Return type:

str

bartender.web.api.job.interactive_apps.MediaEncoding#
bartender.web.api.job.interactive_apps.encoding_of_prop(prop: dict[Any, Any]) → MediaEncoding | None#

Determines the encoding of a property.

Parameters:

prop (dict) – The property to determine the encoding of.

Returns:

The encoding of the property, or None if it is encoded with UTF-8, or False if it has no encoding.

Raises:

ValueError – If the encoding is not supported.

Return type:

Union[MediaEncoding, None]

bartender.web.api.job.interactive_apps.media_in_schema(schema: Any) → dict[str, MediaEncoding]#

Extracts properties from JSON schema that store media.

See https://json-schema.org/understanding-json-schema/reference/non_json_data

Parameters:

schema (Any) – The JSON schema to extract from.

Returns:

A dictionary containing the property path and their encoding.

Return type:

dict

async bartender.web.api.job.interactive_apps.write_file(encoded_data: str, encoding: MediaEncoding, fn: str) → None#

Stages a file to disk.

Parameters:

  • encoded_data (str) – The encoded data to be written to disk.

  • encoding (MediaEncoding) – The encoding of the encoded data.

  • fn (str) – The path to the file to be written.

Return type:

None

bartender.web.api.job.interactive_apps.get_path_in_payload(data: dict[Any, Any], path: str) → Any | None#

Get the value at the specified path in the given dictionary.

Parameters:

  • data (dict[Any, Any]) – The dictionary or list to search for the value.

  • path (str) – The path to the value, separated by dots. Can be nested. For example, “foo.bar.1” will set the value of data[“foo”][“bar”][1].

Returns:

The value at the specified path, or None if the path does not exist.

Return type:

Union[Any, None]

bartender.web.api.job.interactive_apps.set_path_in_payload(data: dict[Any, Any], path: str, value: Any) → None#

Sets the value of a path in a dictionary.

Parameters:

  • data (dict[Any, Any]) – The dictionary to modify.

  • path (str) – The path to set the value for. Can be nested. For example, “foo.bar.1” will set the value of data[“foo”][“bar”][1].

  • value (Any) – The value to set for the path.

Returns:

None

Return type:

None

async bartender.web.api.job.interactive_apps.stage_embedded_file(payload: dict[Any, Any], path: str, encoding: MediaEncoding, media_dir: str) → None#

Stage embedded file.

Extracts an embedded file from the given payload, writes it to disk in the specified media directory, and updates the payload to reference the newly written file.

Parameters:

  • payload (dict[Any, Any]) – The payload containing the embedded file.

  • path (str) – The path to the embedded file within the payload.

  • encoding (MediaEncoding) – The encoding of the embedded file.

  • media_dir (str) – The directory in which to write the extracted file.

Return type:

None

async bartender.web.api.job.interactive_apps.stage_embedded_files(payload: dict[Any, Any], schema: dict[Any, Any]) → AsyncGenerator[None, None]#

Stages embedded files in the payload to disk.

Parameters:

  • payload (dict[Any, Any]) – A dictionary containing the input data for the application.

  • schema (dict[Any, Any]) – The schema for the payload

Yields:

Nothing.

Return type:

AsyncGenerator[None, None]

async bartender.web.api.job.interactive_apps.run(job_dir: pathlib.Path, payload: dict[Any, Any], app: bartender.config.InteractiveApplicationConfiguration) → InteractiveAppResult#

Runs an interactive application with the given payload and configuration.

Parameters:
Returns:

The result of running the interactive application.

Return type:

InteractiveAppResult