### Root Level Setup and Teardown Example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/example_dags/example_setup_teardown.html

This snippet shows a basic Airflow DAG with a root-level setup task, a normal task, and a root-level teardown task. The teardown task is explicitly linked to the setup task using `.as_teardown(setups=root_setup)`.

```python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""Example DAG demonstrating the usage of setup and teardown tasks."""

from __future__ import annotations

import pendulum

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.task_group import TaskGroup

with DAG(
    dag_id="example_setup_teardown",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
) as dag:


[docs]
    root_setup = BashOperator(task_id="root_setup", bash_command="echo 'Hello from root_setup'").as_setup()


    root_normal = BashOperator(task_id="normal", bash_command="echo 'I am just a normal task'")
    root_teardown = BashOperator(
        task_id="root_teardown", bash_command="echo 'Goodbye from root_teardown'"
    ).as_teardown(setups=root_setup)
    root_setup >> root_normal >> root_teardown
    with TaskGroup("section_1") as section_1:
        inner_setup = BashOperator(
            task_id="taskgroup_setup", bash_command="echo 'Hello from taskgroup_setup'"
        ).as_setup()
        inner_normal = BashOperator(task_id="normal", bash_command="echo 'I am just a normal task'")
        inner_teardown = BashOperator(
            task_id="taskgroup_teardown", bash_command="echo 'Hello from taskgroup_teardown'"
        ).as_teardown(setups=inner_setup)
        inner_setup >> inner_normal >> inner_teardown
    root_normal >> section_1

```

--------------------------------

### Example mp_start_method

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/configurations-ref.html

Illustrates a possible value for the multiprocessing start method. This setting directly corresponds to options available in Python's multiprocessing module.

```text
fork
```

--------------------------------

### Updating plugins.rst examples to use pyproject.toml

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/release_notes.html

This example shows how to update plugin examples to use pyproject.toml instead of setup.py.

```toml
[tool.poetry.plugins."airflow.plugins]"
my_plugin = "my_plugin.my_plugin:MyPlugin"
```

--------------------------------

### Setup Without Teardown

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/howto/setup-and-teardown.html

Demonstrates a setup task without an explicit teardown, where all downstream tasks implicitly require the setup task.

```python
create_cluster >> run_query >> other_task
```

--------------------------------

### Run Airflow Standalone

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/start.html

Initializes the database, creates a user, and starts all Airflow components for a standalone setup. Access the UI at localhost:8080.

```bash
airflow standalone
```

--------------------------------

### Example: Create User with Admin Role

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/cli-and-env-variables-ref.html

Example of creating a user with the 'Admin' role and a specific username.

```bash
$ airflow users create 
    
> –username admin –firstname FIRST_NAME –lastname LAST_NAME –role Admin –email admin@example.org
```

--------------------------------

### Install Core Airflow Extras

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/extra-packages-ref.html

Use this command to install core Airflow extras, which add specific Python dependencies to extend Airflow's functionality. Examples include 'aiobotocore' for asynchronous AWS operations or 'graphviz' for DAG visualization.

```bash
pip install 'apache-airflow[aiobotocore]'
```

```bash
pip install 'apache-airflow[async]'
```

```bash
pip install 'apache-airflow[cgroups]'
```

```bash
pip install 'apache-airflow[deprecated-api]'
```

```bash
pip install 'apache-airflow[github-enterprise]'
```

```bash
pip install 'apache-airflow[google-auth]'
```

```bash
pip install 'apache-airflow[graphviz]'
```

```bash
pip install 'apache-airflow[kerberos]'
```

```bash
pip install 'apache-airflow[ldap]'
```

```bash
pip install 'apache-airflow[leveldb]'
```

```bash
pip install 'apache-airflow[otel]'
```

```bash
pip install 'apache-airflow[pandas]'
```

```bash
pip install 'apache-airflow[password]'
```

```bash
pip install 'apache-airflow[pydantic]'
```

```bash
pip install 'apache-airflow[rabbitmq]'
```

```bash
pip install 'apache-airflow[sentry]'
```

```bash
pip install 'apache-airflow[s3fs]'
```

```bash
pip install 'apache-airflow[saml]'
```

```bash
pip install 'apache-airflow[statsd]'
```

```bash
pip install 'apache-airflow[uv]'
```

```bash
pip install 'apache-airflow[virtualenv]'
```

```bash
pip install apache-airflow[cloudpickle]
```

--------------------------------

### Example of Metrics allow_list

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/release_notes.html

Provides a complete example for configuring the metrics allow_list.

```yaml
metrics_allow_list:
  - airflow_dag_run_created_total
  - airflow_dag_run_duration_seconds_bucket
  - airflow_dag_run_duration_seconds_count
  - airflow_dag_run_duration_seconds_sum
  - airflow_dag_run_failed_total
  - airflow_dag_run_running_total
  - airflow_dag_run_success_total
  - airflow_dag_run_total
  - airflow_task_duration_seconds_bucket
  - airflow_task_duration_seconds_count
  - airflow_task_duration_seconds_sum
  - airflow_task_failed_total
  - airflow_task_running_total
  - airflow_task_success_total
  - airflow_task_total
  - airflow_ti_duration_seconds_bucket
  - airflow_ti_duration_seconds_count
  - airflow_ti_duration_seconds_sum
  - airflow_ti_failed_total
  - airflow_ti_running_total
  - airflow_ti_success_total
  - airflow_ti_total
```

--------------------------------

### Import Users JSON Format Example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/cli-and-env-variables-ref.html

Example JSON structure for importing users.

```json
[
    {
        "email": "foo@bar.org",
        "firstname": "Jon",
        "lastname": "Doe",
        "roles": ["Public"],
        "username": "jondoe"
    }
]
```

--------------------------------

### Define Setup and Teardown with Multiple Upstreams

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/howto/setup-and-teardown.html

Illustrates a scenario where a teardown task is associated with multiple setup tasks, and tasks within the setup scope have implicit ALL_SUCCESS constraints.

```python
s1 >> w1 >> w2 >> t1.as_teardown(setups=s1) >> w3
w2 >> w4
```

--------------------------------

### Conveniently Define Teardown with Setups

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/howto/setup-and-teardown.html

Use the `as_teardown` method with the `setups` argument to define a teardown task and its associated setup task in a single line.

```python
create_cluster >> run_query >> delete_cluster.as_teardown(setups=create_cluster)
```

--------------------------------

### Using Setup and Teardown Decorators

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/example_dags/example_setup_teardown_taskflow.html

Defines tasks using `@setup`, `@teardown`, and `@task` decorators. The return value of the setup task is passed to the teardown task and tasks within the `with` block.

```python
    @setup
    def outer_setup():
        print("I am outer_setup")
        return "some cluster id"

    @teardown
    def outer_teardown(cluster_id):
        print("I am outer_teardown")
        print(f"Tearing down cluster: {cluster_id}")

    @task
    def outer_work():
        print("I am just a normal task")

    # by using the decorators, outer_setup and outer_teardown are already marked as setup / teardown
    # now we just need to make sure they are linked directly.  At a low level, what we need
    # to do so is the following::
    #     s = outer_setup()
    #     t = outer_teardown()
    #     s >> t
    #     s >> outer_work() >> t
    # Thus, s and t are linked directly, and outer_work runs in between.  We can take advantage of
    # the fact that we are in taskflow, along with the context manager on teardowns, as follows:
    with outer_teardown(outer_setup()):
        outer_work()

```

--------------------------------

### Fixing sql_alchemy_engine_args config example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/release_notes.html

Corrects the configuration example for sql_alchemy_engine_args.

```python
sql_alchemy_engine_args = {
    "pool_recycle": 1800,
    "echo": True,
    "future": True,
}
```

--------------------------------

### setup and teardown tasks

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/decorators.html

Decorators for defining setup and teardown tasks that can be associated with a task or a task group.

```APIDOC
## Setup and Teardown Tasks

### Description
Decorators for defining setup and teardown tasks that can be associated with a task or a task group. These are useful for managing resources or performing actions before or after a set of tasks.

### Methods
- **setup**: Decorator to define a setup task.
- **teardown**: Decorator to define a teardown task.
```

--------------------------------

### Install Airflow with Pandas dependency

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/release_notes.html

To install Airflow with pandas support, use the '[pandas]' extra. This example shows installation for Python 3.8 and Airflow 2.1.2.

```bash
pip install -U "apache-airflow[pandas]==2.1.2" \
  --constraint https://raw.githubusercontent.com/apache/airflow/constraints-2.1.2/constraints-3.8.txt"

```

--------------------------------

### Get Previous Task Instance Start Date

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/models/taskinstance.html

Retrieves the start date of the previous successful task instance. Handles cases where the previous task instance or its start date might not exist.

```python
def _get_previous_ti_start_date(
    *, task_instance: TaskInstance | TaskInstancePydantic, state: str | None = None, session: Session
) -> pendulum.DateTime | None:
    """
    Return the start date from property previous_ti_success.

    :param task_instance: the task instance
    :param state: If passed, it only take into account instances of a specific state.
    :param session: SQLAlchemy ORM Session
    """
    log.debug("previous_start_date was called")
    prev_ti = task_instance.get_previous_ti(state=state, session=session)
    # prev_ti may not exist and prev_ti.start_date may be None.
    return pendulum.instance(prev_ti.start_date) if prev_ti and prev_ti.start_date else None
```

--------------------------------

### Parallel Setup and Teardown Tasks

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/howto/setup-and-teardown.html

Illustrates how to define and execute setup and teardown tasks in parallel. This is useful for managing resources that can be provisioned or cleaned up concurrently.

```python
(
    [create_cluster, create_bucket]
    >> run_query
    >> [delete_cluster.as_teardown(setups=create_cluster), delete_bucket.as_teardown(setups=create_bucket)]
)

```

--------------------------------

### Example DAG with @task_group Decorator

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/example_dags/example_task_group_decorator.html

This snippet defines an example DAG that utilizes the @task_group decorator to create a reusable task group. It includes tasks for starting and ending the DAG, and a task group that chains several other tasks together. Dependencies are set to run the task group sequentially between the start and end tasks.

```python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""Example DAG demonstrating the usage of the @taskgroup decorator."""

from __future__ import annotations

import pendulum

from airflow.decorators import task, task_group
from airflow.models.dag import DAG


# [START howto_task_group_decorator]
# Creating Tasks
@task


[docs]
def task_start():
    """Empty Task which is First Task of Dag"""
    return "[Task_start]"




@task


[docs]
def task_1(value: int) -> str:
    """Empty Task1"""
    return f"[ Task1 {value} ]"




@task


[docs]
def task_2(value: str) -> str:
    """Empty Task2"""
    return f"[ Task2 {value} ]"




@task


[docs]
def task_3(value: str) -> None:
    """Empty Task3"""
    print(f"[ Task3 {value} ]")




@task


[docs]
def task_end() -> None:
    """Empty Task which is Last Task of Dag"""
    print("[ Task_End  ]")




# Creating TaskGroups
@task_group


[docs]
def task_group_function(value: int) -> None:
    """TaskGroup for grouping related Tasks"""
    task_3(task_2(task_1(value)))




# Executing Tasks and TaskGroups
with DAG(
    dag_id="example_task_group_decorator",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
) as dag:


[docs]
    start_task = task_start()


    end_task = task_end()
    for i in range(5):
        current_task_group = task_group_function(i)
        start_task >> current_task_group >> end_task

# [END howto_task_group_decorator]
```

--------------------------------

### DAG Packaging Structure Example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/core-concepts/dags.html

Illustrates a typical directory structure for packaging multiple DAGs and their dependencies into a single zip file.

```text
my_dag1.py
my_dag2.py
package1/__init__.py
package1/functions.py
```

--------------------------------

### Download Docker Compose and Initialize Airflow

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/tutorial/pipeline.html

Download the docker-compose.yaml file, create necessary directories, set the AIRFLOW_UID environment variable, and initialize the Airflow database using Docker Compose.

```bash
# Download the docker-compose.yaml file
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'

# Make expected directories and set an expected environment variable
mkdir -p ./dags ./logs ./plugins
echo -e "AIRFLOW_UID=$(id -u)" > .env

# Initialize the database
docker compose up airflow-init

# Start up all services
docker compose up

```

--------------------------------

### Get task dependencies blocking task from getting scheduled

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/stable-rest-api-ref.html

Retrieves information about task dependencies that are blocking a specific task from being scheduled. This endpoint is available starting from version 2.10.0.

```APIDOC
## GET /dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}/dependencies

### Description
Get task dependencies blocking task from getting scheduled.

### Method
GET

### Endpoint
/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}/dependencies

### Parameters
#### Path Parameters
- **dag_id** (string) - Required - The DAG ID.
- **dag_run_id** (string) - Required - The DAG run ID.
- **task_id** (string) - Required - The task ID.
- **map_index** (integer) - Required - The map index.

### Responses
#### Success Response (200)
- **dependencies** (array) - A list of dependencies.
  - **name** (string) - The name of the dependency.
  - **reason** (string) - The reason for the dependency.

#### Error Responses
- **400** - Client specified an invalid argument.
- **401** - Request not authenticated due to missing, invalid, authentication info.
- **403** - Client does not have sufficient permission.
- **404** - A specified resource is not found.
```

--------------------------------

### Get Root Tasks in a DAG

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/models/dag.html

Retrieves a list of tasks that have no upstream dependencies, indicating they are the starting points of the DAG.

```python
def roots(self) -> list[Operator]:
    """Return nodes with no parents. These are first to execute and are called roots or root nodes."""
    return [task for task in self.tasks if not task.upstream_list]
```

--------------------------------

### Get Model Data Interval Utility

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/models/dag.html

A utility function to retrieve the data interval (start and end times) from a model instance. It handles potential inconsistencies and raises an exception if the start and end fields are not both present or both absent.

```python
def _get_model_data_interval(
    instance: Any,
    start_field_name: str,
    end_field_name: str,
) -> DataInterval | None:
    start = timezone.coerce_datetime(getattr(instance, start_field_name))
    end = timezone.coerce_datetime(getattr(instance, end_field_name))
    if start is None:
        if end is not None:
            raise InconsistentDataInterval(instance, start_field_name, end_field_name)
        return None
    elif end is None:
        raise InconsistentDataInterval(instance, start_field_name, end_field_name)
    return DataInterval(start, end)
```

--------------------------------

### Configure Plugin Entrypoint in pyproject.toml

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/authoring-and-scheduling/plugins.html

Shows how to register a plugin using setuptools entrypoints in the `pyproject.toml` file for automatic loading by Airflow.

```toml
[project.entry-points.airflow.plugins]
my_plugin = "my_package.my_plugin:MyAirflowPlugin"
```

--------------------------------

### Check Any Scheduler Job Status (HA)

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/cli-and-env-variables-ref.html

Example command to check if any scheduler job is running in a high availability setup.

```bash
airflow jobs check –job-type SchedulerJob –allow-multiple –limit 100
```

--------------------------------

### Initialize Package

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/administration-and-deployment/modules_management.html

Create an __init__.py file to mark the directory as a Python package. This code will execute upon import.

```python
print("Hello from airflow_operators")
```

--------------------------------

### List Resources with Limit and Offset

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/stable-rest-api-ref.html

Demonstrates how to list resources using query parameters to control the number of items fetched and the starting offset.

```http
v1/connections?limit=25&offset=25
```

--------------------------------

### Get Email Backend Configuration

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/howto/email-config.html

Use this command to retrieve the currently configured email backend in Airflow. This helps in verifying your email setup.

```bash
$ airflow config get-value email email_backend
airflow.utils.email.send_email_smtp

```

--------------------------------

### Start New Webserver UI

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/release_notes.html

Run the 'airflow webserver' command to start the new UI. You will be prompted to log in with the admin credentials you created.

```bash
airflow webserver
```

--------------------------------

### Define Basic Setup and Teardown Tasks

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/howto/setup-and-teardown.html

Mark tasks as setup or teardown using `as_setup` and `as_teardown` methods and establish their upstream/downstream relationships.

```python
create_cluster.as_setup() >> run_query >> delete_cluster.as_teardown()
```

```python
create_cluster >> delete_cluster
```

--------------------------------

### Get DAG Response Sample

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/stable-rest-api-ref.html

Example JSON response when retrieving basic information about a specific DAG. This response contains details available in the DAGModel.

```json
{
  "dag_id": "string",
  "dag_display_name": "string",
  "root_dag_id": "string",
  "is_paused": true,
  "is_active": true,
  "is_subdag": true,
  "last_parsed_time": "2019-08-24T14:15:22Z",
  "last_pickled": "2019-08-24T14:15:22Z",
  "last_expired": "2019-08-24T14:15:22Z",
  "scheduler_lock": true,
  "pickle_id": "string",
  "default_view": "string",
  "fileloc": "string",
  "file_token": "string",
  "owners": [
    "string"
],
  "description": "string",
  "schedule_interval": {
    "__type": "string",
    "days": 0,
    "seconds": 0,
    "microseconds": 0
},
  "timetable_description": "string",
  "tags": [
    {
      "name": "string"
}
],
  "max_active_tasks": 0,
  "max_active_runs": 0,
  "has_task_concurrency_limits": true,
  "has_import_errors": true,
  "next_dagrun": "2019-08-24T14:15:22Z",
  "next_dagrun_data_interval_start": "2019-08-24T14:15:22Z",
  "next_dagrun_data_interval_end": "2019-08-24T14:15:22Z",
  "next_dagrun_create_after": "2019-08-24T14:15:22Z",
  "max_consecutive_failed_dag_runs": 0
}
```

--------------------------------

### Use BashOperator with Jinja Templating

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/core-concepts/operators.html

This example shows how to pass the start of a data interval as an environment variable to a Bash script using the BashOperator and Jinja templating.

```python
BashOperator(
    task_id="run_bash_script",
    bash_command="export START_DATE={{ ds }}; /path/to/your/script.sh",
)
```

--------------------------------

### DagModel next_dagrun_data_interval Property

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/models/dag.html

Gets or sets the data interval for the next DAG run. The setter handles None values by clearing interval start and end.

```python
@property
def next_dagrun_data_interval(self) -> DataInterval | None:
    return _get_model_data_interval(
        self,
        "next_dagrun_data_interval_start",
        "next_dagrun_data_interval_end",
    )

@next_dagrun_data_interval.setter
def next_dagrun_data_interval(self, value: tuple[datetime, datetime] | None) -> None:
    if value is None:
        self.next_dagrun_data_interval_start = self.next_dagrun_data_interval_end = None
    else:
        self.next_dagrun_data_interval_start, self.next_dagrun_data_interval_end = value
```

--------------------------------

### Configure Airflow Policy Entrypoint in pyproject.toml

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/administration-and-deployment/cluster-policies.html

Example of configuring a setuptools entrypoint for Airflow policies in a pyproject.toml file. This allows custom policy modules to be discovered.

```toml
[build-system]
requires = ["setuptools", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "my-airflow-plugin"
version = "0.0.1"
# ...

dependencies = ["apache-airflow>=2.6"]
[project.entry-points.'airflow.policy']
_ = 'my_airflow_plugin.policies'

```

--------------------------------

### Example DAG with SLA Configuration

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/core-concepts/tasks.html

Demonstrates how to configure an SLA for a task using `datetime.timedelta` and a custom `sla_miss_callback` within a DAG definition.

```python
def sla_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    print(
        "The callback arguments are: ",
        {
            "dag": dag,
            "task_list": task_list,
            "blocking_task_list": blocking_task_list,
            "slas": slas,
            "blocking_tis": blocking_tis,
        },
    )


@dag(
    schedule="*/2 * * * *",
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    sla_miss_callback=sla_callback,
    default_args={"email": "email@example.com"},
)
def example_sla_dag():
    @task(sla=datetime.timedelta(seconds=10))
    def sleep_20():
        """Sleep for 20 seconds"""
        time.sleep(20)

    @task
    def sleep_30():
        """Sleep for 30 seconds"""
        time.sleep(30)

    sleep_20() >> sleep_30()


example_dag = example_sla_dag()

```

--------------------------------

### Get DagRuns within a Date Range

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/models/dag.html

Retrieves a list of DagRuns for a specific DAG between a start and end date (inclusive). Requires an active database session.

```python
dagruns = session.scalars(
            select(DagRun).where(
                DagRun.dag_id == self.dag_id,
                DagRun.execution_date >= start_date,
                DagRun.execution_date <= end_date,
            )
        ).all()
```

--------------------------------

### Install Airflow with Extras and Providers

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/installation/installing-from-pypi.html

Use this command to install a specific version of Apache Airflow along with extras like 'async', 'postgres', and 'google' providers. It ensures compatibility by using the constraint file for the specified Airflow version and Python version.

```bash
AIRFLOW_VERSION=2.11.0
PYTHON_VERSION="$(python -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')"
CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
pip install "apache-airflow[async,postgres,google]==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"
```

--------------------------------

### Task Dependency Example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/tutorial/pipeline.html

This snippet shows how tasks are defined and their dependencies are set using bitshift operators. It illustrates a linear dependency chain with parallel starting tasks.

```python
[create_employees_table, create_employees_temp_table] >> get_data() >> merge_data()
```

--------------------------------

### Response Sample for GET /eventLogs

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/stable-rest-api-ref.html

Example JSON response for listing event log entries. It includes details for each log entry such as ID, timestamp, DAG/task information, and event type.

```json
{
  "event_logs": [
    {
      "event_log_id": 0,
      "when": "2019-08-24T14:15:22Z",
      "dag_id": "string",
      "task_id": "string",
      "run_id": "string",
      "map_index": 0,
      "try_number": 0,
      "event": "string",
      "execution_date": "2019-08-24T14:15:22Z",
      "owner": "string",
      "extra": "string"
}
],
  "total_entries": 0

}
```

--------------------------------

### DAG Initialization Example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/models/dag.html

Example demonstrating how to initialize a DAG with specific Jinja environment options to control template rendering behavior. This is useful for advanced template customization.

```python
DAG(
    dag_id="my-dag",
    jinja_environment_kwargs={
        "keep_trailing_newline": True,
        # some other jinja2 Environment options here
    },
)
```

--------------------------------

### Get Airflow Variables in Python

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/core-concepts/variables.html

Demonstrates how to retrieve variable values using the Variable model in Python. Includes examples for normal retrieval, auto-deserializing JSON, and providing default values.

```python
from airflow.models import Variable

# Normal call style
foo = Variable.get("foo")

# Auto-deserializes a JSON value
bar = Variable.get("bar", deserialize_json=True)

# Returns the value of default_var (None) if the variable is not set
baz = Variable.get("baz", default_var=None)
```

--------------------------------

### Example Plugin Implementation

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/authoring-and-scheduling/plugins.html

This example demonstrates how to derive from AirflowPlugin to create a custom plugin. It includes a custom hook, macro, and a Flask blueprint.

```python
# This is the class you derive to create a plugin
from airflow.plugins_manager import AirflowPlugin
from airflow.security import permissions
from airflow.www.auth import has_access

from flask import Blueprint
from flask_appbuilder import expose, BaseView as AppBuilderBaseView

# Importing base classes that we need to derive
from airflow.hooks.base import BaseHook
from airflow.providers.amazon.aws.transfers.gcs_to_s3 import GCSToS3Operator


# Will show up in Connections screen in a future version
class PluginHook(BaseHook):
    pass


# Will show up under airflow.macros.test_plugin.plugin_macro
# and in templates through {{ macros.test_plugin.plugin_macro }}
def plugin_macro():
    pass


# Creating a flask blueprint to integrate the templates and static folder
bp = Blueprint(
    "test_plugin",
    __name__,
    template_folder="templates",  # registers airflow/plugins/templates as a Jinja template folder
    static_folder="static",
    static_url_path="/static/test_plugin",
)


```

--------------------------------

### Define TaskFlow API DAG

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/tutorial/taskflow.html

Defines a DAG using the @dag decorator. The function name serves as the DAG ID. This example sets up a simple DAG with no schedule, a start date, and no catchup.

```python
import json

import pendulum

from airflow.decorators import dag, task
@dag(
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
)
def tutorial_taskflow_api():
    """
    ### TaskFlow API Tutorial Documentation
    This is a simple data pipeline example which demonstrates the use of
    the TaskFlow API using three simple tasks for Extract, Transform, and Load.
    Documentation that goes along with the Airflow TaskFlow API tutorial is
    located
    [here](https://airflow.apache.org/docs/apache-airflow/stable/tutorial_taskflow_api.html)
    """
    
```

--------------------------------

### Valid Operator Constructor: No Constructor Defined in Subclass

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/howto/custom-operator.html

This example demonstrates a valid scenario where a subclass inherits from a base operator and does not define its own constructor, relying on the parent's proper templated field setup.

```python
class HelloOperator(BaseOperator):
    template_fields = "field_a"

    def __init__(field_a) -> None:
        self.field_a = field_a


class MyHelloOperator(HelloOperator):
    template_fields = "field_a"


```

--------------------------------

### Import and Use Package

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/administration-and-deployment/modules_management.html

After installation, you can import your custom package. The __init__.py code will execute upon import.

```python
>>> import airflow_operators
Hello from airflow_operators
>>>
```

--------------------------------

### TaskFlow API DAG with Virtualenv Example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/example_dags/tutorial_taskflow_api_virtualenv.html

This DAG demonstrates the TaskFlow API using virtual environments for isolated task execution. It requires the 'virtualenv' package to be installed and includes tasks for extracting, transforming, and loading data.

```python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations

import logging
from datetime import datetime

from airflow.decorators import dag, task
from airflow.operators.python import is_venv_installed




log = logging.getLogger(__name__)



if not is_venv_installed():
    log.warning("The tutorial_taskflow_api_virtualenv example DAG requires virtualenv, please install it.")
else:

    @dag(schedule=None, start_date=datetime(2021, 1, 1), catchup=False, tags=["example"])



    def tutorial_taskflow_api_virtualenv():
        """
        ### TaskFlow API example using virtualenv
        This is a simple data pipeline example which demonstrates the use of
        the TaskFlow API using three simple tasks for Extract, Transform, and Load.
        """

        @task.virtualenv(
            serializer="dill",  # Use `dill` for advanced serialization.
            system_site_packages=False,
            requirements=["funcsigs"],
        )
        def extract():
            """
            #### Extract task
            A simple Extract task to get data ready for the rest of the data
            pipeline. In this case, getting data is simulated by reading from a
            hardcoded JSON string.
            """
            import json

            data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'

            order_data_dict = json.loads(data_string)
            return order_data_dict

        @task(multiple_outputs=True)
        def transform(order_data_dict: dict):
            """
            #### Transform task
            A simple Transform task which takes in the collection of order data and
            computes the total order value.
            """
            total_order_value = 0

            for value in order_data_dict.values():
                total_order_value += value

            return {"total_order_value": total_order_value}

        @task()
        def load(total_order_value: float):
            """
            #### Load task
            A simple Load task which takes in the result of the Transform task and
            instead of saving it to end user review, just prints it out.
            """

            print(f"Total order value is: {total_order_value:.2f}")

        order_data = extract()
        order_summary = transform(order_data)
        load(order_summary["total_order_value"])



    tutorial_dag = tutorial_taskflow_api_virtualenv()

```

--------------------------------

### Basic Dynamic Task Mapping Example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/authoring-and-scheduling/dynamic-task-mapping.html

Demonstrates expanding a task to run multiple times with different inputs. The results are aggregated by a downstream task.

```python
from __future__ import annotations

from datetime import datetime

from airflow.decorators import task
from airflow.models.dag import DAG

with DAG(dag_id="example_dynamic_task_mapping", schedule=None, start_date=datetime(2022, 3, 4)) as dag:

    @task
    def add_one(x: int):
        return x + 1

    @task
    def sum_it(values):
        total = sum(values)
        print(f"Total was {total}")

    added_values = add_one.expand(x=[1, 2, 3])
    sum_it(added_values)
```

--------------------------------

### Install Devel All Extras

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/extra-packages-ref.html

Installs everything needed for development, including Hadoop, all devel extras, and all doc extras, when Airflow is in editable mode. Excludes providers.

```bash
pip install -e '.[devel-all]'
```

--------------------------------

### Get Task Instances with Date Range

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/models/dag.html

Retrieves task instances for a DAG within a specified date range. If no start date is provided, it defaults to 30 days prior to the current UTC time, set to the beginning of that day.

```python
def get_task_instances(
        self,
        start_date: datetime | None = None,
        end_date: datetime | None = None,
        state: list[TaskInstanceState] | None = None,
        session: Session = NEW_SESSION,
    ) -> list[TaskInstance]:
        if not start_date:
            start_date = (timezone.utcnow() - timedelta(30)).replace(
                hour=0, minute=0, second=0, microsecond=0
            )

        query = self._get_task_instances(
            task_ids=None,
            start_date=start_date,
            end_date=end_date,
            state=state or (),
            session=session,
        )
        return session.scalars(cast(Select, query).order_by(DagRun.execution_date)).all()
```

--------------------------------

### Bash Task Building Dynamic Command

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/example_dags/example_bash_decorator.html

This example shows how to construct a dynamic Bash command within a task. It first defines a Python function to get a list of files and then uses `shlex.join` to build a safe command string to execute `stat` on those files.

```python
def _get_files_in_cwd() -> list[str]:
    from pathlib import Path

    dir_contents = Path.cwd().glob("airflow/example_dags/*.py")
    files = [str(elem) for elem in dir_contents if elem.is_file()]

    return files

@task.bash
def get_file_stats() -> str:
    from shlex import join

    files = _get_files_in_cwd()
    cmd = join(["stat", *files])

    return cmd

get_file_stats()
```

--------------------------------

### Initialize MetastoreBackend

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/secrets/metastore.html

Instantiates the MetastoreBackend class. This backend retrieves connection and variable information directly from the Airflow metastore database.

```python
# No instantiation code provided in the source.
```

--------------------------------

### Install Airflow with Constraints

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/start.html

Installs a specific version of Airflow using a constraints file tailored to the installed Python version. This ensures a reproducible installation.

```bash
AIRFLOW_VERSION=2.11.0

# Extract the version of Python you have installed. If you're currently using a Python version that is not supported by Airflow, you may want to set this manually.
# See above for supported versions.
PYTHON_VERSION="$(python -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')"

CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
```

--------------------------------

### Simple Dynamic Task Mapping Example

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/authoring-and-scheduling/dynamic-task-mapping.html

Demonstrates basic dynamic task mapping by expanding a list defined directly in the DAG file using the `expand()` function. This is useful for creating multiple task instances based on a static list.

```python
#
# Licensed to the Apache Software Foundation (ASF) under one

```

--------------------------------

### Start Airflow Webserver

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/cli-and-env-variables-ref.html

Starts the Airflow webserver. Configurable with options for log files, host, port, SSL, workers, and worker class.

```bash
airflow webserver [-h] [-A ACCESS_LOGFILE] [-L ACCESS_LOGFORMAT] [-D] [-d]
                  [-E ERROR_LOGFILE] [-H HOSTNAME] [-l LOG_FILE] [--pid [PID]]
                  [-p PORT] [--ssl-cert SSL_CERT] [--ssl-key SSL_KEY]
                  [--stderr STDERR] [--stdout STDOUT] [-t WORKER_TIMEOUT]
                  [-k {sync,eventlet,gevent,tornado}] [-w WORKERS]
```

--------------------------------

### Install Amazon Web Services Integration

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/release_notes.html

Use this command to install the Amazon Web Services integration. It simplifies the installation of AWS-related extras.

```bash
pip install 'apache-airflow[aws]'
```

--------------------------------

### Task Groups for Parallel Setup and Teardown

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/howto/setup-and-teardown.html

Shows how to visually group parallel setup and teardown tasks using TaskGroups. This improves the readability and organization of complex parallel operations.

```python
with TaskGroup("setup") as tg_s:
    create_cluster = create_cluster()
    create_bucket = create_bucket()
run_query = run_query()
with TaskGroup("teardown") as tg_t:
    delete_cluster = delete_cluster().as_teardown(setups=create_cluster)
    delete_bucket = delete_bucket().as_teardown(setups=create_bucket)
tg_s >> run_query >> tg_t

```

--------------------------------

### Install Sentry Requirement

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/administration-and-deployment/logging-monitoring/errors.html

Install the Sentry requirement for Airflow using pip. This command ensures all necessary packages for Sentry integration are installed.

```bash
pip install 'apache-airflow[sentry]'
```

--------------------------------

### Using ContextWrapper for Setup/Teardown Tasks

Source: https://airflow.apache.org/docs/apache-airflow/2.11.0/_modules/airflow/decorators/setup_teardown.html

The `ContextWrapper` class allows you to manage setup and teardown tasks within a context. Only tasks marked as setup or teardown can be used within this context manager.

```python
class ContextWrapper(list):
    """A list subclass that has a context manager that pushes setup/teardown tasks to the context."""

    def __init__(self, tasks: list[BaseOperator | XComArg]):

        self.tasks = tasks


        super().__init__(tasks)


    def __enter__(self):
        operators = []
        for task in self.tasks:
            if isinstance(task, BaseOperator):
                operators.append(task)
                if not task.is_setup and not task.is_teardown:
                    raise AirflowException("Only setup/teardown tasks can be used as context managers.")
            elif not task.operator.is_setup and not task.operator.is_teardown:
                raise AirflowException("Only setup/teardown tasks can be used as context managers.")
        if not operators:
            # means we have XComArgs
            operators = [task.operator for task in self.tasks]
        SetupTeardownContext.push_setup_teardown_task(operators)
        return SetupTeardownContext




    def __exit__(self, exc_type, exc_val, exc_tb):
        SetupTeardownContext.set_work_task_roots_and_leaves()




context_wrapper = ContextWrapper

```