ke-airflow-dag-generator 0.1.dev1
Python package that implements generating dag from yaml
Requires Python: >=3.9,<3.11

# Airflow Dag Generator

airflow-dag-generator - проект, реализующий возможность создавать шаблонные даги из yaml файлов

## Оглавление
 - [API](#api)
 - [Структура yaml файла](#yaml)
 - [Обертки(Упрощенное описание task)](#handlers)
   - [ClickhouseExecuteOperator](#clickhouse_execute_operator)
   - [PostgresToClickhouseOperatorHandler](#postgres_to_clickhouse_operator_handler)
   - [SQLToFileOperatorHandler](#sql_to_file_operator_handler)
 - [Встроенные macro](#implemented_macro)
 - [Примеры](#examples)
 

## <a name="api">Описание API</a>

`generate` - функция собирающая даг из конфига

Параметры:
 - `config_path` - путь к yaml файлу, в котором лежит конфиг дага

`add_hook_mapping` - функция добавляющая в индекс конкретный хук

Параметры:
 - `hook_type` - задаем тип хука, по которому будем к нему обращаться в yaml
 - `hook_class` - объект класс хука (ClickHouseHook, S3Hook etc)

`add_task_mapping` - функция добавляющая в индекс оператор или таксу, обявлянную с помощью `@task`

Параметры:
 - `type` - задаем тип оператора(такси), по которому будем к нему обращаться в yaml
 - `operator` - объект класс оператора(такси) (ClickhouseExecuteOperator etc)

`add_macro` - функция добавляющая в индекс python-функцию, которую можно будет использовать в yaml-e

Параметры:
 - `name` - задаем имя макро, по которому к нему можно будет обратиться из yaml файла
 - `macro` - python функция, которая будет вызываться в случае указания данного macro в yaml файле

## <a name="yaml">Структура yaml файла</a>

```yaml
<dag_id>:
    description: <description>  # Обязательный параметр
    start_date: <start_date>  # Обязательный параметр
    # Любые дополнительные параметры dag-a
    [<dag_param_name>: <dag_param_value>
    ...]
    tasks:
        <task_id>:
            type: <task_type>  # Обязательный параметр
            description: <task_description>  # Обязательный параметр
            # Любые дополнительные параметры такси (набор зависит от типа)
            [<task_param_name>: <task_param_value>
            ...]
    flow:
        # Задание такси, не зависящий от других таксок
        <task_id_1>: {}
        # Задание такси, не зависящей от других таксок
        <task_id_2>:
            depends_on: [<task_id_2>, ...]
```

**Замечание:** если в конфигурации используются пути к файлам(например путь в sql query) - достаточно указать относительный к директории конфига путь 

## <a name="handlers">Обертки(Упрощенное описание task)</a>

Данный интерфейс позволяет использовать обертки над стандартными операторами, что привод к сокращению кода в yaml файле

### <a name="clickhouse_execute_operator">ClickhouseExecuteOperator</a>

Формат yaml файла для описания ClickhouseExecuteOperator

```yaml
<dag_id>:
  ...
  tasks:
    ...
    clickhouse_exeute:
      type: clickhouse_execute
      description: description
      sql: '<query>'
      connection: '<clickhouse_connection_id>'
    ...
  flow:
    ...
```

Вид `dag.py` файла:

```python
import os

# Важно импортировать функцию `generate` из `airflow_dag_generator.commons` 
from airflow_dag_generator.commons import generate


DAG_DIR = os.path.abspath(os.path.dirname(__file__))
CONFIG_PATH = f"{DAG_DIR}/config.yaml"

dag = generate(CONFIG_PATH)
```

### <a name="postgres_to_clickhouse_operator_handler">PostgresToClickhouseOperatorHandler</a>

### <a name="sql_to_file_operator_handler">SQLToFileOperatorHandler</a>

## <a name="implemented_macro">Встроенные macro</a>

`hook` - позволяем получить объект индексируемого хука по типу

Параметры:
 - `type` - тип хука, который надо получить
 - `**kwargs` - любые параметры, которые надо передать для инициализации хука (например conn_name)

## <a name="api">Примеры</a>

### Пример создания dag-a из BashOperator

`dag.py` файл:

```python
import os

from airflow_dag_generator import generate

DAG_DIR = os.path.abspath(os.path.dirname(__file__))
CONFIG_PATH = f"{DAG_DIR}/config.yaml"

dag = generate(CONFIG_PATH)
```

`config.yaml` файл:

```yaml
example_dag: # dag_id
  description: dag description
  catchup: True
  schedule_interval: '9 2 * * *'
  start_date: '2023-04-01'
  tasks:
   # Определяем таску
   bash_task_1:
     type: bash  # Задаем тип такси
     description: >
       some text
     bash_command: echo 1  # Задаем значение bash_command параметра 
  flow:
    # Задаем флоу
    # Флоу состоит из одной такси, которая не зависит от других тасок
    bash_task_1: {}
```

### Пример создания dag-a с использованием кастомных хуков и операторов

`dag.py` файл:

```python
import os

from airflow_dag_generator import generate, add_hook_mapping, add_task_mapping
from airflow_commons.operators.clickhouse_execute_operator import ClickhouseExecuteOperator
from airflow_commons.hooks.clickhouse_hook import ClickHouseHook

add_hook_mapping('clickhouse', ClickHouseHook)  # добавляем в индекс хук для Clickhouse
add_task_mapping('clickhouse_execute', ClickhouseExecuteOperator)  # добавляем в индекс ClickhouseExecuteOperator

DAG_DIR = os.path.abspath(os.path.dirname(__file__))
CONFIG_PATH = f"{DAG_DIR}/config.yaml"

dag = generate(CONFIG_PATH)
```

`config.yaml` файл:

```yaml
example_dag:
  description: dag description
  catchup: True
  schedule_interval: '9 2 * * *'
  start_date: '2023-04-01'
  tasks:
   dummy_1:
     type: clickhouse_execute
     description: >
       some text
     sql: 'select {{ ds }} as id'
     ch_hook:
       from_macro: hook('clickhouse', clickhouse_conn_id='clickohouse_um_dev_m_poroshin')
   dummy_2:
     type: clickhouse_execute  # используем, добавленный в индекс тип опреатора
     description: >
       some text
     sql: 'select {{ ds }} as id'
     ch_hook:  # Задаем хук с помощью макро `hook` и указываем первым параметром тип добавленного в dag.py хука
       from_macro: hook('clickhouse', clickhouse_conn_id='clickohouse_um_dev_m_poroshin')
  flow:
    dummy_1: {}
    dummy_2:
      depends_on: ['dummy_1']
```