Python task configuration

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!")

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

python
Copied
1
import airplane
2

3
# You can share code between tasks by defining it outside of the task implementation function.
4
def get_airplane_task_name(slug: str) -> str:
5
    return f"Airplane task {slug}"
6

7
# But be careful! This code will be executed when you run `airplane deploy`.
8
make_api_call()
9

10
@airplane.task(
11
    slug="my_task",
12
    name=get_airplane_task_name("my_task"),
13
)
14
def my_task():
15
    print("Hello world!")

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}!")

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}!")

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}!")

python
Copied
1
import airplane
2

3
@airplane.task(
4
    resources=[
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

python
Copied
1
import airplane
2

3
@airplane.task(
4
    permissions=airplane.ExplicitPermissions(
5
        viewers=airplane.PermissionAssignees(groups=["support"]),
6
        executers=airplane.PermissionAssignees(users=["jordan@airplane.dev"]),
7
    )
8
)
9
def my_task(text: str):
10
    pass

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

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

python
Copied
1
import datetime
2
# For versions of Python prior to 3.9, `Annotated` can be imported from the `typing_extensions` package.
3
from typing import Annotated
4

5
import airplane
6

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

python
Copied
1
# For versions of Python prior to 3.9, `Annotated` can be imported from the `typing_extensions` package.
2
from typing import Annotated
3

4
import airplane
5

6
@airplane.task()
7
def my_task(
8
    # Task accepts a `name` argument that is either Eric or Beth
9
    name: Annotated[
10
        str,
11
        airplane.ParamConfig(options=["Eric", "Beth"])
12
    ],
13
):
14
    print(f"Hello {name}!")

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}!")

yaml
Copied
1
parameters:
2
- slug: name
3
  name: Name
4
  type: shorttext
5
  description: The user's name.
6
  default: Alfred Pennyworth
7
  required: false
8
  options:
9
  - Alfred Pennyworth
10
  - Bruce Wayne
11
  regex: "^[a-zA-Z ]+$"

yaml
Copied
1
options:
2
- Alfred Pennyworth
3
- Bruce Wayne
4
- label: BW
5
  value: Bruce Wayne

yaml
Copied
1
constraints:
2
  aws-region: us-west-2

yaml
Copied
1
schedules:
2
  my_first_schedule:
3
    cron: 0 0 * * *
4
    name: My First Schedule
5
    description: This is my first daily midnight UTC schedule!
6
    paramValues:
7
      my_first_param: my_first_param_value

yaml
Copied
1
resources:
2
  db: prod_postgres_db

yaml
Copied
1
resources:
2
  - prod_postgres_db
3
  - prod_rest_api

yaml
Copied
1
permissions: "team_access"

yaml
Copied
1
permissions:
2
  viewers:
3
    groups:
4
      - customer_success
5
    users:
6
      - success@airplane.dev
7
  requesters:
8
    groups:
9
      - support
10
    users:
11
      - support@airplane.dev
12
  executers:
13
    groups:
14
      - devs
15
  admins:
16
    groups:
17
      - team_admins

yaml
Copied
1
python:
2
  entrypoint: my_task.py

yaml
Copied
1
entrypoint: my_folder/my_task.py

yaml
Copied
1
envVars:
2
  # From value
3
  API_ENDPOINT:
4
    value: api.myco.dev/
5
  # From config variable
6
  MY_SECRET:
7
    config: MY_SECRET_CONFIG_VAR