### Specify Package Installation with Extra Flags - Requirements File

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/howto/operator/python.html

This example shows how to define package dependencies for a virtual environment, including specific versions and custom installation flags like `--pre`, `--index-url`, `--no-index`, and `--find-links`. This allows for precise control over package sources and installation behavior.

```text
SomePackage==0.2.1 --pre --index-url http://some.archives.com/archives

AnotherPackage==1.4.3 --no-index --find-links /my/local/archives

```

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

### Install Airflow with Apache Software Extras

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

Installs apache-airflow with additional dependencies for integration with specific Apache projects. Each 'extra' corresponds to a different Apache project, enabling related operators and hooks. Examples include 'apache.atlas', 'apache.beam', and 'apache.spark'.

```bash
pip install 'apache-airflow[apache.atlas]'
```

```bash
pip install 'apache-airflow[apache.beam]'
```

```bash
pip install 'apache-airflow[apache.cassandra]'
```

```bash
pip install 'apache-airflow[apache.drill]'
```

```bash
pip install 'apache-airflow[apache.druid]'
```

```bash
pip install 'apache-airflow[apache.hdfs]'
```

```bash
pip install 'apache-airflow[apache.hive]'
```

```bash
pip install 'apache-airflow[apache.kylin]'
```

```bash
pip install 'apache-airflow[apache.livy]'
```

```bash
pip install 'apache-airflow[apache.pig]'
```

```bash
pip install 'apache-airflow[apache.pinot]'
```

```bash
pip install 'apache-airflow[apache.spark]'
```

```bash
pip install 'apache-airflow[apache.sqoop]'
```

```bash
pip install 'apache-airflow[apache.webhdfs]'
```

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

### Airflow Logging Configuration File Setup (Shell)

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

This example shows the shell commands and environment setup required for Airflow's logging configuration. It involves placing a `airflow_local_settings.py` file in a directory included in the `PYTHONPATH`, such as `$AIRFLOW_HOME/config`.

```bash
# Example of setting PYTHONPATH if needed
export PYTHONPATH="$AIRFLOW_HOME/config:$PYTHONPATH"

# Copying the default config as a starting point
cp airflow/config_templates/airflow_local_settings.py ${AIRFLOW_HOME}/config/airflow_local_settings.py

# Then edit ${AIRFLOW_HOME}/config/airflow_local_settings.py as needed
```

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

### Install Python Packaging Tools

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/modules_management.html

Installs or upgrades pip, setuptools, and wheel, which are essential for building and distributing Python packages. Ensure you have the latest versions for optimal performance and compatibility.

```bash
pip install --upgrade pip setuptools wheel

```

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

### Compile and Install SQLite from Source

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/howto/set-up-database.html

Downloads the SQLite source code, configures build options with specific flags for enhanced features, compiles, and installs it locally. This process ensures a newer version of SQLite is available.

```bash
wget https://www.sqlite.org/src/tarball/sqlite.tar.gz
tar xzf sqlite.tar.gz
cd sqlite/
export CFLAGS="-DSQLITE_ENABLE_FTS3 \
    -DSQLITE_ENABLE_FTS3_PARENTHESIS \
    -DSQLITE_ENABLE_FTS4 \
    -DSQLITE_ENABLE_FTS5 \
    -DSQLITE_ENABLE_JSON1 \
    -DSQLITE_ENABLE_LOAD_EXTENSION \
    -DSQLITE_ENABLE_RTREE \
    -DSQLITE_ENABLE_STAT4 \
    -DSQLITE_ENABLE_UPDATE_DELETE_LIMIT \
    -DSQLITE_SOUNDEX \
    -DSQLITE_TEMP_STORE=3 \
    -DSQLITE_USE_URI \
    -O2 \
    -fPIC"
export PREFIX="/usr/local"
LIBS="-lm" ./configure --disable-tcl --enable-shared --enable-tempstore=always --prefix="$PREFIX"
make
make install
```

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

### Example Airflow Plugin Implementation

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/plugins.html

Demonstrates how to create a functional Airflow plugin by deriving from `AirflowPlugin`. This example includes a custom hook, macro, Flask blueprint, and an AppBuilder view.

```python
# This is the class you derive to create a plugin
from airflow.plugins_manager import AirflowPlugin

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

# Creating a flask appbuilder BaseView
class TestAppBuilderBaseView(AppBuilderBaseView):
    default_view = "test"

    @expose("/")
    def test(self):
        return self.render_template("test_plugin/test.html", content="Hello galaxy!")


```

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

### Executor Configuration Example for KubernetesExecutor

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

Shows how to configure executor-specific settings, using the KubernetesExecutor as an example. The `executor_config` argument allows customization, such as specifying a Docker image for task execution on Kubernetes.

```python
from airflow.operators.bash import BashOperator

MyOperator(
    ...,
    executor_config={
        "KubernetesExecutor":
            {"image": "myCustomDockerImage"}
    }
)

```

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

### Install Prerequisites for SQLite Upgrade on Amazon Linux

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/howto/set-up-database.html

Installs necessary packages (wget, tar, gzip, gcc, make, expect) required for compiling and installing a newer version of SQLite from source on Amazon Linux.

```bash
yum -y install wget tar gzip gcc make expect
```

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

### Start Airflow Services

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

Starts all the necessary Airflow services defined in the docker-compose.yaml file, including the webserver, scheduler, and worker. This command brings up the complete Airflow environment for use.

```bash
docker-compose up
```

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

### Python Package Structure Example

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/modules_management.html

Illustrates a typical directory structure for Python packages intended for use with Airflow. It includes __init__.py files to define packages and subpackages, and example Python files.

```text
<DIRECTORY ON PYTHONPATH>
| .airflowignore  -- only needed in ``dags`` folder, see below
| -- my_company
              | __init__.py
              | common_package
              |              |  __init__.py
              |              | common_module.py
              |              | subpackage
              |                         | __init__.py
              |                         | subpackaged_util_module.py
              |
              | my_custom_dags
                              | __init__.py
                              | my_dag1.py
                              | my_dag2.py
                              | base_dag.py
```

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

### Install Airflow with Specific Provider Versions and Constraints

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

Installs apache-airflow and specified provider packages (e.g., google, amazon, apache.spark) with exact versions matching a release (e.g., 2.3.4). It uses a constraint file to ensure consistent dependencies compatible with the Airflow version. This is useful for repeatable installations.

```bash
pip install apache-airflow[google,amazon,apache.spark]==2.3.4 \
  --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.3.4/constraints-3.7.txt"
```

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

### List Installed Airflow Providers (Airflow CLI)

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

Lists all installed Airflow providers in the environment. The output can be customized to table, JSON, YAML, or plain text format, and verbose logging is supported.

```bash
airflow providers list [-h] [-o table, json, yaml, plain] [-v]
```

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

### Start Airflow Triggerer Instance

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

Starts a triggerer instance, which is responsible for handling deferred tasks. It allows configuration of capacity, logging, and process management. Options include specifying log files, PID files, and standard error/output streams.

```bash
airflow triggerer [-h] [--capacity CAPACITY] [-D] [-l LOG_FILE] [--pid [PID]]
                  [--stderr STDERR] [--stdout STDOUT]
```

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

### Install Airflow Core via Pip

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

Installs a specific version of Apache Airflow using a constraint file to ensure dependency compatibility. It dynamically detects the Python version to fetch the correct constraint URL.

```bash
AIRFLOW_VERSION=2.3.4
PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-no-providers-${PYTHON_VERSION}.txt"
pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"
```

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

### Install Airflow with optional pandas dependency

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

Provides the command to install Apache Airflow with the pandas extra dependency, which is no longer a core requirement as of version 2.2.0.

```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
```

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

### Logging within an Airflow DAG (Python)

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

Demonstrates how to import Airflow settings and use the logging capabilities within a Python interpreter or a DAG file. This example shows how to log an error message using the task's logger.

```python
from airflow.settings import *

from datetime import datetime
from airflow.models.dag import DAG
from airflow.operators.dummy import DummyOperator

dag = DAG('simple_dag', start_date=datetime(2017, 9, 1))

task = DummyOperator(task_id='task_1', dag=dag)

task.log.error('I want to say something..')

```

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

### Setting and Getting Airflow Variables (Python)

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

Demonstrates how to set an Airflow Variable to an empty string and retrieve it. Previously, retrieving an empty string resulted in None, but this behavior has been corrected to return an empty string.

```python
from airflow.models import Variable

# Set a variable with an empty string
Variable.set("test_key", "")

# Get the variable
value = Variable.get("test_key")

# Previously, 'value' would be None. Now it will be "".
```

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

### Elasticsearch Logging Configuration (INI)

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

Example configuration for enabling remote logging with Elasticsearch, including JSON output to stdout. This ensures log lines contain `log_id` and `offset` fields for easier log reading.

```ini
[logging]
remote_logging = True
[elasticsearch]
host = http://es-host:9200
write_stdout = True
json_format = True
```

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

### Replace airflow.utils.log.logging_mixin.redirect_stderr/stdout with contextlib

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

The `redirect_stderr` and `redirect_stdout` functions from `airflow.utils.log.logging_mixin` have been removed and can be replaced by their counterparts in the standard `contextlib` module. The standard library functions offer more flexibility. The example demonstrates redirecting stdout and stderr to a logger using both the old and new methods.

```python
import logging

from airflow.utils.log.logging_mixin import redirect_stderr, redirect_stdout

logger = logging.getLogger("custom-logger")
with redirect_stdout(logger, logging.INFO), redirect_stderr(logger, logging.WARN):
    print("I love Airflow")
```

```python
from contextlib import redirect_stdout, redirect_stderr
import logging

from airflow.utils.log.logging_mixin import StreamLogWriter

logger = logging.getLogger("custom-logger")

with redirect_stdout(StreamLogWriter(logger, logging.INFO)), redirect_stderr(
    StreamLogWriter(logger, logging.WARN)
):
    print("I Love Airflow")
```

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

### Calling get_task_instances After Signature Change (Python)

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

Provides examples of how to correctly call the `get_task_instances` method on a DAG object after its signature has been updated in Apache Airflow. It shows two scenarios: relying on the `@provide_session` decorator for automatic session management, and explicitly providing a session object.

```python
# if you can rely on @provide_session
dag.get_task_instances()

# if you need to provide the session
dag.get_task_instances(session=your_session)

```

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

### Initialize Airflow with Docker Compose

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

Downloads the official Airflow docker-compose configuration, sets up local directories, and initializes the database services.

```bash
# Download the docker-compose.yaml file
curl -Lf0 '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 /providers

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

Retrieves a list of all installed providers in the Airflow environment.

```APIDOC
## GET /providers

### Description
Returns a list of installed provider packages and their versions.

### Method
GET

### Endpoint
/providers

### Response
#### Success Response (200)
- **providers** (array) - List of provider objects
- **total_entries** (integer) - Total count of providers

#### Response Example
{
  "providers": [
    {
      "package_name": "apache-airflow-providers-google",
      "description": "Google Cloud provider",
      "version": "5.0.0"
    }
  ],
  "total_entries": 1
}
```

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

### Get Previous Start Date (Python)

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

Retrieves the start date of the previous task instance, optionally filtered by state. It logs the call and uses `get_previous_ti` to find the relevant task instance, returning its start date as a pendulum.DateTime object. Handles cases where the previous task instance or its start date might not exist.

```python
@provide_session
def get_previous_start_date(
    self, state: Optional[DagRunState] = None, session: Session = NEW_SESSION
) -> Optional[pendulum.DateTime]:
    """
    The start date from property previous_ti_success.

    :param state: If passed, it only take into account instances of a specific state.
    :param session: SQLAlchemy ORM Session
    """
    self.log.debug("previous_start_date was called")
    prev_ti = self.get_previous_ti(state=state, session=session)
    # prev_ti may not exist and prev_ti.start_date may be None.

```

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

### Package Airflow Plugins with Setuptools Entry Points

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/plugins.html

Shows how to define a plugin using a Flask Blueprint and register it as a Python package using setuptools entry points for automatic discovery by Airflow.

```python
from airflow.plugins_manager import AirflowPlugin
from flask import Blueprint

bp = Blueprint(
    "test_plugin",
    __name__,
    template_folder="templates",
    static_folder="static",
    static_url_path="/static/test_plugin",
)

class MyAirflowPlugin(AirflowPlugin):
    name = "my_namespace"
    flask_blueprints = [bp]

# setup.py entry point
from setuptools import setup
setup(
    name="my-package",
    entry_points={"airflow.plugins": ["my_plugin = my_package.my_plugin:MyAirflowPlugin"]},
)
```

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

### Initialize DAG with Custom Timetable

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/howto/timetable.html

Demonstrates how to instantiate a DAG using a custom parameterized timetable.

```python
with DAG(
    timetable=SometimeAfterWorkdayTimetable(Time(8)),
    ...,
) as dag:
    ...
```

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

### Get Variable with None Default Value in Airflow

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

Demonstrates how to use `None` as a default value when retrieving Airflow variables using `Variable.get()`. This allows for checking if a variable exists and handling its absence gracefully. If a `KeyError` is expected, the `default_var` argument should not be provided.

```python
foo = Variable.get("foo", default_var=None)
if foo is None:
    handle_missing_foo()
```

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

### Initialize and Configure Airflow Environment

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

Commands to initialize the metadata database, create an administrative user, and start the webserver and scheduler components manually.

```bash
airflow db init

airflow users create \
    --username admin \
    --firstname Peter \
    --lastname Parker \
    --role Admin \
    --email spiderman@superhero.org

airflow webserver --port 8080

airflow scheduler
```

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

### Get Pandas DataFrame

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/_modules/airflow/hooks/dbapi.html

Executes a SQL query and returns the results as a Pandas DataFrame. It requires the 'pandas' library to be installed and handles potential import errors.

```python
def get_pandas_df(self, sql, parameters=None, **kwargs):
    """Executes the sql and returns a pandas dataframe

    :param sql: the sql statement to be executed (str) or a list of
        sql statements to execute
    :param parameters: The parameters to render the SQL query with.
    :param kwargs: (optional) passed into pandas.io.sql.read_sql method
    """
    try:
        from pandas.io import sql as psql
    except ImportError:
        raise Exception("pandas library not installed, run: pip install 'apache-airflow[pandas]'.")

    with closing(self.get_conn()) as conn:
        return psql.read_sql(sql, con=conn, params=parameters, **kwargs)
```

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

### Airflow Import Path Changes (2.0+)

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

Details the reorganization of Airflow's import paths starting from version 2.0. Integrations with third-party services are moved to the `airflow.providers` package. While backward compatible, using old paths will raise deprecation warnings. The `_operator` suffix has also been removed from operator names.

```python
from airflow.hooks.base_hook.BaseHook import BaseHook
```

```python
from airflow.hooks.base.BaseHook import BaseHook
```

```python
from airflow.operators.dummy_operator.DummyOperator import DummyOperator
```

```python
from airflow.operators.dummy.DummyOperator import DummyOperator
```

```python
from airflow.operators.trigger_dagrun_operator.TriggerDagRunOperator import TriggerDagRunOperator
```

```python
from airflow.operators.trigger_dagrun.TriggerDagRunOperator import TriggerDagRunOperator
```

```python
from airflow.operators.branch_operator.BaseBranchOperator import BaseBranchOperator
```

```python
from airflow.operators.branch.BaseBranchOperator import BaseBranchOperator
```

```python
from airflow.operators.subdag_operator.SubDagOperator import SubDagOperator
```

```python
from airflow.operators.subdag.SubDagOperator import SubDagOperator
```

```python
from airflow.sensors.base_sensor_operator.BaseSensorOperator import BaseSensorOperator
```

```python
from airflow.sensors.base.BaseSensorOperator import BaseSensorOperator
```

```python
from airflow.sensors.date_time_sensor.DateTimeSensor import DateTimeSensor
```

```python
from airflow.sensors.date_time.DateTimeSensor import DateTimeSensor
```

```python
from airflow.sensors.external_task_sensor.ExternalTaskMarker import ExternalTaskMarker
```

```python
from airflow.sensors.external_task.ExternalTaskMarker import ExternalTaskMarker
```

```python
from airflow.sensors.external_task_sensor.ExternalTaskSensor import ExternalTaskSensor
```

```python
from airflow.sensors.external_task.ExternalTaskSensor import ExternalTaskSensor
```

```python
from airflow.sensors.sql_sensor.SqlSensor import SqlSensor
```

```python
from airflow.sensors.sql.SqlSensor import SqlSensor
```

```python
from airflow.sensors.time_delta_sensor.TimeDeltaSensor import TimeDeltaSensor
```

```python
from airflow.sensors.time_delta.TimeDeltaSensor import TimeDeltaSensor
```

```python
from airflow.contrib.sensors.weekday_sensor.DayOfWeekSensor import DayOfWeekSensor
```

```python
from airflow.sensors.weekday.DayOfWeekSensor import DayOfWeekSensor
```

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

### Get On Execute Callback Property (Python)

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/_modules/airflow/models/mappedoperator.html

Retrieves the callback function to be executed when the task starts running. This allows for custom actions to be performed at the beginning of task execution.

```python
@property
def on_execute_callback(self) -> Optional[TaskStateChangeCallback]:
    return self.partial_kwargs.get("on_execute_callback")
```

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

### Get Airflow Database Connection URI (Bash)

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/howto/set-up-database.html

This command retrieves the current value of the `sql_alchemy_conn` configuration option for Airflow. It shows the default SQLite connection string. This is useful for verifying your database connection setup.

```bash
$ airflow config get-value database sql_alchemy_conn
sqlite:////tmp/airflow/airflow.db

```

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

### Define Python Package Metadata

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/modules_management.html

Creates the `setup.py` file, which contains metadata about your package, such as its name and the packages to include. This file is used by setuptools to build and distribute your package.

```python
import setuptools

setuptools.setup(
    name="airflow_operators",
    packages=setuptools.find_packages(),
)

```

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

### Get Root Tasks (Python)

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

Retrieves a list of root tasks within a DAG. Root tasks are defined as tasks that have no upstream dependencies, meaning they are the starting points of the DAG execution flow. This property is useful for understanding the entry points of a DAG.

```python
@property
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]
```

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

### Python DAG: Passing Parameters via Test Command

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/_modules/airflow/example_dags/example_passing_params_via_test_command.html

Demonstrates how to pass parameters to tasks within an Airflow DAG using the `airflow tasks test` command. It shows how to access these parameters in both PythonOperators and BashOperators, including using environment variables. This example requires Airflow to be installed.

```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 params arguments in templated arguments."""

import datetime
import os
from textwrap import dedent

import pendulum

from airflow import DAG
from airflow.decorators import task
from airflow.operators.bash import BashOperator


@task(task_id="run_this")
def my_py_command(params, test_mode=None, task=None):
    """
    Print out the "foo" param passed in via
    `airflow tasks test example_passing_params_via_test_command run_this <date>
    -t '{"foo":"bar"}'`
    """
    if test_mode:
        print(
            f" 'foo' was passed in via test={test_mode} command : kwargs[params][foo] = {task.params['foo']}"
        )
    # Print out the value of "miff", passed in below via the Python Operator
    print(f" 'miff' was passed in via task params = {params['miff']}")
    return 1



@task(task_id="env_var_test_task")
def print_env_vars(test_mode=None):
    """
    Print out the "foo" param passed in via
    `airflow tasks test example_passing_params_via_test_command env_var_test_task <date>
    --env-vars '{"foo":"bar"}'`
    """
    if test_mode:
        print(f"foo={os.environ.get('foo')}")
        print(f"AIRFLOW_TEST_MODE={os.environ.get('AIRFLOW_TEST_MODE')}")



with DAG(
    "example_passing_params_via_test_command",
    schedule_interval='*/1 * * * *',
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    dagrun_timeout=datetime.timedelta(minutes=4),
    tags=['example'],
) as dag:

    run_this = my_py_command(params={"miff": "agg"})


    my_command = dedent(
        """
        echo "'foo' was passed in via Airflow CLI Test command with value '$FOO'"
        echo "'miff' was passed in via BashOperator with value '$MIFF'"
        """
    )

    also_run_this = BashOperator(
        task_id='also_run_this',
        bash_command=my_command,
        params={"miff": "agg"},
        env={"FOO": "{{ params.foo }}", "MIFF": "{{ params.miff }}"},
    )

    env_var_test_task = print_env_vars()

    run_this >> also_run_this

```

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

### Airflow Config CLI: Get and List Configuration

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/cli-ref.html

This section details how to interact with Airflow's configuration settings using the CLI. It covers retrieving specific configuration values by section and option, and listing all available configuration options. No external dependencies are required beyond the Airflow installation.

```bash
airflow config get-value <section> <option>
airflow config list [--color {auto,on,off}]
```

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

### Airflow CLI: List Configuration Options

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

Lists all available configuration options within Airflow, optionally with colored output. This command helps users discover available settings and their default values.

```bash
airflow config list [-h] [--color {auto,on,off}]
```

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

### Display Airflow Cheat Sheet

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

Displays a quick reference guide for Airflow commands with an optional verbose mode.

```bash
airflow cheat-sheet [-h] [-v]
```

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

### Airflow DAG with Custom Skip Operator (Python)

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/_modules/airflow/example_dags/example_skip_dag.html

This Python code defines an example Airflow DAG that demonstrates the use of a custom `EmptySkipOperator` which always skips its task. It also showcases the standard `EmptyOperator` and how to use different `TriggerRule` settings for task dependencies. The DAG is configured with a start date and set to not catch up.

```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 (
# "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 EmptyOperator and a custom EmptySkipOperator which skips by default."""

import pendulum

from airflow import DAG
from airflow.exceptions import AirflowSkipException
from airflow.operators.empty import EmptyOperator
from airflow.utils.context import Context
from airflow.utils.trigger_rule import TriggerRule


# Create some placeholder operators

class EmptySkipOperator(EmptyOperator):
    """Empty operator which always skips the task."""


    ui_color = '#e8b7e4'


    def execute(self, context: Context):
        raise AirflowSkipException



def create_test_pipeline(suffix, trigger_rule):
    """
    Instantiate a number of operators for the given DAG.

    :param str suffix: Suffix to append to the operator task_ids
    :param str trigger_rule: TriggerRule for the join task
    :param DAG dag_: The DAG to run the operators on
    """
    skip_operator = EmptySkipOperator(task_id=f'skip_operator_{suffix}')
    always_true = EmptyOperator(task_id=f'always_true_{suffix}')
    join = EmptyOperator(task_id=trigger_rule, trigger_rule=trigger_rule)
    final = EmptyOperator(task_id=f'final_{suffix}')

    skip_operator >> join
    always_true >> join
    join >> final


with DAG(
    dag_id='example_skip_dag',
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=['example'],
) as dag:
    create_test_pipeline('1', TriggerRule.ALL_SUCCESS)
    create_test_pipeline('2', TriggerRule.ONE_SUCCESS)

```

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

### Get DAG Runs Between Dates (Python)

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

Fetches a list of DAG runs that fall within a specified date range (inclusive). This method queries the DagRun model for runs associated with the current DAG ID and filters them based on the provided start and end execution dates. It returns a list of DagRun objects.

```python
@provide_session
def get_dagruns_between(self, start_date, end_date, session=NEW_SESSION):
        """
        Returns the list of dag runs between start_date (inclusive) and end_date (inclusive).

        :param start_date: The starting execution date of the DagRun to find.
        :param end_date: The ending execution date of the DagRun to find.
        :param session:
        :return: The list of DagRuns found.
        """
        dagruns = (
            session.query(DagRun)
            .filter(
                DagRun.dag_id == self.dag_id,
                DagRun.execution_date >= start_date,
                DagRun.execution_date <= end_date,
            )
            .all()
        )

        return dagruns
```

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

### Install Docker Provider Package for Airflow

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/upgrading-from-1-10/index.html

Installs the Docker provider package for Airflow. This can be done individually or as part of the Airflow extras installation. The backport provider packages are compatible with Airflow 1.10.

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

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

### Define an Airflow DAG with BashOperator Tasks

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

Demonstrates how to instantiate a DAG with default arguments and define tasks using the BashOperator. Includes examples of task documentation and Jinja templating for dynamic bash commands.

```python
from airflow.operators.bash import BashOperator
from airflow import DAG
from datetime import timedelta, datetime
from textwrap import dedent

with DAG(
    'tutorial',
    default_args={'retries': 1, 'retry_delay': timedelta(minutes=5)},
    schedule_interval=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False
) as dag:

    t1 = BashOperator(task_id='print_date', bash_command='date')
    t2 = BashOperator(task_id='sleep', bash_command='sleep 5', retries=3)
    
    templated_command = dedent("{% for i in range(5) %}\n    echo \"{{ ds }}\"\n{% endfor %}")
    t3 = BashOperator(task_id='templated', bash_command=templated_command)

    t1 >> [t2, t3]
```

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

### Install Airflow Upgrade Check Package via pip

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/upgrading-from-1-10/upgrade-check.html

Installs the official Apache Airflow upgrade check utility. This package is required to run diagnostic checks before migrating from Airflow 1.10 to 2.0.

```bash
pip install apache-airflow-upgrade-check
```

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

### SimpleTaskInstance

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/_api/airflow/models/taskinstance/index.html

Documentation for the SimpleTaskInstance class, a lightweight representation of a TaskInstance.

```APIDOC
## SimpleTaskInstance Object

### Description
The `SimpleTaskInstance` class provides a simplified representation of a `TaskInstance`, often used for serialization or when full object details are not required.

### Methods

*   **`__eq__(self, other)`**: Checks if two `SimpleTaskInstance` objects are equal.
*   **`as_dict(self)`**: Converts the `SimpleTaskInstance` object to a dictionary.
*   **`from_ti(cls, ti)`**: Creates a `SimpleTaskInstance` from a `TaskInstance` object.
*   **`from_dict(cls, data)`**: Creates a `SimpleTaskInstance` from a dictionary.
```

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

### Uninstall Python Package

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/modules_management.html

Removes the installed Python package from your environment using pip. This is useful for cleaning up or managing installed packages.

```bash
pip uninstall airflow_operators

```

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

### Add Documentation to DAGs and Tasks

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

Shows how to attach markdown documentation to DAGs and individual tasks, which is then rendered in the Airflow UI for better observability.

```python
t1.doc_md = "#### Task Documentation\nThis is a markdown-formatted task description."
dag.doc_md = "This is a documentation placed anywhere for the DAG."
```

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

### Start Triggerer Instance

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/cli-ref.html

Starts a triggerer instance, which is responsible for running deferred tasks.

```APIDOC
## airflow triggerer

### Description
Start a triggerer instance.

### Method
CLI Command

### Endpoint
N/A (CLI Command)

### Parameters
#### Named Arguments
- **-h** (boolean) - Optional - Show help message and exit
- **--capacity** (integer) - Optional - The capacity of the triggerer
- **-D** (boolean) - Optional - Run in daemon mode
- **-l, --log-file** (string) - Optional - Log file location
- **--pid** (string) - Optional - PID file location
- **--stderr** (string) - Optional - Standard error file location
- **--stdout** (string) - Optional - Standard output file location

### Request Example
```bash
airflow triggerer --capacity 10 --log-file /var/log/airflow/triggerer.log
```

### Response
#### Success Response (0)
- Triggerer instance started successfully.

#### Response Example
(No direct output, runs as a background process)
```

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

### Initialize Python Package

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/modules_management.html

Creates the `__init__.py` file within the package directory. This file signifies that the directory should be treated as a Python package and can optionally contain initialization code.

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

```

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

### Install Python Package

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/modules_management.html

Installs the built Python wheel package using pip. This makes your package available for import and use in your Python environment.

```bash
pip install dist/airflow_operators-0.0.0-py3-none-any.whl

```

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

### Install StatsD dependencies for Airflow

Source: https://airflow.apache.org/docs/apache-airflow/2.3.4/logging-monitoring/metrics.html

Installs the necessary Python package to enable StatsD support within the Airflow environment.

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

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

### Instantiate an Airflow DAG with Default Arguments (Python)

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

This code demonstrates how to instantiate an Airflow DAG object. It specifies a unique `dag_id`, passes a dictionary of `default_args`, sets a `schedule_interval`, defines a `start_date`, and includes optional parameters like `description`, `catchup`, and `tags`. The `with DAG(...) as dag:` syntax ensures proper context management.

```python
with DAG(
    'tutorial',
    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args={
        'depends_on_past': False,
        'email': ['airflow@example.com'],
        'email_on_failure': False,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'sla': timedelta(hours=2),
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function,
        # 'on_success_callback': some_other_function,
        # 'on_retry_callback': another_function,
        # 'sla_miss_callback': yet_another_function,
        # 'trigger_rule': 'all_success'
    },
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=['example'],
) as dag:
```