[Experimental] Python inline task configuration

Define task configuration directly in Python.
Inline task configuration is an experimental feature. Details are subject to change.
Python tasks can now be fully defined within Python without the need of the Airplane UI or task definition YAML files. To use it, your Python SDK version must be at least 0.3.12 and Airplane CLI version must be at least v0.3.44.

Writing tasks

Inline configured tasks must be defined in files that end in _airplane.py. Airplane looks for the _airplane.py suffix to determine which files contain task definitions. The Python SDK provides an airplane.task() decorator that configures your task, for example:
python
Copied
1
# my_task_airplane.py
2
import airplane
3
4
@airplane.task()
5
def my_task():
6
""""This is my task defined entirely in Python!"""
7
print("Hello world!")
No additional files are needed. Deploying this task creates an Airplane task that prints Hello World! when you run it.

Parameters

Task parameters are inferred from the parameters of the decorated Python function.
python
Copied
1
import airplane
2
3
@airplane.task()
4
def my_task(
5
# Task accepts a `name` argument
6
name: str,
7
):
8
print(f"Hello {name}!")

Parameter types

The function parameter's type hint is used determine the task parameter's type. All function parameters must have a type hint. The following table shows the supported Python types and their equivalent Airplane parameter types:
Python typeAirplane type
strshort text
airplane.LongTextlong text
airplane.SQLSQL
boolboolean
intinteger
floatnumber
airplane.Filefile
datetime.datedate
datetime.datetimedate + time
airplane.ConfigVarconfig variable

Optional parameters

Task parameters can be marked as optional via typing.Optional.
python
Copied
1
from typing import Optional
2
3
import airplane
4
5
@airplane.task()
6
def my_task(
7
# Task accepts an optional `name` argument
8
name: Optional[str],
9
):
10
if name is None:
11
print("Hellow world!")
12
else:
13
print(f"Hello {name}!")

Default parameters

Default parameter values can be specified via the Python default argument syntax.
python
Copied
1
import airplane
2
3
@airplane.task()
4
def my_task(
5
# Task accepts a `name` argument that defaults to Eric
6
name: str = "Eric",
7
):
8
print(f"Hello {name}!")

Additional configuration

Additional parameter configuration can be specified using a typing.Annotated typehint annotated with an airplane.ParamConfig object. For versions of Python prior to 3.9, typing_extensions.Annotated can be used.
python
Copied
1
class ParamConfig:
2
# Human-friendly identifier used to reference this parameter. Parameter slugs
3
# must be unique within an individual task / workflow. Defaults to the function
4
# argument's name.
5
slug: Optional[str] = None
6
# Parameter name displayed on the Airplane app. Defaults to the function
7
# argument's name in sentence case.
8
name: Optional[str] = None
9
# Parameter description displayed on the Airplane app. If not provided, the description
10
# will be pulled from the docstring of the decorated function.
11
description: Optional[str] = None
12
# Option constraint for the parameter. Select options allow users to specify exactly
13
# which values are allowed for a given parameter. Options may specify a label which
14
# is shown to the user on the Airplane app instead of the value.
15
options: Optional[AllOptions] = None
16
# Regex contraint for the parameter, only valid for string arguments.
17
regex: Optional[str] = None
To specify a short text parameter with select options, you can use typing.Annotated and airplane.ParamConfig like so:
python
Copied
1
from typing import Annotated
2
3
import airplane
4
5
@airplane.task()
6
def my_task(
7
# Task accepts a `name` argument that is either Eric or Beth
8
name: Annotated[
9
str,
10
airplane.ParamConfig(options=["Eric", "Beth"])
11
],
12
):
13
print(f"Hello {name}!")
For details on the types of parameters and supported configuration options, see the task definition reference.

Task configuration

The airplane.task() decorator allows you to define your task's attributes via the function's arguments.
python
Copied
1
def task(
2
# Human-friendly identifier used to reference this task. Must be unique
3
# across tasks and workflows. Defaults to function name.
4
slug: Optional[str] = None,
5
# Task name displayed on the Airplane app. Defaults to funcion name in sentence case.
6
name: Optional[str] = None,
7
# Task description displayed on the Airplane app. If not provided, the description
8
# will be pulled from the docstring of the decorated function.
9
description: Optional[str] = None,
10
# Whether or not this task requires a request to execute.
11
require_requests: bool = False,
12
# Whether or not this task allows self approvals.
13
allow_self_approvals: bool = False,
14
# How long a task can run (in seconds) for before it is automatically cancelled.
15
timeout: int = 3600,
16
# Constraints for which agents are allowed to execute this task's runs, only
17
# applicable for users with self hosted agents.
18
constraints: Optional[Dict[str, str]] = None,
19
# Resources to attach to this task. Resources can be accessed through environment
20
# variables or built-ins. Resources accessed by this task must be explicitly attached
21
# in the task's definition.
22
resources: Optional[List[airplane.Resource]] = None,
23
# Schedules to attach to this task. Schedules allow users to automatically run
24
# task on a recurring schedule.
25
schedules: Optional[List[airplane.Schedule]] = None,
26
# Enviornment variables to attach to this task. Environment variables allow users
27
# to configure constant values or reference config variables.
28
env_vars: Optional[List[airplane.EnvVar]] = None,
29
)
For details on the supported configuration options, see the task definition reference.

Docstrings

Descriptions for tasks and parameters can be optionally specified in the function's docstring. Descriptions explicitly defined in the task/parameter configuration take precedence over docstring descriptions.
python
Copied
1
import airplane
2
3
@airplane.task()
4
def my_task(name: str):
5
""""Says hello to you!
6
7
Args:
8
name: Your first and last name.
9
"""
10
print(f"Hello {name}!")
Airplane currently supports ReST, Google, Numpydoc-style and Epydoc docstrings.

Task discovery

Airplane imports your files to discover and deploy them. This allows you to set up dynamic configuration and use all of the power of the Python language to define your tasks. Multiple tasks can be defined in the same file.
python
Copied
1
import airplane
2
3
@airplane.task()
4
def my_first_task():
5
pass
6
7
@airplane.task()
8
def my_second_task():
9
pass

Developing and testing

You can run inline configured tasks locally using the airplane dev <task_file> CLI command. Since you can write multiple tasks per file, you may need to specify which task you would like to test by appending ::<function_name> to the file name.
  • If you only have one task defined in the file, leaving out the identifier will run the task.
  • If multiple tasks are defined in the file, leaving out the identifier will raise an error. Providing a function_name identifier that doesn't exist in the file will also raise an error.

Examples

All options in the task definition reference are supported.

Attach a resource to a task and create a schedule

python
Copied
1
import airplane
2
3
@airplane.task(
4
resouces=[
5
# Attach the resource with slug "backend_db" under the alias "db"
6
airplane.Resource(
7
slug="backend_db",
8
alias="db",
9
)
10
],
11
schedules=[
12
airplane.Schedule(
13
slug="midnight_daily",
14
cron="0 0 * * *",
15
description="Runs at midnight each day",
16
param_values={
17
# Schedule must provide a value for the `text` parameter since
18
# the parameter is required.
19
"text": "param value"
20
},
21
)
22
],
23
)
24
def my_task(text: str):
25
pass

Pass environment variables to a task and set label constraints for self-hosted agents

python
Copied
1
import airplane
2
3
@airplane.task(
4
env_vars=[
5
# Set an env var with a predefined value
6
airplane.EnvVar(
7
name="MY_ENV_VAR",
8
value="MY_ENV_VAR_VALUE",
9
),
10
# Set an env var from a defined config var
11
airplane.EnvVar(
12
name="MY_ENV_VAR",
13
config_var_name="my_config_var_name",
14
),
15
],
16
# Run only on agents with label "region:us-west-2"
17
constraints={
18
"region": "us-west-2",
19
},
20
)
21
def my_task():
22
pass

Require requests for task execution and set a custom timeout

python
Copied
1
import airplane
2
3
@airplane.task(
4
# Require requests to execute
5
require_requests=True,
6
allow_self_approvals=False,
7
# Set a timeout of 200 seconds
8
timeout=200,
9
)
10
def my_task():
11
pass

Task with a date parameter that has labeled select options and a default value

python
Copied
1
import datetime
2
from typing import Annotated
3
4
import airplane
5
6
@airplane.task()
7
def my_task(
8
my_date: Annotated[
9
datetime.date,
10
airplane.ParamConfig(
11
options=[
12
airplane.LabeledOption(
13
label="first date", value=datetime.date(2019, 1, 2)
14
),
15
airplane.LabeledOption(
16
label="second date", value=datetime.date(2019, 1, 2)
17
),
18
],
19
),
20
] = datetime.date(2019, 1, 1),
21
):
22
pass