Operators

Note

These operators are included and are intended to represent some of the most common Airflow Operators, but not all Airflow Operators.

Additional Operators can be created by subclassing OrbiterOperator or using OrbiterTask directly.

Review the Astronomer Registry to find additional Airflow Operators.

classDiagram
    direction LR

    class OrbiterBashOperator["orbiter.objects.operators.bash.OrbiterBashOperator"] {
        <<OrbiterOperator>>
            operator = "BashOperator"
            task_id: str
            bash_command: str
    }
    click OrbiterBashOperator href "#orbiter.objects.operators.bash.OrbiterBashOperator" "OrbiterBashOperator Documentation"

    class OrbiterEmailOperator["orbiter.objects.operators.smtp.OrbiterEmailOperator"] {
        <<OrbiterOperator>>
            operator = "EmailOperator"
            task_id: str
            to: str | list[str]
            subject: str
            html_content: str
            files: list | None
            conn_id: str
    }
    click OrbiterEmailOperator href "#orbiter.objects.operators.smtp.OrbiterEmailOperator" "OrbiterEmailOperator Documentation"

    class OrbiterEmptyOperator["orbiter.objects.operators.empty.OrbiterEmptyOperator"] {
        <<OrbiterOperator>>
            operator = "EmptyOperator"
            task_id: str
    }
    click OrbiterEmptyOperator href "#orbiter.objects.operators.empty.OrbiterEmptyOperator" "OrbiterEmptyOperator Documentation"

    class OrbiterKubernetesPodOperator["orbiter.objects.operators.kubernetes_pod.OrbiterKubernetesPodOperator"] {
        <<OrbiterOperator>>
            operator = "KubernetesPodOperator"
            task_id: str
            kubernetes_conn_id: str
            image: str
            cmds: list | None
            arguments: list | None
            env_vars: dict | None
            container_resources: V1ResourceRequirements | None
            image_pull_secrets: list | None
            pod_template_dict: dict | None
    }
    click OrbiterKubernetesPodOperator href "#orbiter.objects.operators.kubernetes_pod.OrbiterKubernetesPodOperator" "OrbiterKubernetesPodOperator Documentation"

    class OrbiterLivyOperator["orbiter.objects.operators.livy.OrbiterLivyOperator"] {
        <<OrbiterOperator>>
            operator = "LivyOperator"
            task_id: str
            livy_conn_id: str
            kwargs: Any
    }
    click OrbiterLivyOperator href "#orbiter.objects.operators.livy.OrbiterLivyOperator" "OrbiterLivyOperator Documentation"

    class OrbiterPythonOperator["orbiter.objects.operators.python.OrbiterPythonOperator"] {
        <<OrbiterOperator>>
            operator = "PythonOperator"
            task_id: str
            python_callable: Callable
            op_args: list | None
            op_kwargs: dict | None
    }
    click OrbiterPythonOperator href "#orbiter.objects.operators.python.OrbiterPythonOperator" "OrbiterPythonOperator Documentation"

    OrbiterPythonOperator "implements" <|-- OrbiterDecoratedPythonOperator
    class OrbiterDecoratedPythonOperator["orbiter.objects.operators.python.OrbiterDecoratedPythonOperator"] {
        <<OrbiterOperator>>
            operator = "PythonOperator"
            task_id: str
            python_callable: Callable
            op_args: list | None
            op_kwargs: dict | None
    }
    click OrbiterDecoratedPythonOperator href "#orbiter.objects.operators.python.OrbiterDecoratedPythonOperator" "OrbiterDecoratedPythonOperator Documentation"

    class OrbiterSQLExecuteQueryOperator["orbiter.objects.operators.sql.OrbiterSQLExecuteQueryOperator"] {
        <<OrbiterOperator>>
            operator = "SQLExecuteQueryOperator"
            task_id: str
            conn_id: str
            sql: str | list[str]
    }
    click OrbiterSQLExecuteQueryOperator href "#orbiter.objects.operators.sql.OrbiterSQLExecuteQueryOperator" "OrbiterSQLExecuteQueryOperator Documentation"

    class OrbiterSSHOperator["orbiter.objects.operators.ssh.OrbiterSSHOperator"] {
        <<OrbiterOperator>>
            operator = "SSHOperator"
            task_id: str
            ssh_conn_id: str
            command: str
            environment: Dict[str, str] | None
    }
    click OrbiterSSHOperator href "#orbiter.objects.operators.ssh.OrbiterSSHOperator" "OrbiterSSHOperator Documentation"

    class OrbiterWinRMOperator["orbiter.objects.operators.win_rm.OrbiterWinRMOperator"] {
        <<OrbiterOperator>>
             operator = "WinRMOperator"
             task_id: str
             ssh_conn_id: str
             command: str
             kwargs: Any
    }
    click OrbiterWinRMOperator href "#orbiter.objects.operators.win_rm.OrbiterWinRMOperator" "OrbiterWinRMOperator Documentation"

orbiter.objects.operators.bash.OrbiterBashOperator ¶

An Airflow BashOperator. Used to run shell commands.

>>> OrbiterBashOperator(task_id="foo", bash_command="echo 'hello world'")
foo_task = BashOperator(task_id='foo', bash_command="echo 'hello world'")

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`bash_command`	`str`	The shell command to execute
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties

orbiter.objects.operators.empty.OrbiterEmptyOperator ¶

An Airflow EmptyOperator. Does nothing.

>>> OrbiterEmptyOperator(task_id="foo")
foo_task = EmptyOperator(task_id='foo')

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties

orbiter.objects.operators.kubernetes_pod.OrbiterKubernetesPodOperator ¶

An Airflow KubernetesPodOperator. Used to launch a Docker container in a Kubernetes cluster.

>>> OrbiterKubernetesPodOperator(
...     task_id="foo",
...     kubernetes_conn_id="KUBERNETES",
...     image="my-docker-image"
... )
foo_task = KubernetesPodOperator(task_id='foo', kubernetes_conn_id='KUBERNETES', image='my-docker-image')
>>> OrbiterKubernetesPodOperator(
...     task_id="bar",
...     pod_template_dict={
...         "apiVersion": "v1", "kind": "Pod",
...         "spec": {"containers": [{"name": "bar", "image": "ubuntu", "command": ["echo", "hello world"]}]}
...     },
... )
bar_task = KubernetesPodOperator(task_id='bar', kubernetes_conn_id='KUBERNETES', pod_template_dict={'apiVersion': 'v1', 'kind': 'Pod', 'spec': {'containers': [{'name': 'bar', 'image': 'ubuntu', 'command': ['echo', 'hello world']}]}})

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`kubernetes_conn_id`	`str, optional`	The Kubernetes connection to use. Defaults to "KUBERNETES"
`image`	`str \| None`	The Docker image to launch, required if not using pod_template_dict
`cmds`	`list[str] \| None, optional`	The commands to run in the container, defaults container Entrypoint
`arguments`	`list[str] \| None, optional`	The arguments to pass to the commands, defaults container commands
`env_vars`	`dict[str, str] \| list[V1EnvVar] \| None, optional`	The environment variables to set in the container, defaults to None
`container_resources`	`V1ResourceRequirements \| None, optional`	The resource requirements for the container, defaults to None
`image_pull_secrets`	`list[V1LocalObjectReference] \| None, optional`	The secrets to use for pulling the Docker image, defaults to None
`pod_template_dict`	`dict \| None, optional`	The yaml (as a python dict) to use for the pod, defaults to None
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties

orbiter.objects.operators.livy.OrbiterLivyOperator ¶

An Airflow LivyOperator. Used to submit Spark Jobs.

>>> from orbiter.objects import conn_id
>>> OrbiterLivyOperator(task_id="foo", **conn_id(conn_id='livy_default', prefix="livy", conn_type='livy'), file="/a/b/c.jar")
foo_task = LivyOperator(task_id='foo', livy_conn_id='livy_default', file='/a/b/c.jar')

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`livy_conn_id`	`str`	The Livy connection to use. (Note: use the `**conn_id(...)` utility function)
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties

orbiter.objects.operators.python.OrbiterPythonOperator ¶

An Airflow PythonOperator. Used to execute any Python Function.

>>> def foo(a, b):
...    print(a + b)
>>> OrbiterPythonOperator(task_id="foo", python_callable=foo)
def foo(a, b):
   print(a + b)
foo_task = PythonOperator(task_id='foo', python_callable=foo)

You can utilize the orbiter_includes and imports to include additional Python code and imports (or subclass to default).

>>> from orbiter.objects.include import OrbiterInclude
>>> OrbiterPythonOperator(
...     task_id="foo",
...     orbiter_includes={OrbiterInclude(filepath="include/bar.py", contents="def baz(): pass")},
...     imports=[OrbiterRequirement(module="include.bar", names=["baz"])],
...     python_callable="baz"
... )
foo_task = PythonOperator(task_id='foo', python_callable=baz)

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`python_callable`	`Callable`	The python function to execute
`op_args`	`list \| None, optional`	The arguments to pass to the python function, defaults to None
`op_kwargs`	`dict \| None, optional`	The keyword arguments to pass to the python function, defaults to None
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties

orbiter.objects.operators.python.OrbiterDecoratedPythonOperator ¶

An Airflow TaskFlow @task. Used to execute any Python Function.

>>> def foo(a, b):
...    print(a + b)
>>> OrbiterDecoratedPythonOperator(task_id="foo", python_callable=foo)
@task()
def foo(a, b):
    print(a + b)

Parameters as per OrbiterPythonOperator

orbiter.objects.operators.smtp.OrbiterEmailOperator ¶

An Airflow EmailOperator. Used to send emails.

>>> OrbiterEmailOperator(
...   task_id="foo", to="humans@astronomer.io", subject="Hello", html_content="World!"
... )
foo_task = EmailOperator(task_id='foo', to='humans@astronomer.io', subject='Hello', html_content='World!', conn_id='SMTP')

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`to`	`str \| list[str]`	The recipient of the email
`subject`	`str`	The subject of the email
`html_content`	`str`	The content of the email
`files`	`list, optional`	The files to attach to the email, defaults to None
`conn_id`	`str, optional`	The SMTP connection to use. Defaults to "SMTP" and sets `orbiter_conns` property
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties

orbiter.objects.operators.sql.OrbiterSQLExecuteQueryOperator ¶

An Airflow Generic SQL Operator. Used to run SQL against any Database.

>>> OrbiterSQLExecuteQueryOperator(
...   task_id="foo", conn_id='sql', sql="select 1;"
... )
foo_task = SQLExecuteQueryOperator(task_id='foo', conn_id='sql', sql='select 1;')

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`conn_id`	`str`	The SQL connection to utilize. (Note: use the `**conn_id(...)` utility function)
`sql`	`str \| list[str]`	The SQL to execute
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties

orbiter.objects.operators.ssh.OrbiterSSHOperator ¶

An Airflow SSHOperator. Used to run shell commands over SSH.

>>> OrbiterSSHOperator(task_id="foo", ssh_conn_id="SSH", command="echo 'hello world'")
foo_task = SSHOperator(task_id='foo', ssh_conn_id='SSH', command="echo 'hello world'")

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`ssh_conn_id`	`str`	The SSH connection to use. (Note: use the `**conn_id(...)` utility function)
`command`	`str`	The command to execute
`environment`	`dict, optional`	The environment variables to set, defaults to None
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties

orbiter.objects.operators.win_rm.OrbiterWinRMOperator ¶

An Airflow WinRMOperator. Used to run commands on remote Window Servers.

>>> from orbiter.objects import conn_id
>>> OrbiterWinRMOperator(task_id="foo", **conn_id(conn_id="winrm_default", prefix="ssh", conn_type="ssh"), command="echo hello")
foo_task = WinRMOperator(task_id='foo', ssh_conn_id='winrm_default', command='echo hello')

Parameters:

Name	Type	Description
`task_id`	`str`	The `task_id` for the operator
`ssh_conn_id`	`str`	The SSH connection to use. (Note: use the `**conn_id(...)` utility function)
`command`	`str`	The command to execute
`**kwargs`		Extra arguments to pass to the operator
`**OrbiterBase`		OrbiterBase inherited properties