Index

Datatailr Scheduler Module

The datatailr.scheduler module provides a framework for scheduling and managing batch jobs.

The main job types are:

---

• Batch: Represents a batch job that can be scheduled and executed. The job can include multiple tasks which can be run in parallel or sequentially.
• Service: Represents a service job that runs continuously.
• App: Represents a web app or a dashboard, which can be built using one of the supported frameworks, such as Streamlit <https://streamlit.io/>, Dash <https://dash.plotly.com/>, or Panel <https://panel.holoviz.org/>_.
• Excel: Represents an Excel add-in.

App

Deploy a web application or dashboard to Datatailr.

This supports Streamlit, Dash, or any framework that can be started via a standard entrypoint (for example, streamlit run app.py).

Parameters:

• name

  (str) –

  App display name.

• entrypoint

  (Callable | str) –

  Callable that starts the app server.

• environment

  (Environment | None, default: DEV ) –

  Target environment for the job.

• image

  (Image | None, default: None ) –

  Container image to use for execution.

• run_as

  (str | User | None, default: None ) –

  User to execute the job as.

• resources

  (Resources | None, default: None ) –

  Resource requirements for the job.

• acl

  (ACL | None, default: None ) –

  Access control settings for the job.

• framework

  (str | None, default: None ) –

  Framework to use for the app.

• python_version

  (str, default: '3.12' ) –

  Python version to use.

• python_requirements

  (str | list[str], default: '' ) –

  Requirements to install (string or list).

• build_script_pre

  (str, default: '' ) –

  Shell script to run before build.

• build_script_post

  (str, default: '' ) –

  Shell script to run after build.

• env_vars

  (dict[str, str | int | float | bool] | None, default: None ) –

  Environment variables to set.

• get_existing

  (bool, default: False ) –

  Update an existing job with the same name.

Examples:

Minimal Streamlit app.

# app.py
import streamlit as st

def main():
    st.title("Hello Datatailr App")

if __name__ == "__main__":
    main()

Test locally.

streamlit run app.py

Deploy to Datatailr.

from app import main
from datatailr import App

app = App(
    name="Simple Dashboard App",
    entrypoint=main,
    python_requirements="streamlit")

app.run()

Parameters:

• name

  (str) –

  Display name for the app.

• entrypoint

  (Callable | str) –

  The callable (function) that starts the application.

• environment

  (Environment | None, default: DEV ) –

  Target environment for the deployment.

• image

  (Image | None, default: None ) –

  Pre-configured container Image. When None, an image is built from python_version, python_requirements, and the build scripts.

• run_as

  (str | User | None, default: None ) –

  User or username to run the app as. Defaults to the currently signed-in user.

• resources

  (Resources | None, default: None ) –

  CPU, memory, and optional GPU memory for the container. Defaults to Resources() (128 MB, 0.25 CPU, no GPU request). Set GPU memory with vram, using the same size strings as memory. If vram is omitted, the job JSON omits vram.

• acl

  (ACL | None, default: None ) –

  Access control list. Defaults to standard permissions for the current user.

• framework

  (str | None, default: None ) –

  Optional framework name (e.g., 'streamlit', 'dash', 'flask'). If not provided, the framework is inferred from the entrypoint.

• python_version

  (str, default: '3.12' ) –

  Python version for the container image.

• python_requirements

  (str | list[str], default: '' ) –

  Python dependencies (see Image).

• build_script_pre

  (str, default: '' ) –

  Dockerfile commands to run before installing requirements.

• build_script_post

  (str, default: '' ) –

  Dockerfile commands to run after installing requirements.

• env_vars

  (dict[str, str | int | float | bool] | None, default: None ) –

  Environment variables passed to the running container.

• get_existing

  (bool, default: False ) –

  If True, load and update an existing job definition with the same name instead of creating a new one.

• app_section

  (str, default: '' ) –

  The section to which the app belongs. If not provided, the app will be assigned to the default section. This affects the app launcher page.

• budget

  (Budget | str | None, default: None ) –

  Optional spend budget as Budget("name") or a name str. Will be set to the 'default' budget if not provided.

DuplicateJobNameError

Exception raised when a job with a duplicate name is added to the batch.

EntryPoint

Represents an entry point for a Datatailr job. This can be a function or a callable object.

Environment

Deployment stage for Datatailr compute processes.

Each environment is a fully isolated stage in the development and release lifecycle. Workflows, apps, services, and other jobs run in Docker containers scoped to the selected environment, so changes in one stage do not affect the others.

New jobs default to DEV. After validation, promote them through PRE (staging) to PROD with Job.promote().

When a job runs inside the platform, the active environment is read from the DATATAILR_JOB_ENVIRONMENT variable (see get_dt_env()).

Attributes:

• DEV –

  Development environment for building and testing changes.

• PRE –

  Pre-production (staging) environment for final validation.

• PROD –

  Production environment for live workloads.

Examples:

>>> from datatailr import Environment
>>> from datatailr.scheduler import App
>>> app = App(name="my_app", environment=Environment.DEV)
>>> app.promote(from_environment=Environment.DEV)

ExcelAddin

Represents an Excel add-in deployment on Datatailr.

An Excel add-in exposes Python functions as Excel worksheet functions, allowing users to call server-side computations directly from Excel spreadsheets.

At construction time, type annotations on @addin.expose(...) functions in the entrypoint source file are validated. Only plain int, float, list, bool, and str are allowed (plus Queue on a parameter named queue for streaming functions). Invalid annotations raise ValueError before the job definition is built.

Example

from datatailr import ExcelAddin
from datatailr.excel import Addin

addin_def = Addin("Options Pricer", "Option pricing functions")

@addin_def.expose(description="Black-Scholes price")
def price_option(spot, strike, vol, rate, expiry):
    ...

def __excel_main__(port=8080, ws_port=8000):
    addin_def.run(port, ws_port)

addin = ExcelAddin(
    name="Options Pricer",
    entrypoint=__excel_main__,
    python_requirements=["numpy", "scipy"],
)
addin.run()

Parameters:

• name

  (str) –

  Display name for the add-in.

• entrypoint

  (Callable) –

  The callable (function) that starts the add-in server. Its source file is scanned for @*.expose(...) functions; parameter and return annotations must use only int, float, list, bool, str, or Queue (on a parameter named queue).

• environment

  (Environment | None, default: DEV ) –

  Target environment for the deployment.

• image

  (Image | None, default: None ) –

  Pre-configured container Image.

• run_as

  (str | User | None, default: None ) –

  User or username to run the add-in as.

• resources

  (Resources | None, default: None ) –

  CPU, memory, and optional GPU memory for the container. Defaults to Resources() (128 MB, 0.25 CPU, no GPU request). Set GPU memory with vram, using the same size strings as memory. If vram is omitted, the job JSON omits vram.

• acl

  (ACL | None, default: None ) –

  Access control list.

• python_version

  (str, default: '3.12' ) –

  Python version for the container image.

• python_requirements

  (str | list[str], default: '' ) –

  Python dependencies (see Image).

• build_script_pre

  (str, default: '' ) –

  Dockerfile commands to run before installing requirements.

• build_script_post

  (str, default: '' ) –

  Dockerfile commands to run after installing requirements.

• env_vars

  (dict[str, str | int | float | bool] | None, default: None ) –

  Environment variables passed to the running container.

• get_existing

  (bool, default: False ) –

  If True, update an existing job definition.

• version

  (str | int | None, default: None ) –

  The version of the job to get.

• app_section

  (str, default: '' ) –

  The section to which the app belongs. If not provided, the app will be assigned to the default section. This affects the app launcher page.

• budget

  (Budget | str | None, default: None ) –

  Optional spend budget as Budget("name") or a name str. Will be set to the 'default' budget if not provided.

InstantRunHandle dataclass

Handle returned by Workflow.instant_run for script and notebook usage.

Job

Base class for all Datatailr compute processes (App, Service, ExcelAddin, and workflows).

A Job encapsulates the container image configuration, resource requirements, access control, and deployment metadata needed to schedule and run work on the Datatailr platform. In most cases you should use the concrete subclasses (App, Service, ExcelAddin) or the @workflow decorator rather than instantiating Job directly.

Parameters:

• name

  (str) –

  Display name for the job.

• environment

  (Environment | None, default: DEV ) –

  Target environment (dev / pre / prod).

• image

  (Image | None, default: None ) –

  Pre-configured container Image. When None, an image is built from python_version, python_requirements, and the build scripts.

• run_as

  (str | User | None, default: None ) –

  User or username under which the job runs. Defaults to the currently signed-in user.

• resources

  (Resources | None, default: None ) –

  CPU, memory, and optional GPU memory for the container. Set GPU memory with vram on Resources, using the same size strings as memory. If vram is omitted, the job JSON does not include a vram field.

• acl

  (ACL | None, default: None ) –

  Access control list. Defaults to standard permissions for the current user.

• python_version

  (str, default: 'auto' ) –

  Python version for the container image.

• python_requirements

  (str | list[str], default: '' ) –

  Python dependencies (see Image).

• build_script_pre

  (str, default: '' ) –

  Dockerfile commands before pip install.

• build_script_post

  (str, default: '' ) –

  Dockerfile commands after pip install.

• env_vars

  (dict[str, str | int | float | bool] | None, default: None ) –

  Environment variables passed to the container.

• type

  (JobType | None, default: UNKNOWN ) –

  The kind of job (workflow, service, app, excel).

• entrypoint

  (EntryPoint | None, default: None ) –

  The entry point that starts the job.

• get_existing

  (bool, default: False ) –

  If True, load an existing job with the same name and update it instead of creating a new definition.

• app_section

  (str, default: '' ) –

  The section to which the app belongs. If not provided, the app will be assigned to the default section. This affects the app launcher page.

• budget

  (Budget | str | None, default: None ) –

  Optional spend budget as Budget("name") (preferred) or a budget name str (same resolution). Serialized to JSON as budget_id (see internal-docs/budgets.md). If not provided, will be set to the 'default' budget.

budget property

Job budget. If existing definition was loaded and the current user doesn't have permissions for job's budget, this will return None

budget_id property

Budget id when set explicitly; None if omitted (platform default).

id property

Unique identifier for the job.

__enrich_task_runs__

Attach version-specific definition data (name, dependencies, resources) to each task run.

We can't rely on the currently-loaded workflow definition because the workflow might have been modified since the run completed, so the graph structure, task ids and resource sizing could all be stale. Instead we fetch the historical definition that produced this run and use it as the source of truth.

When the historical definition isn't available (e.g. it was pruned from the platform) we fall back to the current workflow's name map so the run details remain at least partially usable.

Tasks that are part of the workflow definition but never produced a run (e.g. they were skipped because an upstream task failed, or the run was a partial rerun) are surfaced as synthetic did_not_run task entries so the run details reflect the full workflow graph rather than only the tasks that executed.

__find_missing_python_packages

Compare the imports detected in the entry point code with the packages listed in python_requirements. Raise a warning if there are any imported packages that are not listed in python_requirements.

__get_existing__ classmethod

Retrieve an existing job instance from the Datatailr platform. Based on the job name and environment. If version is specified, it will retrieve the job with the specified version. Otherwise, it will retrieve the latest version of the job.

__resolve_task_id__

Resolve a workflow task id from either an explicit task_id or a task_name.

task_id takes precedence: when provided we use it directly without consulting the currently-loaded workflow definition. This matters for runs that come from older workflow versions whose graph (and therefore task names / ids) may have diverged from the current definition.

Returns None when the job is not a workflow (in which case the callers should address the job itself rather than a child task). Raises ValueError when task_name was given but no task with that name exists in the currently-loaded workflow.

__run_command__

Run a command in the context of the job. This is used to execute the job's entry point.

from_dict

Populate this Job instance from a dictionary (e.g. a platform API response).

Parameters:

• job_dict

  (dict) –

  A dictionary containing the serialized job data.

get_schedule_args

Returns additional arguments for scheduling the job. Override or extend this method as needed.

logs

Fetch the captured stdout (or stderr) of a task within a run.

Parameters:

• run_id

  (int) –

  The id of the run to read logs from.

• task_name

  (str | None, default: None ) –

  For workflows, the display name of the task whose logs to fetch. For non-workflow jobs (App, Service, ExcelAddin, …) task_name is informational only — the job itself is addressed.

• stderr

  (bool, default: False ) –

  If True, return the stderr stream instead of stdout.

• lines

  (int, default: 5000 ) –

  Maximum number of trailing lines to return.

• task_id

  (int | None, default: None ) –

  For workflows, the numeric task id (as exposed by run_details(...)['tasks'][i]['id']). When provided it takes precedence over task_name and is used as-is, which is the right choice for runs of older workflow versions whose graph may differ from the current one.

Returns:

• str –

  The captured log output as a string. May be empty when no log

• str –

  output is available for that run.

ls classmethod

List all jobs of this type.

Parameters:

• environment

  (Environment | str | None, default: None ) –

  The environment to list jobs from. If not provided, it will list jobs from all environments.

• filter_by

  (str | None, default: None ) –

  A filter expression to apply to the jobs. If not provided, it will list all jobs of this type.

Returns:

• list[dict] –

  A list of job dictionaries.

Example

from datatailr import App App.ls(environment="dev") # doctest: +SKIP [{..."environment":"dev",..."name":"test-app",..."type":"app",...}]

from datatailr import Service Service.ls(environment="dev") # doctest: +SKIP [{..."environment":"dev",..."name":"test-service",..."type":"service",...}]

promote

Promote the job to the next environment. This method is used to promote a version of the job from one environment to the next one. If none of the environments to promote from are specified, it defaults to promote from all environments.

:param from_environment: The environment to promote from. If None, it will promote from all environments. :param version: The version to promote. If None, it will promote the latest version. :return: A tuple of (success: bool, message: str).

from datatailr import Environment from datatailr.scheduler import Job job = Job(name="my_job", environment=Environment.DEV) job.promote(from_environment=Environment.DEV, version=3)

This will promote version 3 of the job from the DEV environment to the next environment (PRE).

remove

Remove the job definition from the platform (stopping it first if needed).

Parameters:

• all_versions

  (bool, default: True ) –

  If True (default), remove all versions of the job.

Returns:

• tuple[bool, str] –

  A tuple of (success: bool, message: str).

result

Fetch the cached return value of a task within a run.

Parameters:

• run_id

  (int | str) –

  The id of the run to read the result from.

• task_name

  (str | None, default: None ) –

  For workflows, the display name of the task whose result to load. For non-workflow jobs task_name is informational only.

• task_id

  (int | None, default: None ) –

  For workflows, the numeric task id (as exposed by run_details(...)['tasks'][i]['id']). When provided it takes precedence over task_name and is used as-is, which is the right choice for runs of older workflow versions whose graph may differ from the current one.

Returns:

• Any ( Any ) –

  The cached Python object returned by the task, or None if the task is still running, did not return anything, or its arguments cache entry has expired.

run

Run the job. This is equivalent to running job.save() and then job.start().

run_details

Get the details of a run.

Each task entry is enriched with workflow-definition data taken from the workflow version that actually produced this run (job_version). This means that historical runs surface the graph structure they were executed with, even if the workflow has since been redefined.

The returned tasks list covers every task in the workflow definition, not just the ones that executed. Tasks that never ran (e.g. skipped after an upstream failure, or excluded from a partial rerun) appear as synthetic entries with state == 'did_not_run'.

Example output

{'cost': 0.00012762248102453157, 'end_time': datetime.datetime(2026, 4, 29, 11, 46, 32, 14506), 'job_version': 1, 'original_run_id': -1, 'partial_rerun': False, 'scheduled_time': None, 'start_provisioning_time': datetime.datetime(2026, 4, 29, 11, 45, 57, 641051), 'start_time': datetime.datetime(2026, 4, 29, 11, 45, 59, 95784), 'state': 'completed', 'submit_time': datetime.datetime(2026, 4, 29, 11, 45, 57, 652091), 'tried_to_start_time': 0, 'name': 'Simple Data Pipeline', 'run_id': 12, 'tasks': [{'cost': 1.3018850769988e-05, 'end_time': datetime.datetime(2026, 4, 29, 11, 46, 17, 598195), 'scheduled_time': None, 'start_provisioning_time': datetime.datetime(2026, 4, 29, 11, 45, 59, 162721), 'start_time': datetime.datetime(2026, 4, 29, 11, 46, 8, 697732), 'state': 'completed', 'submit_time': datetime.datetime(2026, 4, 29, 11, 45, 57, 652091), 'tried_to_start_time': 0, 'name': 'get_data', 'id': 0, 'dependencies': [], 'resources': Resources(memory='256m', cpu=1.0)}]}

:param run_id: The id of the run to get the details of. :return: A dictionary containing the details of the run.

runs

List all runs of the job. The list includes the top-level data and not the tasks. To retrieve the tasks, use the run_details method.

Example output

[]

:param start_time: The start time of the runs to filter by. :param end_time: The end time of the runs to filter by. :param refresh: If True, it will refresh the cached runs. :return: A list of mother-job run dictionaries.

save

Save the job to the Datatailr platform. If the job already exists, it will be updated. The repository state is verified and an image is prepared for execution.

start

Start the job. This will start the job execution on a schedule for workflows if a schedule was specified. For other types of jobs and for workflows without a schedule the job will be run immediately.

stop

Stop all running instances of the job.

Returns:

• tuple[bool, str] –

  A tuple of (success: bool, message: str).

Example

from datatailr import Service svc = Service(name="My Service", get_existing=True) # doctest: +SKIP svc.stop() # doctest: +SKIP (True, '...')

to_dict

Convert the Job instance to a dictionary representation.

to_json

Convert the Job instance to a JSON string representation.

verify_repo_is_ready

Verify if the repository is ready for job execution. The check consists of two parts: 1. Check if there are uncommitted changes in the repository. 2. Check if the local commit matches the current branch's upstream ref on the remote. The upstream remote and ref come from the branch configuration, and the commit is read with git ls-remote. Returns a tuple of (branch: str, commit_hash: str).

versions

List all versions of the job in the specified environment If no environment is specified, it lists versions across all environments.

JobType

Enum representing different types of Datatailr jobs.

from_str classmethod

Create a JobType from a string while normalizing legacy aliases.

Resources dataclass

Represents the compute resources allocated to a job container.

Attributes:

• memory (str) –

  Memory limit as a string (e.g. "256m", "1g").

• cpu (float) –

  Number of CPU cores to allocate (e.g. 1, 0.5).

• vram (str | None) –

  GPU memory limit as a string (e.g. "256m", "1g"), the same form as memory. If omitted, the job JSON does not include a vram field.

Schedule

Build a schedule object for batch/workflow jobs using friendly fields.

Examples:

schedule = Schedule(at_hours=[0])
schedule = Schedule(at_minutes=[0, 30], weekdays=["Mon", "Wed", "Fri"])

You can either provide a raw cron_expression or use the human-readable helper parameters (which are compiled into cron syntax automatically).

Parameters:

• cron_expression

  (str, default: '' ) –

  A raw cron string (e.g. "0 */2 * * *"). If provided alongside helper parameters, the helpers take precedence.

• at_minutes

  (list[int] | None, default: None ) –

  Specific minutes past the hour to run (e.g. [0, 30] for ":00" and ":30").

• every_minute

  (int | None, default: None ) –

  Run every N minutes (e.g. 5 for every 5 minutes).

• at_hours

  (list[int] | None, default: None ) –

  Specific hours of the day to run (0 -- 23).

• every_hour

  (int | None, default: None ) –

  Run every N hours.

• weekdays

  (list[str] | None, default: None ) –

  Days of the week to run (e.g. ["Mon", "Wed", "Fri"]).

• day_of_month

  (int | None, default: None ) –

  Day of the month to run (1 -- 31).

• in_month

  (list[str] | None, default: None ) –

  Months to run in (e.g. ["Jan", "Apr", "Jul", "Oct"]).

• every_month

  (int | None, default: None ) –

  Run every N months.

• timezone

  (str | None, default: None ) –

  IANA timezone name (e.g. "America/New_York").

• run_after_job_uuid

  (str | None, default: None ) –

  UUID of a job that must complete before this schedule triggers.

• run_after_job_name

  (str | None, default: None ) –

  Name of a job that must complete before this schedule triggers.

• run_after_job_condition

  (str | None, default: None ) –

  Required completion status of the predecessor job (e.g. "success").

Example

>>> from datatailr import Schedule
>>> # Every weekday at 08:00 and 16:00 UTC
>>> s = Schedule(in_month=['Jul'], weekdays=['Sun'], every_minute=[24], at_hours=[2, 15])
>>> print(s)
0 */24 2,15 * 7 7

get_cron_string

Return the compiled cron string.

Returns:

• str –

  Cron string (str).

Examples:

>>> Schedule(
...     at_minutes=[0, 15, 30, 45],
...     at_hours=[0, 12],
...     weekdays=["Mon", "Wed", "Fri"],
...     day_of_month=15,
...     in_month=["Jan", "Jul"],
... ).get_cron_string()
'0 0,15,30,45 0,12 15 1,7 1,3,5'

Service

Represents a long-running background service deployment on Datatailr.

A service runs continuously (e.g., an API server, a message consumer, or any always-on process). It is restarted automatically if it exits.

Example

from datatailr import Service

# service.py
from flask import Flask

app = Flask(__name__)

@app.route("/health")
def health_check():
    return "OK"

# Service entrypoints receive the port from Datatailr.
def run_server(port):
    app.run("0.0.0.0", port=int(port), debug=False)

svc = Service(
    name="Simple Service",
    entrypoint=run_server,
    python_requirements=["flask"],
)
svc.run()

Parameters:

• name

  (str) –

  Display name for the service.

• entrypoint

  (Callable) –

  The callable (function) that starts the service.

• environment

  (Environment | None, default: DEV ) –

  Target environment for the deployment.

• image

  (Image | None, default: None ) –

  Pre-configured container Image.

• run_as

  (str | User | None, default: None ) –

  User or username to run the service as.

• resources

  (Resources | None, default: None ) –

  CPU, memory, and optional GPU memory for the container. Defaults to Resources() (128 MB, 0.25 CPU, no GPU request). Set GPU memory with vram, using the same size strings as memory. If vram is omitted, the job JSON omits vram.

• acl

  (ACL | None, default: None ) –

  Access control list.

• python_version

  (str, default: '3.12' ) –

  Python version for the container image.

• python_requirements

  (str | list[str], default: '' ) –

  Python dependencies (see Image).

• build_script_pre

  (str, default: '' ) –

  Dockerfile commands to run before installing requirements.

• build_script_post

  (str, default: '' ) –

  Dockerfile commands to run after installing requirements.

• env_vars

  (dict[str, str | int | float | bool] | None, default: None ) –

  Environment variables passed to the running container.

• get_existing

  (bool, default: False ) –

  If True, update an existing job definition.

• version

  (str | int | None, default: None ) –

  The version of the job to get.

• app_section

  (str, default: '' ) –

  The section to which the app belongs. If not provided, the app will be assigned to the default section. This affects the app launcher page.

• budget

  (Budget | str | None, default: None ) –

  Optional spend budget as Budget("name") or a name str. Will be set to the 'default' budget if not provided.

instant_run

Run the service immediately using the workstation runner image.

Same buildless path as :meth:Workflow.instant_run: submit as the {workstation}--runner per-user job so the scheduler forks a unique run ({runner}-{uuid}) that reuses the already-built runner image. After a successful call, :attr:_internal_name is the forked service name (use that for DNS / client connections).

Parameters:

• instant_run_mode

  (str, default: 'source' ) –

  Entry-point transport strategy.

  • "source" (default): bundle the entrypoint module tree and extract it into the runner image at start time.
  • "preinstalled": no bundling; the entrypoint must already exist on the workstation image (e.g. datatailr.joblib).
  • "auto": resolves to "source".
  • "cloudpickle": not supported for services.

• image_from_workstation

  (str | None, default: None ) –

  Workstation job name whose --runner image should be used. Defaults to DATATAILR_JOB_NAME or default-workstation.

Returns:

• tuple[bool, str] –

  A tuple of (success, message).

Examples:

User service with source bundling:

from datatailr import Service
from my_service import run_server

svc = Service(name="My Service", entrypoint=run_server)
ok, _ = svc.instant_run(instant_run_mode="source")

Preinstalled entrypoint (joblib broker style):

from datatailr import Service
from datatailr.joblib._broker import main as broker_main

svc = Service(name="joblib-broker", entrypoint=broker_main)
ok, _ = svc.instant_run(instant_run_mode="preinstalled")
broker_name = svc._internal_name  # forked unique name

Task

A single task (node) within a Workflow DAG.

Task instances are usually created by calling a @task()-decorated function inside a @workflow-decorated function. You can then chain configuration methods such as alias and set_resources.

Parameters:

• name

  (str) –

  Display name for the task.

• entrypoint

  (EntryPoint) –

  Entrypoint that runs when the task executes.

• resources

  (Resources | None, default: None ) –

  CPU, memory, and optional GPU memory. When None, the parent workflow's resources are used as defaults. If vram is omitted, the task JSON omits the vram field.

• dependencies

  (Sequence[str | Task] | None, default: None ) –

  Upstream tasks this task depends on, as task names or Task instances.

• dag

  (Workflow | None, default: None ) –

  Parent Workflow. When None, the current workflow manager is used.

• argument_mapping

  (dict[str, str] | None, default: None ) –

  Maps upstream task names to this task's argument names for dependency wiring.

• env_vars

  (dict[str, str | int | float | bool] | None, default: None ) –

  Environment variables for this task. Merged on top of variables inherited from the parent workflow.

• id

  (int | None, default: None ) –

  Explicit task ID. When None, the next available ID from the parent workflow is assigned.

args property writable

Arguments bound to this task.

Returns:

• dict[str, Any] –

  A dictionary of argument names to values.

id property

Unique identifier of this task within its workflow.

Returns:

• int –

  The task ID.

internal_name property

Internal platform name for this task.

Returns:

• str | None –

  {workflow_internal_name}-{task_id}, or None if the task has

• str | None –

  no parent workflow.

__call__

Allow the task to be called like a function.

Returns:

• Task –

  The same Task instance, for

• Task –

  chaining or functional-style usage.

alias

Set a display-name alias for this task.

Parameters:

• name

  (str) –

  New alias name. Must be unique within the parent workflow.

Returns:

• Task –

  This Task instance.

from_dict classmethod

Create a Task from a dictionary representation.

Parameters:

• job_dict

  (dict) –

  Dictionary produced by to_dict or returned by the platform.

• batch

  (Workflow) –

  Parent Workflow to attach the task to.

Returns:

• Task –

  A new Task instance.

run

Execute this task's entrypoint.

Sets DATATAILR_BATCH_ID, DATATAILR_JOB_ID, and DATATAILR_JOB_NAME in the entrypoint environment before running.

set_resources

Set CPU, memory, and GPU memory resources for this task.

Pass either a Resources instance via resources, or set memory, cpu, and/or vram individually. When resources is omitted, a new Resources is built from memory, cpu, and vram. memory and cpu fall back to the default task limits when None. vram is omitted unless given.

Parameters:

• resources

  (Resources | None, default: None ) –

  Complete resource specification. When provided, memory, cpu, and vram are ignored.

• memory

  (str | None, default: None ) –

  Memory limit (e.g. "256m", "1g"). Used only when resources is None.

• cpu

  (float | None, default: None ) –

  Number of CPU cores (e.g. 1, 0.5). Used only when resources is None.

• vram

  (str | None, default: None ) –

  GPU memory limit (e.g. "256m", "1g"), the same form as memory. Used only when resources is None. If omitted, the task JSON does not include a vram field.

Returns:

• Task –

  This Task instance.

to_dict

Convert this task to a dictionary representation.

Returns:

• –

  A dictionary suitable for serialization and platform submission.

to_json

Convert this task to a JSON string.

Returns:

• –

  JSON serialization of to_dict.

translate_dependencies

Resolve dependency names/tasks to upstream task IDs.

Returns:

• set[int] –

  A set of integer task IDs this task depends on.

update_env_vars

Update environment variables for this task.

Existing variables are preserved; keys in env_vars overwrite matching entries.

Parameters:

• env_vars

  (dict[str, str | int | float | bool]) –

  Environment variables to merge into this task's env.

Returns:

• Task –

  This Task instance.

TaskError

Exception raised for errors related to workflow tasks.

Workflow

Represents a workflow in the scheduler.

Inherits from Job and is used to define workflows with specific configurations.

next_job_id property

Returns a generator for the next task ID in the workflow.

add_job

Adds a task to the workflow.

:param job: The Task instance to add.

get_env_vars_copy

Returns a copy of the environment variables for the Workflow instance.

instant_run

Run the workflow immediately using the workstation runner image.

Parameters:

• instant_run_mode

  (str, default: 'auto' ) –

  Entry-point transport strategy. "auto" preserves current behavior: cloudpickle in Jupyter kernels, source bundle otherwise. Explicit "cloudpickle" and "source" bypass auto-detection.

• instant_run_embed

  (Sequence[str | ModuleType], default: () ) –

  Modules to force cloudpickle-by-value embedding. Accepts imported module objects or module-name strings. Only valid when resolved mode is "cloudpickle".

Raises:

• ValueError –

  In cloudpickle mode, if the runner image's Python minor version differs from the local one, or its python_requirements does not list cloudpickle.

instant_run_arguments

Arguments keyed by child number, for the instant-run entrypoint bundle.

Task references are stored as {TASK_CHILD_REF: child_id}. The runner expands those to {mother}-{child_id} once the forked name is known.

rerun

Rerun a run id with specific version (latest version by default). If full is true then all the tasks will be rerun, including the ones that already ran successfully.

:param run_id: The run id to rerun. :param version: The version to rerun. If None, it will rerun the latest version. :param full: If True, it will rerun the full workflow, including the ones that already ran successfully. :return: A tuple of (success: bool, message: str).

reset

Reset the Workflow instance to its initial state.

set_local_run

Set the local run flag for the Workflow instance.

:param local_run: A boolean indicating whether to run locally.

to_dict

Convert the Workflow instance to a dictionary representation.

to_json

Convert the Workflow instance to a JSON string representation.

task

Decorator that turns a function into a workflow task.

Apply this via the public alias @task(). When the decorated function is called inside a workflow-decorated function, the call returns a Task object that becomes a node in the workflow DAG. Dependencies between tasks are inferred automatically from the data flow between calls.

Tasks are only registered for deployment when they are invoked from within the body of a @workflow-decorated function, which builds a Workflow. Calling a @task-decorated function outside that context executes it locally as a normal Python function instead.

See Task below for methods available on the returned object (for example alias() and set_resources()).

Parameters:

• memory

  (str | None, default: None ) –

  Memory limit for this task's container (e.g. "256m", "1g"). When omitted, the workflow's memory is used.

• cpu

  (float | None, default: None ) –

  Number of CPU cores to allocate for this task (e.g. 1, 0.5). When omitted, the workflow's CPU count is used.

• vram

  (str | None, default: None ) –

  GPU memory limit for this task's container (e.g. "256m", "1g"), the same form as memory. When omitted, the workflow value is used. If that value is also omitted, the task JSON does not include a vram field.

Returns:

• Task ( Callable[[Callable[..., Any]], Callable[..., Any]] ) –

  The decorated function. When that function is called inside a @workflow-decorated function, the call returns a Task object that becomes a node in the workflow DAG.

Example

# tasks.py
from datatailr import task

@task(memory="512m", cpu=2)
def heavy_computation(x, y):
    return x ** y

@task()
def prepare():
    return 2

Then to deploy the workflow, use the following code:

# deploy.py
from datatailr import workflow
from tasks import heavy_computation, prepare

@workflow("My Workflow")
def my_workflow():
    base = prepare()
    heavy_computation(base, 10)

if __name__ == "__main__":
    my_workflow()  # deploy to Datatailr