Repository: getindata/dbt-airflow-factory (language: python)

Issue:
Release 0.18.0
Bump version and CHANGELOG for next release.


Discussion / hints:

Fix patch:
diff --git a/.codeclimate.yml b/.codeclimate.yml
index d84f5fc..65103fa 100644
--- a/.codeclimate.yml
+++ b/.codeclimate.yml
@@ -1,3 +1,52 @@
+version: "2"         # required to adjust maintainability checks
+
+checks:
+  argument-count:
+    enabled: true
+    config:
+      threshold: 4
+  complex-logic:
+    enabled: true
+    config:
+      threshold: 4
+  file-lines:
+    enabled: true
+    config:
+      threshold: 250
+  method-complexity:
+    enabled: true
+    config:
+      threshold: 5
+  method-count:
+    enabled: true
+    config:
+      threshold: 20
+  method-lines:
+    enabled: true
+    config:
+      threshold: 25
+  nested-control-flow:
+    enabled: true
+    config:
+      threshold: 4
+  return-statements:
+    enabled: true
+    config:
+      threshold: 4
+  similar-code:
+    enabled: true
+    config:
+      threshold: #language-specific defaults. overrides affect all languages.
+  identical-code:
+    enabled: true
+    config:
+      threshold: #language-specific defaults. overrides affect all languages.
+
 plugins:
   pylint:
-    enabled: true
\ No newline at end of file
+    enabled: true
+    checks:
+      import-error:
+        enabled: false
+      bad-continuation:
+        enabled: false
diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md
index 4ab52dd..2e59a6e 100644
--- a/.github/pull_request_template.md
+++ b/.github/pull_request_template.md
@@ -5,4 +5,4 @@ Resolves `<issue nr here>`
 ---
 Keep in mind:
 - [ ] Documentation updates
-- [ ] [Changelog](CHANGELOG.md) updates 
\ No newline at end of file
+- [ ] [Changelog](CHANGELOG.md) updates
\ No newline at end of file
diff --git a/.github/workflows/prepare-release.yml b/.github/workflows/prepare-release.yml
index c6a662a..90ffbbd 100644
--- a/.github/workflows/prepare-release.yml
+++ b/.github/workflows/prepare-release.yml
@@ -15,7 +15,7 @@ jobs:
       matrix:
         python-version: [3.9]
     env:
-      PYTHON_PACKAGE: dbt_airflow_manifest_parser
+      PYTHON_PACKAGE: dbt_airflow_factory
     steps:
       - uses: actions/checkout@v2
       - name: Set up Python ${{ matrix.python-version }}
diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml
index 7b85b1b..ba2865c 100644
--- a/.github/workflows/publish.yml
+++ b/.github/workflows/publish.yml
@@ -12,7 +12,7 @@ jobs:
       matrix:
         python-version: [3.9]
     env:
-      PYTHON_PACKAGE: dbt_airflow_manifest_parser
+      PYTHON_PACKAGE: dbt_airflow_factory
     steps:
       - name: Checkout the repo
         uses: actions/checkout@v2
diff --git a/.gitignore b/.gitignore
index dac0ab2..b4cd6f6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,6 +5,7 @@ dist/
 dbt_modules/
 logs/
 dbt_airflow_manifest_parser.egg-info
+dbt_airflow_factory.egg-info
 .idea
 
 .user.yml
@@ -13,4 +14,6 @@ dbt_airflow_manifest_parser.egg-info
 coverage.xml
 .tox
 
-*.iml
\ No newline at end of file
+*.iml
+
+docs/_build
diff --git a/.isort.cfg b/.isort.cfg
index e1cfc44..767cf73 100644
--- a/.isort.cfg
+++ b/.isort.cfg
@@ -1,9 +1,7 @@
-[tool:isort]
-include_trailing_comma = True
-multi_line_output = 3
-line_length = 120
-known_first_party = dbt_airflow_manifest_parser
+[settings]
+profile = black
+known_first_party = dbt_airflow_factory
 default_section = THIRDPARTY
 
 [settings]
-known_third_party = airflow,kubernetes,pytimeparse,setuptools,yaml
\ No newline at end of file
+known_third_party = airflow,jinja2,pytimeparse,setuptools,yaml
\ No newline at end of file
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 9c2f3ec..4355cbf 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1,18 +1,42 @@
 repos:
+
   - repo: https://github.com/asottile/seed-isort-config
     rev: v2.2.0
     hooks:
       - id: seed-isort-config
+
   - repo: https://github.com/pycqa/isort
-    rev: 5.6.4
+    rev: 5.10.1
     hooks:
       - id: isort
         args: ["--profile", "black", "--filter-files"]
+
   - repo: https://github.com/psf/black
-    rev: 21.9b0
+    rev: 21.12b0
     hooks:
       - id: black
+
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v2.3.0
+    rev: v4.1.0
     hooks:
-      - id: flake8
\ No newline at end of file
+      - id: trailing-whitespace
+        args: [ --markdown-linebreak-ext=md ]
+      - id: check-merge-conflict
+      - id: debug-statements
+
+  - repo: https://github.com/pycqa/flake8
+    rev: 4.0.1
+    hooks:
+      - id: flake8
+        additional_dependencies: [
+            'flake8-blind-except',
+            'flake8-comprehensions',
+            'flake8-pep3101',
+        ]
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v0.931
+    hooks:
+      - id: mypy
+        additional_dependencies:
+          - 'types-PyYAML'
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 553e632..0a084b0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,24 @@
+# Changelog
+
 ## [Unreleased]
 
+## [0.18.0] - 2022-01-14
+
+### Added
+
+-   ReadTheDocs documentation in `docs` directory.
+-   `{{ var.value.VARIABLE_NAME }}` gets replaced with Airflow variables when parsing `airflow.yml` file.
+
+### Changed
+
+-   Rename project from `dbt-airflow-manifest-parser` to `dbt-airflow-factory`.
+
+### Fixed
+
+-   `KubernetesExecutionParameters.env_vars` works in Airflow 1 too. Airflow 1 is expecting a real dictionary of
+    environment variables instead of a list of `k8s.V1EnvVar` objects.
+-   Fix `DummyOperator` import in `operator.py` to work in Airflow 1.
+
 ## [0.17.0] - 2022-01-11
 
 ### Changed
@@ -56,9 +75,11 @@
 
 ## [0.2.0] - 2021-10-25
 
--   Initial implementation of `dbt_airflow_manifest_parser` library. 
+-   Initial implementation of `dbt_airflow_manifest_parser` library.
+
+[Unreleased]: https://github.com/getindata/dbt-airflow-factory/compare/0.18.0...HEAD
 
-[Unreleased]: https://github.com/getindata/dbt-airflow-manifest-parser/compare/0.17.0...HEAD
+[0.18.0]: https://github.com/getindata/dbt-airflow-factory/compare/0.17.0...0.18.0
 
 [0.17.0]: https://github.com/getindata/dbt-airflow-manifest-parser/compare/0.16.0...0.17.0
 
diff --git a/README.md b/README.md
index 54a247a..0311ecd 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,67 @@
-# DBT Airflow Manifest Parser
-
-## About
+# DBT Airflow Factory
 
 Library to convert DBT manifest metadata to Airflow tasks
+
+## Installation
+
+Use the package manager [pip][pip] to install the library:
+
+```bash
+pip install dbt-airflow-factory
+```
+
+## Usage
+
+The library is expected to be used inside an Airflow environment with a Kubernetes image referencing **dbt**.
+
+**dbt-airflow-factory**'s main task is to parse `manifest.json` and create Airflow DAG out of it. It also reads config
+files from `config` directory and therefore is highly customizable (e.g., user can set path to `manifest.json`).
+
+To start, create a directory with a following structure, where `manifest.json` is a file generated by **dbt**:
+```
+.
+├── config
+│   ├── base
+│   │   ├── airflow.yml
+│   │   ├── dbt.yml
+│   │   └── k8s.yml
+│   └── dev
+│       └── dbt.yml
+├── dag.py
+└── manifest.json
+```
+
+Then, put the following code into `dag.py`:
+```python
+from dbt_airflow_factory.airflow_dag_factory import AirflowDagFactory
+from os import path
+
+dag = AirflowDagFactory(path.dirname(path.abspath(__file__)), "dev").create()
+```
+
+When uploaded to Airflow DAGs directory, it will get picked up by Airflow, parse `manifest.json` and prepare a DAG to run.
+
+### Configuration files
+
+It is best to look up the example configuration files in [tests directory][tests] to get a glimpse of correct configs.
+
+You can use [Airflow template variables][airflow-vars] in your `dbt.yml` and `k8s.yml` files, as long as they are inside
+quotation marks:
+```yaml
+target: "{{ var.value.env }}"
+some_other_field: "{{ ds_nodash }}"
+```
+
+Analogously, you can use `"{{ var.value.VARIABLE_NAME }}"` in `airflow.yml`, but only the Airflow variable getter.
+Any other Airflow template variables will not work in `airflow.yml`.
+
+### Creation of the directory with data-pipelines-cli
+
+**DBT Airflow Factory** works best in tandem with [data-pipelines-cli][dp-cli] tool. **dp** not only prepares directory
+for the library to digest, but also automates Docker image building and pushes generated directory to the cloud storage
+of your choice.
+
+[airflow-vars]: https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#variables
+[dp-cli]: https://pypi.org/project/data-pipelines-cli/
+[pip]: https://pip.pypa.io/en/stable/
+[tests]: https://github.com/getindata/dbt-airflow-factory/tree/develop/tests/config
diff --git a/dbt_airflow_factory/__init__.py b/dbt_airflow_factory/__init__.py
new file mode 100644
index 0000000..561d56e
--- /dev/null
+++ b/dbt_airflow_factory/__init__.py
@@ -0,0 +1,1 @@
+version = "0.18.0"
diff --git a/dbt_airflow_factory/airflow_dag_factory.py b/dbt_airflow_factory/airflow_dag_factory.py
new file mode 100644
index 0000000..488ae0e
--- /dev/null
+++ b/dbt_airflow_factory/airflow_dag_factory.py
@@ -0,0 +1,117 @@
+"""Factory creating Airflow DAG."""
+
+import os
+
+import airflow
+from airflow import DAG
+from airflow.models import BaseOperator
+
+if airflow.__version__.startswith("1."):
+    from airflow.operators.dummy_operator import DummyOperator
+else:
+    from airflow.operators.dummy import DummyOperator
+
+from pytimeparse import parse
+
+from dbt_airflow_factory.builder import DbtAirflowTasksBuilder
+from dbt_airflow_factory.builder_factory import DbtAirflowTasksBuilderFactory
+from dbt_airflow_factory.config_utils import read_config
+
+
+class AirflowDagFactory:
+    """
+    Factory creating Airflow DAG.
+
+    :param dag_path: path to ``manifest.json`` file.
+    :type dag_path: str
+    :param env: name of the environment.
+    :type env: str
+    :param dbt_config_file_name: name of the DBT config file.
+        If not specified, default value is ``dbt.yml``.
+    :type dbt_config_file_name: str
+    :param k8s_config_file_name: name of the Kubernetes config file.
+        If not specified, default value is ``k8s.yml``.
+    :type k8s_config_file_name: str
+    :param airflow_config_file_name: name of the Airflow config file.
+        If not specified, default value is ``airflow.yml``.
+    :type airflow_config_file_name: str
+    """
+
+    _builder: DbtAirflowTasksBuilder
+    dag_path: str
+    """path to ``manifest.json`` file."""
+    env: str
+    """name of the environment."""
+    airflow_config_file_name: str
+    """name of the Airflow config file (default: ``airflow.yml``)."""
+
+    def __init__(
+        self,
+        dag_path: str,
+        env: str,
+        dbt_config_file_name: str = "dbt.yml",
+        k8s_config_file_name: str = "k8s.yml",
+        airflow_config_file_name: str = "airflow.yml",
+    ):
+        self._builder = DbtAirflowTasksBuilderFactory(
+            dag_path, env, dbt_config_file_name, k8s_config_file_name
+        ).create()
+        self.dag_path = dag_path
+        self.env = env
+        self.airflow_config_file_name = airflow_config_file_name
+
+    def create(self) -> DAG:
+        """
+        Parse ``manifest.json`` and create tasks based on the data contained there.
+
+        :return: Generated DAG.
+        :rtype: airflow.models.dag.DAG
+        """
+        config = self.read_config()
+        with DAG(default_args=config["default_args"], **config["dag"]) as dag:
+            self.create_tasks(config)
+        return dag
+
+    def create_tasks(self, config: dict) -> None:
+        """
+        Parse ``manifest.json`` and create tasks based on the data contained there.
+
+        :param config: Dictionary representing ``airflow.yml``.
+        :type config: dict
+        """
+        start = self._create_starting_task(config)
+        end = DummyOperator(task_id="end")
+        tasks = self._builder.parse_manifest_into_tasks(
+            self._manifest_file_path(config), config.get("use_task_group") or False
+        )
+        for starting_task in tasks.get_starting_tasks():
+            start >> starting_task.run_airflow_task
+        for ending_task in tasks.get_ending_tasks():
+            ending_task.test_airflow_task >> end
+
+    def _create_starting_task(self, config: dict) -> BaseOperator:
+        if config.get("seed_task", True):
+            return self._builder.create_seed_task()
+        else:
+            return DummyOperator(task_id="start")
+
+    def _manifest_file_path(self, config: dict) -> str:
+        file_dir = config.get("manifest_dir_path", self.dag_path)
+        return os.path.join(file_dir, config.get("manifest_file_name", "manifest.json"))
+
+    def read_config(self) -> dict:
+        """
+        Read ``airflow.yml`` from ``config`` directory into a dictionary.
+
+        :return: Dictionary representing ``airflow.yml``.
+        :rtype: dict
+        :raises KeyError: No ``default_args`` key in ``airflow.yml``.
+        """
+        config = read_config(
+            self.dag_path, self.env, self.airflow_config_file_name, replace_jinja=True
+        )
+        if "retry_delay" in config["default_args"]:
+            config["default_args"]["retry_delay"] = parse(
+                config["default_args"]["retry_delay"]
+            )
+        return config
diff --git a/dbt_airflow_manifest_parser/builder.py b/dbt_airflow_factory/builder.py
similarity index 80%
rename from dbt_airflow_manifest_parser/builder.py
rename to dbt_airflow_factory/builder.py
index 8d47ad7..1017462 100644
--- a/dbt_airflow_manifest_parser/builder.py
+++ b/dbt_airflow_factory/builder.py
@@ -1,6 +1,8 @@
+"""Class parsing ``manifest.json`` into Airflow tasks."""
+
 import json
 import logging
-from typing import Dict
+from typing import Any, ContextManager, Dict, Tuple
 
 import airflow
 from airflow.models.baseoperator import BaseOperator
@@ -8,14 +10,18 @@
 if not airflow.__version__.startswith("1."):
     from airflow.utils.task_group import TaskGroup
 
-from dbt_airflow_manifest_parser.operator import (
-    DbtRunOperatorBuilder,
-    EphemeralOperator,
-)
-from dbt_airflow_manifest_parser.tasks import ModelExecutionTask, ModelExecutionTasks
+from dbt_airflow_factory.operator import DbtRunOperatorBuilder, EphemeralOperator
+from dbt_airflow_factory.tasks import ModelExecutionTask, ModelExecutionTasks
 
 
 class DbtAirflowTasksBuilder:
+    """
+    Parses ``manifest.json`` into Airflow tasks.
+
+    :param operator_builder: DBT node operator.
+    :type operator_builder: DbtRunOperatorBuilder
+    """
+
     def __init__(self, operator_builder: DbtRunOperatorBuilder):
         self.operator_builder = operator_builder
 
@@ -59,7 +65,9 @@ def _is_ephemeral_task(node: dict) -> bool:
         return node["config"]["materialized"] == "ephemeral"
 
     @staticmethod
-    def _create_task_group_for_model(model_name: str, use_task_group: bool):
+    def _create_task_group_for_model(
+        model_name: str, use_task_group: bool
+    ) -> Tuple[Any, ContextManager]:
         import contextlib
 
         is_first_version = airflow.__version__.startswith("1.")
@@ -71,7 +79,9 @@ def _create_task_group_for_model(model_name: str, use_task_group: bool):
         task_group_ctx = task_group or contextlib.nullcontext()
         return task_group, task_group_ctx
 
-    def _create_task_for_model(self, model_name: str, use_task_group: bool):
+    def _create_task_for_model(
+        self, model_name: str, use_task_group: bool
+    ) -> ModelExecutionTask:
         (task_group, task_group_ctx) = self._create_task_group_for_model(
             model_name, use_task_group
         )
@@ -132,7 +142,23 @@ def _make_dbt_tasks(
     def parse_manifest_into_tasks(
         self, manifest_path: str, use_task_group: bool = True
     ) -> ModelExecutionTasks:
+        """
+        Parse ``manifest.json`` into tasks.
+
+        :param manifest_path: Path to ``manifest.json``.
+        :type manifest_path: str
+        :param use_task_group: Whether to use TaskGroup (does not work in Airflow 1).
+        :type use_task_group: bool
+        :return: Dictionary of tasks created from ``manifest.json`` parsing.
+        :rtype: ModelExecutionTasks
+        """
         return self._make_dbt_tasks(manifest_path, use_task_group)
 
     def create_seed_task(self) -> BaseOperator:
+        """
+        Create ``dbt_seed`` task.
+
+        :return: Operator for ``dbt_seed`` task.
+        :rtype: BaseOperator
+        """
         return self.operator_builder.create("dbt_seed", "seed")
diff --git a/dbt_airflow_factory/builder_factory.py b/dbt_airflow_factory/builder_factory.py
new file mode 100644
index 0000000..e5640b7
--- /dev/null
+++ b/dbt_airflow_factory/builder_factory.py
@@ -0,0 +1,95 @@
+"""Factory creating Airflow tasks."""
+
+from typing import Any, List
+
+import airflow
+
+from dbt_airflow_factory.builder import DbtAirflowTasksBuilder
+from dbt_airflow_factory.config_utils import read_config
+from dbt_airflow_factory.dbt_parameters import DbtExecutionEnvironmentParameters
+from dbt_airflow_factory.k8s_parameters import KubernetesExecutionParameters
+from dbt_airflow_factory.operator import KubernetesPodOperatorBuilder
+
+
+class DbtAirflowTasksBuilderFactory:
+    """
+    Factory creating Airflow tasks.
+
+    :param dag_path: path to ``manifest.json`` file.
+    :type dag_path: str
+    :param env: name of the environment.
+    :type env: str
+    :param dbt_config_file_name: name of the DBT config file.
+        If not specified, default value is ``dbt.yml``.
+    :type dbt_config_file_name: str
+    :param k8s_config_file_name: name of the Kubernetes config file.
+        If not specified, default value is ``k8s.yml``.
+    :type k8s_config_file_name: str
+    """
+
+    base_config_name: str
+    """Name of the ``base`` environment (default: ``base``)."""
+    dag_path: str
+    """path to ``manifest.json`` file."""
+    env: str
+    """name of the environment."""
+    dbt_config_file_name: str
+    """name of the DBT config file (default: ``dbt.yml``)."""
+    k8s_config_file_name: str
+    """name of the Kubernetes config file (default: ``k8s.yml``)."""
+
+    def __init__(
+        self,
+        dag_path: str,
+        env: str,
+        dbt_config_file_name: str = "dbt.yml",
+        k8s_config_file_name: str = "k8s.yml",
+    ):
+        self.base_config_name = "base"
+        self.dag_path = dag_path
+        self.env = env
+        self.dbt_config_file_name = dbt_config_file_name
+        self.k8s_config_file_name = k8s_config_file_name
+
+    def create(self) -> DbtAirflowTasksBuilder:
+        """
+        Create :class:`.DbtAirflowTasksBuilder` to use.
+
+        :return: Instance of :class:`.DbtAirflowTasksBuilder`.
+        :rtype: DbtAirflowTasksBuilder
+        """
+        dbt_execution_env_params = self._create_dbt_config()
+        kubernetes_params = self._create_k8s_config()
+        return DbtAirflowTasksBuilder(
+            KubernetesPodOperatorBuilder(dbt_execution_env_params, kubernetes_params)
+        )
+
+    def _create_dbt_config(self) -> DbtExecutionEnvironmentParameters:
+        return DbtExecutionEnvironmentParameters(
+            **read_config(self.dag_path, self.env, self.dbt_config_file_name)
+        )
+
+    def _create_k8s_config(self) -> KubernetesExecutionParameters:
+        config = read_config(self.dag_path, self.env, self.k8s_config_file_name)
+        config["image"] = self._prepare_image(config["image"])
+        config["secrets"] = self._prepare_secrets(config)
+        config.update(config.pop("resources"))
+        return KubernetesExecutionParameters(**config)
+
+    @staticmethod
+    def _prepare_image(config: dict) -> str:
+        return config["repository"] + ":" + str(config["tag"])
+
+    def _prepare_secrets(self, config: dict) -> List[Any]:
+        return [self._prepare_secret(secret) for secret in config["secrets"]]
+
+    @staticmethod
+    def _prepare_secret(secret_dict: dict):  # type: ignore
+        if airflow.__version__.startswith("1."):
+            from airflow.contrib.kubernetes.secret import Secret
+
+            return Secret(**secret_dict)
+        else:
+            from airflow.kubernetes.secret import Secret
+
+            return Secret(**secret_dict)
diff --git a/dbt_airflow_factory/config_utils.py b/dbt_airflow_factory/config_utils.py
new file mode 100644
index 0000000..013e17c
--- /dev/null
+++ b/dbt_airflow_factory/config_utils.py
@@ -0,0 +1,114 @@
+"""Utilities for configuration files reading."""
+
+from __future__ import annotations
+
+import logging
+import os
+import pathlib
+from typing import Any, Union
+
+import yaml
+from airflow.models import Variable
+from jinja2 import FileSystemLoader
+from jinja2.nativetypes import NativeEnvironment
+
+
+def read_config(
+    dag_path: Union[str, os.PathLike[str]],
+    env: str,
+    file_name: str,
+    replace_jinja: bool = False,
+) -> dict:
+    """
+    Reads dictionaries out of *file_name* in both `base` and *env* directories,
+    and compiles them into one. Values from *env* directory get precedence over
+    `base` ones
+
+    :param dag_path: Path to the directory containing ``config`` directory.
+    :type dag_path: Union[str, os.PathLike[str]]
+    :param env: Name of the environment.
+    :type env: str
+    :param file_name: Name of the config file.
+    :type file_name: str
+    :param replace_jinja: Whether replace Airflow vars using Jinja templating.
+    :type replace_jinja: bool
+    :return: Dictionary representing the config file.
+    :rtype: dict
+    """
+    config = read_env_config(dag_path, "base", file_name, replace_jinja)
+    config.update(read_env_config(dag_path, env, file_name, replace_jinja))
+    return config
+
+
+def read_env_config(
+    dag_path: Union[str, os.PathLike[str]],
+    env: str,
+    file_name: str,
+    replace_jinja: bool = False,
+) -> dict:
+    """
+    Read config file, depending on the ``env``.
+
+    :param dag_path: Path to the directory containing ``config`` directory.
+    :type dag_path: Union[str, os.PathLike[str]]
+    :param env: Name of the environment.
+    :type env: str
+    :param file_name: Name of the config file.
+    :type file_name: str
+    :param replace_jinja: Whether replace Airflow vars using Jinja templating.
+    :type replace_jinja: bool
+    :return: Dictionary representing the config file.
+    :rtype: dict
+    """
+    config_file_path = os.path.join(dag_path, "config", env, file_name)
+    if os.path.exists(config_file_path):
+        logging.info("Reading config from " + config_file_path)
+        return read_yaml_file(config_file_path, replace_jinja)
+    logging.warning("Missing config file: " + config_file_path)
+    return {}
+
+
+def read_yaml_file(
+    file_path: Union[str, os.PathLike[str]], replace_jinja: bool
+) -> dict:
+    """
+    Load `yaml` file to dictionary.
+
+    :param file_path: Path to the file.
+    :type file_path: Union[str, os.PathLike[str]]
+    :param replace_jinja: Whether replace Airflow vars using Jinja templating.
+    :type replace_jinja: bool
+    :return: Loaded dictionary.
+    :rtype: dict
+    """
+    if replace_jinja:
+        return yaml.safe_load(_jinja_replace_airflow_vars(file_path))
+
+    with open(file_path, "r") as f:
+        return yaml.safe_load(f)
+
+
+def _jinja_replace_airflow_vars(file_path: Union[str, os.PathLike[str]]) -> str:
+    # Copied from airflow.models.taskinstance
+    class VariableAccessor:
+        def __init__(self) -> None:
+            self.var = None
+
+        def __getattr__(self, item: str) -> Any:
+            self.var = Variable.get(item)
+            return self.var
+
+        def __repr__(self) -> str:
+            return str(self.var)
+
+        @staticmethod
+        def get(item: str) -> Any:
+            return Variable.get(item)
+
+    file_path = pathlib.Path(file_path)
+    jinja_loader = FileSystemLoader(file_path.parent)
+    jinja_env = NativeEnvironment(loader=jinja_loader)
+
+    return jinja_env.get_template(file_path.name).render(
+        var={"value": VariableAccessor()}
+    )
diff --git a/dbt_airflow_factory/dbt_parameters.py b/dbt_airflow_factory/dbt_parameters.py
new file mode 100644
index 0000000..6c1538a
--- /dev/null
+++ b/dbt_airflow_factory/dbt_parameters.py
@@ -0,0 +1,48 @@
+"""POD representing DBT operator config file."""
+
+import sys
+from typing import Any, Dict, Optional
+
+import yaml
+
+
+class DbtExecutionEnvironmentParameters:
+    """POD representing DBT operator config file.
+
+    :param target: Name of the target environment (passed to **dbt** as ``--target``).
+    :type target: str
+    :param project_dir_path: Path to project directory.
+    :type project_dir_path: str
+    :param profile_dir_path: Path to the directory containing ``profiles.yml``.
+    :type project_dir_path: str
+    :param vars: Dictionary of variables to pass to the **dbt**.
+    :type vars: Optional[Dict[str, str]]
+    """
+
+    def __init__(
+        self,
+        target: str,
+        project_dir_path: str = "/dbt",
+        profile_dir_path: str = "/root/.dbt",
+        vars: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> None:
+        self.target = target
+        self.project_dir_path = project_dir_path
+        self.profile_dir_path = profile_dir_path
+        self._vars = vars or {}
+
+    @property
+    def vars(self) -> str:
+        """
+        String representation of dictionary of **dbt** variables.
+
+        DBT expects ``--vars`` passing string in YAML format. This property
+        returns such a string.
+
+        :return: String representation of dictionary of **dbt** variables.
+        :rtype: str
+        """
+        return yaml.dump(
+            self._vars, default_flow_style=True, width=sys.maxsize
+        ).rstrip()
diff --git a/dbt_airflow_factory/k8s_parameters.py b/dbt_airflow_factory/k8s_parameters.py
new file mode 100644
index 0000000..5bd3b08
--- /dev/null
+++ b/dbt_airflow_factory/k8s_parameters.py
@@ -0,0 +1,123 @@
+"""POD representing Kubernetes operator config file."""
+
+from typing import Any, Dict, List, Optional
+
+import airflow
+
+if airflow.__version__.startswith("1."):
+    from airflow.contrib.kubernetes.secret import Secret
+else:
+    from airflow.kubernetes.secret import Secret
+
+
+class KubernetesExecutionParameters:
+    """POD representing Kubernetes operator config file.
+
+    :param image: tag of Docker image you wish to launch.
+    :type image: str
+    :param namespace: the namespace to run within Kubernetes.
+    :type namespace: str
+    :param image_pull_policy: Specify a policy to cache or always pull an image.
+    :type image_pull_policy: str
+    :param node_selectors: A dict containing a group of scheduling rules.
+    :type node_selectors: dict
+    :param tolerations: A list of Kubernetes tolerations.
+    :type tolerations: list
+    :param labels: labels to apply to the Pod. (templated)
+    :type labels: dict
+    :param limit: A dict containing resources limits.
+    :type limit: dict
+    :param requests: A dict containing resources requests.
+    :type requests: dict
+    :param annotations: non-identifying metadata you can attach to the Pod.
+    :type annotations: dict
+    :param envs: Environment variables initialized in the container. (templated)
+    :type envs: Optional[Dict[str, str]]
+    :param secrets: Kubernetes secrets to inject in the container.
+        They can be exposed as environment vars or files in a volume.
+    :type secrets: List[Secret]
+    :param is_delete_operator_pod: What to do when the pod reaches its final
+        state, or the execution is interrupted. If True: delete the pod.
+    :type is_delete_operator_pod: bool
+    """
+
+    def __init__(
+        self,
+        image: str,
+        namespace: str = "default",
+        image_pull_policy: Optional[str] = None,
+        node_selectors: Optional[dict] = None,
+        tolerations: Optional[list] = None,
+        labels: Optional[dict] = None,
+        limit: Optional[dict] = None,
+        requests: Optional[dict] = None,
+        annotations: Optional[dict] = None,
+        envs: Optional[Dict[str, str]] = None,
+        secrets: Optional[List[Secret]] = None,
+        is_delete_operator_pod: bool = True,
+        **kwargs: Any,
+    ) -> None:
+        self.namespace = namespace
+        self.image = image
+        self.image_pull_policy = image_pull_policy
+        self.node_selectors = node_selectors
+        self.tolerations = tolerations
+        self.labels = labels
+        self._limit = limit
+        self._requests = requests
+        self.annotations = annotations
+        self._env_vars = envs
+        self.secrets = secrets
+        self.is_delete_operator_pod = is_delete_operator_pod
+
+    @property
+    def resources(self):  # type: ignore
+        """
+        Return dict containing resources requests and limits.
+
+        In the Airflow 1, it was expected to be a real dictionary with
+        ``request_memory``, ``request_cpu``, ``limit_memory``, and ``limit_cpu``
+        as keys. So for Airflow 1, the function returns such a dictionary.
+
+        Beginning with Airflow 2, :class:`KubernetesPodOperator` expects
+        ``V1ResourceRequirements`` class instead. Hence, for Airflow 2, the
+        function returns instance of this class.
+
+        :return: Dictionary containing resources requests and limits.
+        """
+        if airflow.__version__.startswith("1."):
+            return {
+                "limit_memory": self._limit["memory"] if self._limit else None,
+                "limit_cpu": self._limit["cpu"] if self._limit else None,
+                "request_memory": self._requests["memory"] if self._requests else None,
+                "request_cpu": self._requests["cpu"] if self._requests else None,
+            }
+        else:
+            from kubernetes.client import models as k8s
+
+            return k8s.V1ResourceRequirements(
+                limits=self._limit, requests=self._requests
+            )
+
+    @property
+    def env_vars(self):  # type: ignore
+        """
+        Return dict containing environment variables to set in Kubernetes.
+
+        For Airflow 1, the function returns a dictionary.
+
+        Beginning with Airflow 2, :class:`KubernetesPodOperator` expects
+        a list of ``V1EnvVar`` instances instead. Hence, for Airflow 2, the
+        function returns a ``List[k8s.V1EnvVar]``.
+
+        :return: Dictionary or list containing environment variables.
+        """
+        if self._env_vars is None:
+            return None
+
+        if airflow.__version__.startswith("1."):
+            return self._env_vars
+        else:
+            from kubernetes.client import models as k8s
+
+            return [k8s.V1EnvVar(k, v) for k, v in self._env_vars.items()]
diff --git a/dbt_airflow_manifest_parser/operator.py b/dbt_airflow_factory/operator.py
similarity index 52%
rename from dbt_airflow_manifest_parser/operator.py
rename to dbt_airflow_factory/operator.py
index 8ed8106..02d26a8 100644
--- a/dbt_airflow_manifest_parser/operator.py
+++ b/dbt_airflow_factory/operator.py
@@ -1,28 +1,66 @@
+"""Factories creating Airflow Operators running DBT tasks."""
+
 import abc
+from typing import List, Optional
 
 import airflow
-from airflow.operators.dummy import DummyOperator
 
-from dbt_airflow_manifest_parser.dbt_parameters import DbtExecutionEnvironmentParameters
-from dbt_airflow_manifest_parser.k8s_parameters import KubernetesExecutionParameters
+from dbt_airflow_factory.dbt_parameters import DbtExecutionEnvironmentParameters
+from dbt_airflow_factory.k8s_parameters import KubernetesExecutionParameters
 
 if airflow.__version__.startswith("1."):
     from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
+    from airflow.operators.dummy_operator import DummyOperator
 else:
     from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import (
         KubernetesPodOperator,
     )
+    from airflow.operators.dummy import DummyOperator
 
 from airflow.models.baseoperator import BaseOperator
 
 
 class DbtRunOperatorBuilder(metaclass=abc.ABCMeta):
+    """
+    Base class of a factory creating Airflow
+    :class:`airflow.models.baseoperator.BaseOperator` running a single DBT task.
+    """
+
     @abc.abstractmethod
-    def create(self, name: str, command: str, model: str = None) -> BaseOperator:
+    def create(
+        self, name: str, command: str, model: Optional[str] = None
+    ) -> BaseOperator:
+        """
+        Create Airflow Operator running a single DBT task.
+
+        :param name: task name.
+        :type name: str
+        :param command: DBT command to run.
+        :type command: str
+        :param model: models to include.
+        :type model: Optional[str]
+        :return: Airflow Operator running a single DBT task.
+        :rtype: BaseOperator
+        """
         raise NotImplementedError
 
 
 class KubernetesPodOperatorBuilder(DbtRunOperatorBuilder):
+    """
+    Builder of Kubernetes Pod Operator running a single DBT task.
+
+    :param dbt_execution_env_parameters: POD representing DBT operator config file.
+    :type dbt_execution_env_parameters: DbtExecutionEnvironmentParameters
+    :param kubernetes_execution_parameters:
+        POD representing Kubernetes operator config file.
+    :type kubernetes_execution_parameters: KubernetesExecutionParameters
+    """
+
+    dbt_execution_env_parameters: DbtExecutionEnvironmentParameters
+    """POD representing DBT operator config file."""
+    kubernetes_execution_parameters: KubernetesExecutionParameters
+    """POD representing Kubernetes operator config file."""
+
     def __init__(
         self,
         dbt_execution_env_parameters: DbtExecutionEnvironmentParameters,
@@ -31,10 +69,12 @@ def __init__(
         self.dbt_execution_env_parameters = dbt_execution_env_parameters
         self.kubernetes_execution_parameters = kubernetes_execution_parameters
 
-    def create(self, name: str, command: str, model: str = None) -> BaseOperator:
+    def create(
+        self, name: str, command: str, model: Optional[str] = None
+    ) -> BaseOperator:
         return self._create(self._prepare_arguments(command, model), name)
 
-    def _prepare_arguments(self, command: str, model: str):
+    def _prepare_arguments(self, command: str, model: Optional[str]) -> List[str]:
         args = (
             f"set -e; "
             f"dbt --no-write-json {command} "
@@ -49,7 +89,7 @@ def _prepare_arguments(self, command: str, model: str):
         )
         return [args]
 
-    def _create(self, args, name):
+    def _create(self, args: Optional[List[str]], name: str) -> KubernetesPodOperator:
         return KubernetesPodOperator(
             namespace=self.kubernetes_execution_parameters.namespace,
             image=self.kubernetes_execution_parameters.image,
@@ -62,13 +102,17 @@ def _create(self, args, name):
             labels=self.kubernetes_execution_parameters.labels,
             name=name,
             task_id=name,
-            resources=self.kubernetes_execution_parameters.get_resources(),
+            resources=self.kubernetes_execution_parameters.resources,
             env_vars=self.kubernetes_execution_parameters.env_vars,
             secrets=self.kubernetes_execution_parameters.secrets,
-            is_delete_operator_pod=self.kubernetes_execution_parameters.is_delete_operator_pod,
+            is_delete_operator_pod=self.kubernetes_execution_parameters.is_delete_operator_pod,  # noqa: E501
             hostnetwork=False,
         )
 
 
 class EphemeralOperator(DummyOperator):
+    """
+    :class:`DummyOperator` representing ephemeral DBT model.
+    """
+
     ui_color = "#F3E4F7"
diff --git a/dbt_airflow_factory/tasks.py b/dbt_airflow_factory/tasks.py
new file mode 100644
index 0000000..0c98ab9
--- /dev/null
+++ b/dbt_airflow_factory/tasks.py
@@ -0,0 +1,120 @@
+"""Classes representing tasks corresponding to a single DBT model."""
+
+from typing import Dict, Iterable, List, Optional
+
+from airflow.models.baseoperator import BaseOperator
+
+
+class ModelExecutionTask:
+    """
+    Wrapper around tasks corresponding to a single DBT model.
+
+    :param run_airflow_task: Operator running DBT's ``run`` task.
+    :type run_airflow_task: BaseOperator
+    :param test_airflow_task: Operator running DBT's ``test`` task (optional).
+    :type test_airflow_task: BaseOperator
+    :param task_group: TaskGroup consisting of ``run`` and ``test`` tasks
+        (if Airflow version is at least 2).
+    """
+
+    def __init__(  # type: ignore
+        self,
+        run_airflow_task: BaseOperator,
+        test_airflow_task: Optional[BaseOperator],
+        task_group=None,
+    ) -> None:
+        self.run_airflow_task = run_airflow_task
+        self.test_airflow_task = test_airflow_task
+        self.task_group = task_group
+
+    def __repr__(self) -> str:
+        return (
+            repr(self.task_group)
+            if self.task_group
+            else repr(
+                [self.run_airflow_task]
+                + ([self.test_airflow_task] if self.test_airflow_task else [])
+            )
+        )
+
+    def get_start_task(self):  # type: ignore
+        """
+        Return model's first task.
+
+        It is either a whole TaskGroup or ``run`` task.
+        """
+        return self.task_group or self.run_airflow_task
+
+    def get_end_task(self):  # type: ignore
+        """
+        Return model's last task.
+
+        It is either a whole TaskGroup, ``test`` task, or ``run`` task, depending
+        on version of Airflow and existence of ``test`` task.
+        """
+        return self.task_group or self.test_airflow_task or self.run_airflow_task
+
+
+class ModelExecutionTasks:
+    """
+    Dictionary of all Operators corresponding to DBT tasks.
+
+    :param tasks: Dictionary of model tasks.
+    :type tasks: Dict[str, ModelExecutionTask]
+    :param starting_task_names: List of names of initial tasks (DAG sources).
+    :type starting_task_names: List[str]
+    :param ending_task_names: List of names of ending tasks (DAG sinks).
+    :type ending_task_names: List[str]
+    """
+
+    def __init__(
+        self,
+        tasks: Dict[str, ModelExecutionTask],
+        starting_task_names: List[str],
+        ending_task_names: List[str],
+    ) -> None:
+        self._tasks = tasks
+        self._starting_task_names = starting_task_names
+        self._ending_task_names = ending_task_names
+
+    def __repr__(self) -> str:
+        return f"ModelExecutionTasks(\n {self._tasks} \n)"
+
+    def get_task(self, node_name: str) -> ModelExecutionTask:
+        """
+        Return :class:`ModelExecutionTask` for given model's **node_name**.
+
+        :param node_name: Name of the task.
+        :type node_name: str
+        :return: Wrapper around tasks corresponding to a given model.
+        :rtype: ModelExecutionTask
+        """
+        return self._tasks[node_name]
+
+    def length(self) -> int:
+        """Count TaskGroups corresponding to a single DBT model."""
+        return len(self._tasks)
+
+    def get_starting_tasks(self) -> List[ModelExecutionTask]:
+        """
+        Get a list of all DAG sources.
+
+        :return: List of all DAG sources.
+        :rtype: List[ModelExecutionTask]
+        """
+        return self._extract_by_keys(self._starting_task_names)
+
+    def get_ending_tasks(self) -> List[ModelExecutionTask]:
+        """
+        Get a list of all DAG sinks.
+
+        :return: List of all DAG sinks.
+        :rtype: List[ModelExecutionTask]
+        """
+        return self._extract_by_keys(self._ending_task_names)
+
+    def _extract_by_keys(self, keys: Iterable[str]) -> List[ModelExecutionTask]:
+        tasks = []
+        for key in keys:
+            tasks.append(self._tasks[key])
+        return tasks
diff --git a/dbt_airflow_manifest_parser/__init__.py b/dbt_airflow_manifest_parser/__init__.py
deleted file mode 100644
index 8d361f8..0000000
--- a/dbt_airflow_manifest_parser/__init__.py
+++ /dev/null
@@ -1,1 +0,0 @@
-version = "0.17.0"
diff --git a/dbt_airflow_manifest_parser/airflow_dag_factory.py b/dbt_airflow_manifest_parser/airflow_dag_factory.py
deleted file mode 100644
index 9deeafb..0000000
--- a/dbt_airflow_manifest_parser/airflow_dag_factory.py
+++ /dev/null
@@ -1,66 +0,0 @@
-import os
-
-import airflow
-from airflow import DAG
-
-if airflow.__version__.startswith("1."):
-    from airflow.operators.dummy_operator import DummyOperator
-else:
-    from airflow.operators.dummy import DummyOperator
-
-from pytimeparse import parse
-
-from dbt_airflow_manifest_parser.builder_factory import DbtAirflowTasksBuilderFactory
-from dbt_airflow_manifest_parser.config_utils import read_config
-
-
-class AirflowDagFactory:
-    def __init__(
-        self,
-        dag_path,
-        env: str,
-        dbt_config_file_name: str = "dbt.yml",
-        k8s_config_file_name: str = "k8s.yml",
-        airflow_config_file_name: str = "airflow.yml",
-    ):
-        self._builder = DbtAirflowTasksBuilderFactory(
-            dag_path, env, dbt_config_file_name, k8s_config_file_name
-        ).create()
-        self.dag_path = dag_path
-        self.env = env
-        self.airflow_config_file_name = airflow_config_file_name
-
-    def create(self) -> DAG:
-        config = self.read_config()
-        with DAG(default_args=config["default_args"], **config["dag"]) as dag:
-            self.create_tasks(config)
-        return dag
-
-    def create_tasks(self, config):
-        start = self._create_starting_task(config)
-        end = DummyOperator(task_id="end")
-        tasks = self._builder.parse_manifest_into_tasks(
-            self._manifest_file_path(config), config.get("use_task_group") or False
-        )
-        for starting_task in tasks.get_starting_tasks():
-            start >> starting_task.run_airflow_task
-        for ending_task in tasks.get_ending_tasks():
-            ending_task.test_airflow_task >> end
-
-    def _create_starting_task(self, config):
-        if config.get("seed_task", True):
-            return self._builder.create_seed_task()
-        else:
-            return DummyOperator(task_id="start")
-
-    def _manifest_file_path(self, config: dict) -> str:
-        file_dir = config.get("manifest_dir_path", self.dag_path)
-        return os.path.join(file_dir, config.get("manifest_file_name", "manifest.json"))
-
-    def read_config(self) -> dict:
-        config = read_config(self.dag_path, self.env, self.airflow_config_file_name)
-        if "retry_delay" in config["default_args"]:
-            config["default_args"]["retry_delay"] = parse(
-                config["default_args"]["retry_delay"]
-            )
-        return config
diff --git a/dbt_airflow_manifest_parser/builder_factory.py b/dbt_airflow_manifest_parser/builder_factory.py
deleted file mode 100644
index cada0dc..0000000
--- a/dbt_airflow_manifest_parser/builder_factory.py
+++ /dev/null
@@ -1,59 +0,0 @@
-import airflow
-
-from dbt_airflow_manifest_parser.builder import DbtAirflowTasksBuilder
-from dbt_airflow_manifest_parser.config_utils import read_config
-from dbt_airflow_manifest_parser.dbt_parameters import DbtExecutionEnvironmentParameters
-from dbt_airflow_manifest_parser.k8s_parameters import KubernetesExecutionParameters
-from dbt_airflow_manifest_parser.operator import KubernetesPodOperatorBuilder
-
-
-class DbtAirflowTasksBuilderFactory:
-    def __init__(
-        self,
-        dag_path: str,
-        env: str,
-        dbt_config_file_name: str = "dbt.yml",
-        k8s_config_file_name: str = "k8s.yml",
-    ):
-        self.base_config_name = "base"
-        self.dag_path = dag_path
-        self.env = env
-        self.dbt_config_file_name = dbt_config_file_name
-        self.k8s_config_file_name = k8s_config_file_name
-
-    def create(self) -> DbtAirflowTasksBuilder:
-        dbt_execution_env_params = self._create_dbt_config()
-        kubernetes_params = self._create_k8s_config()
-        return DbtAirflowTasksBuilder(
-            KubernetesPodOperatorBuilder(dbt_execution_env_params, kubernetes_params)
-        )
-
-    def _create_dbt_config(self):
-        return DbtExecutionEnvironmentParameters(
-            **read_config(self.dag_path, self.env, self.dbt_config_file_name)
-        )
-
-    def _create_k8s_config(self):
-        config = read_config(self.dag_path, self.env, self.k8s_config_file_name)
-        config["image"] = self._prepare_image(config["image"])
-        config["secrets"] = self._prepare_secrets(config)
-        config.update(config.pop("resources"))
-        return KubernetesExecutionParameters(**config)
-
-    @staticmethod
-    def _prepare_image(config):
-        return config["repository"] + ":" + str(config["tag"])
-
-    def _prepare_secrets(self, config):
-        return [self._prepare_secret(secret) for secret in config["secrets"]]
-
-    @staticmethod
-    def _prepare_secret(secret_dict):
-        if airflow.__version__.startswith("1."):
-            from airflow.contrib.kubernetes.secret import Secret
-
-            return Secret(**secret_dict)
-        else:
-            from airflow.kubernetes.secret import Secret
-
-            return Secret(**secret_dict)
diff --git a/dbt_airflow_manifest_parser/config_utils.py b/dbt_airflow_manifest_parser/config_utils.py
index 821ce69..e69de29 100644
--- a/dbt_airflow_manifest_parser/config_utils.py
+++ b/dbt_airflow_manifest_parser/config_utils.py
@@ -1,24 +0,0 @@
-import logging
-from os import path
-
-import yaml
-
-
-def read_config(dag_path, env, file_name):
-    config = read_env_config(dag_path, "base", file_name)
-    config.update(read_env_config(dag_path, env, file_name))
-    return config
-
-
-def read_env_config(dag_path, env, file_name):
-    config_file_path = path.join(dag_path, "config", env, file_name)
-    if path.exists(config_file_path):
-        logging.info("Reading config from " + config_file_path)
-        return read_yaml_file(config_file_path)
-    logging.warning("Missing config file: " + config_file_path)
-    return {}
-
-
-def read_yaml_file(file_path):
-    with open(file_path, "r") as f:
-        return yaml.safe_load(f)
diff --git a/dbt_airflow_manifest_parser/dbt_parameters.py b/dbt_airflow_manifest_parser/dbt_parameters.py
deleted file mode 100644
index 0c9e7b1..0000000
--- a/dbt_airflow_manifest_parser/dbt_parameters.py
+++ /dev/null
@@ -1,25 +0,0 @@
-import sys
-from typing import Dict, Optional
-
-import yaml
-
-
-class DbtExecutionEnvironmentParameters:
-    def __init__(
-        self,
-        target: str,
-        project_dir_path: str = "/dbt",
-        profile_dir_path: str = "/root/.dbt",
-        vars: Optional[Dict[str, str]] = None,
-        **kwargs,
-    ):
-        self.target = target
-        self.project_dir_path = project_dir_path
-        self.profile_dir_path = profile_dir_path
-        self._vars = vars or {}
-
-    @property
-    def vars(self) -> str:
-        return yaml.dump(
-            self._vars, default_flow_style=True, width=sys.maxsize
-        ).rstrip()
diff --git a/dbt_airflow_manifest_parser/k8s_parameters.py b/dbt_airflow_manifest_parser/k8s_parameters.py
deleted file mode 100644
index 2dd5a90..0000000
--- a/dbt_airflow_manifest_parser/k8s_parameters.py
+++ /dev/null
@@ -1,53 +0,0 @@
-from typing import Dict, List, Optional
-
-import airflow
-from kubernetes.client import models as k8s
-
-if airflow.__version__.startswith("1."):
-    from airflow.contrib.kubernetes.secret import Secret
-else:
-    from airflow.kubernetes.secret import Secret
-
-
-class KubernetesExecutionParameters:
-    def __init__(
-        self,
-        image: str,
-        namespace: str = "default",
-        image_pull_policy: Optional[str] = None,
-        node_selectors: Optional[dict] = None,
-        tolerations: Optional[list] = None,
-        labels: Optional[dict] = None,
-        limit: Optional[dict] = None,
-        requests: Optional[dict] = None,
-        annotations: Optional[dict] = None,
-        envs: Optional[Dict[str, str]] = None,
-        secrets: Optional[List[Secret]] = None,
-        is_delete_operator_pod: bool = True,
-        **kwargs,
-    ):
-        self.namespace = namespace
-        self.image = image
-        self.image_pull_policy = image_pull_policy
-        self.node_selectors = node_selectors
-        self.tolerations = tolerations
-        self.labels = labels
-        self.limit = limit
-        self.requests = requests
-        self.annotations = annotations
-        self.env_vars = [k8s.V1EnvVar(k, v) for k, v in envs.items()] if envs else None
-        self.secrets = secrets
-        self.is_delete_operator_pod = is_delete_operator_pod
-
-    def get_resources(self):
-        if airflow.__version__.startswith("1."):
-            return {
-                "limit_memory": self.limit["memory"],
-                "limit_cpu": self.limit["cpu"],
-                "request_memory": self.requests["memory"],
-                "request_cpu": self.requests["cpu"],
-            }
-        else:
-            from kubernetes.client import models as k8s
-
-            return k8s.V1ResourceRequirements(limits=self.limit, requests=self.requests)
diff --git a/dbt_airflow_manifest_parser/tasks.py b/dbt_airflow_manifest_parser/tasks.py
deleted file mode 100644
index 8f708b8..0000000
--- a/dbt_airflow_manifest_parser/tasks.py
+++ /dev/null
@@ -1,64 +0,0 @@
-from typing import Dict, List, Optional
-
-from airflow.models.baseoperator import BaseOperator
-
-
-class ModelExecutionTask:
-    def __init__(
-        self,
-        run_airflow_task: BaseOperator,
-        test_airflow_task: Optional[BaseOperator],
-        task_group=None,
-    ):
-        self.run_airflow_task = run_airflow_task
-        self.test_airflow_task = test_airflow_task
-        self.task_group = task_group
-
-    def __repr__(self):
-        return (
-            repr(self.task_group)
-            if self.task_group
-            else repr(
-                [self.run_airflow_task]
-                + ([self.test_airflow_task] if self.test_airflow_task else [])
-            )
-        )
-
-    def get_start_task(self):
-        return self.task_group or self.run_airflow_task
-
-    def get_end_task(self):
-        return self.task_group or self.test_airflow_task or self.run_airflow_task
-
-
-class ModelExecutionTasks:
-    def __init__(
-        self,
-        tasks: Dict[str, ModelExecutionTask],
-        starting_task_names: List[str],
-        ending_task_names: List[str],
-    ):
-        self._tasks = tasks
-        self._starting_task_names = starting_task_names
-        self._ending_task_names = ending_task_names
-
-    def __repr__(self):
-        return f"ModelExecutionTasks(\n {self._tasks} \n)"
-
-    def get_task(self, node_name) -> ModelExecutionTask:
-        return self._tasks[node_name]
-
-    def length(self) -> int:
-        return len(self._tasks)
-
-    def get_starting_tasks(self) -> List[ModelExecutionTask]:
-        return self._extract_by_keys(self._starting_task_names)
-
-    def get_ending_tasks(self) -> List[ModelExecutionTask]:
-        return self._extract_by_keys(self._ending_task_names)
-
-    def _extract_by_keys(self, keys) -> List[ModelExecutionTask]:
-        tasks = []
-        for key in keys:
-            tasks.append(self._tasks[key])
-        return tasks
diff --git a/docs/Makefile b/docs/Makefile
new file mode 100644
index 0000000..6fe7b3f
--- /dev/null
+++ b/docs/Makefile
@@ -0,0 +1,23 @@
+# Minimal makefile for Sphinx documentation
+#
+
+# You can set these variables from the command line, and also
+# from the environment for the first two.
+SPHINXOPTS    ?= -Ea
+SPHINXBUILD   ?= sphinx-build
+SOURCEDIR     = .
+BUILDDIR      = _build
+
+# Put it first so that "make" without argument is like "make help".
+help:
+	@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
+
+apidoc:
+	@sphinx-apidoc ../dbt_airflow_factory -o source -fMT -d 1
+
+.PHONY: help apidoc Makefile
+
+# Catch-all target: route all unknown targets to Sphinx using the new
+# "make mode" option.  $(O) is meant as a shortcut for $(SPHINXOPTS).
+%: Makefile
+	@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
diff --git a/docs/api.rst b/docs/api.rst
new file mode 100644
index 0000000..06dc905
--- /dev/null
+++ b/docs/api.rst
@@ -0,0 +1,10 @@
+API Reference
+=============
+
+If you are looking for information on a specific function, class, or
+method, this part of the documentation is for you.
+
+.. toctree::
+   :maxdepth: 1
+
+   source/dbt_airflow_factory
diff --git a/docs/changelog.rst b/docs/changelog.rst
new file mode 100644
index 0000000..362b326
--- /dev/null
+++ b/docs/changelog.rst
@@ -0,0 +1,2 @@
+.. include:: ../CHANGELOG.md
+   :parser: myst_parser.docutils_
\ No newline at end of file
diff --git a/docs/conf.py b/docs/conf.py
new file mode 100644
index 0000000..2e36223
--- /dev/null
+++ b/docs/conf.py
@@ -0,0 +1,55 @@
+# Configuration file for the Sphinx documentation builder.
+#
+# This file only contains a selection of the most common options. For a full
+# list see the documentation:
+# https://www.sphinx-doc.org/en/master/usage/configuration.html
+
+# -- Path setup --------------------------------------------------------------
+
+import os
+import sys
+
+sys.path.insert(0, os.path.abspath(".."))
+
+
+# -- Project information -----------------------------------------------------
+
+project = "dbt-airflow-factory"
+copyright = "2021, GetInData"
+author = "GetInData"
+
+
+# -- General configuration ---------------------------------------------------
+
+# Add any Sphinx extension module names here, as strings. They can be
+# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
+# ones.
+extensions = [
+    "sphinx.ext.autodoc",
+    "sphinx.ext.viewcode",
+    "sphinx_click.ext",
+    "myst_parser",
+]
+
+# Add any paths that contain templates here, relative to this directory.
+templates_path = ["_templates"]
+
+# List of patterns, relative to source directory, that match files and
+# directories to ignore when looking for source files.
+# This pattern also affects html_static_path and html_extra_path.
+exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
+
+add_module_names = False
+
+
+# -- Options for HTML output -------------------------------------------------
+
+# The theme to use for HTML and HTML Help pages.  See the documentation for
+# a list of builtin themes.
+#
+html_theme = "sphinx_rtd_theme"
+
+# Add any paths that contain custom static files (such as style sheets) here,
+# relative to this directory. They are copied after the builtin static files,
+# so a file named "default.css" will overwrite the builtin "default.css".
+html_static_path = ["_static"]
diff --git a/docs/index.rst b/docs/index.rst
new file mode 100644
index 0000000..a4fe256
--- /dev/null
+++ b/docs/index.rst
@@ -0,0 +1,19 @@
+``dbt-airflow-factory``
+=======================
+
+
+Installation
+------------
+Use the package manager `pip <https://pip.pypa.io/en/stable/>`_ to install the library:
+
+.. code-block:: sh
+
+   pip install dbt-airflow-factory
+
+.. toctree::
+   :maxdepth: 1
+   :caption: Contents:
+
+   usage
+   api
+   changelog
diff --git a/docs/requirements.txt b/docs/requirements.txt
new file mode 100644
index 0000000..3126b70
--- /dev/null
+++ b/docs/requirements.txt
@@ -0,0 +1,4 @@
+sphinx-rtd-theme==1.0.0
+sphinx-click>=3.0,<3.1
+myst-parser>=0.16, <0.17
+docutils>=0.17,<0.18
diff --git a/docs/source/dbt_airflow_factory.rst b/docs/source/dbt_airflow_factory.rst
new file mode 100644
index 0000000..debdf5e
--- /dev/null
+++ b/docs/source/dbt_airflow_factory.rst
@@ -0,0 +1,74 @@
+dbt\_airflow\_factory package
+=============================
+
+.. automodule:: dbt_airflow_factory
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+Submodules
+----------
+
+dbt\_airflow\_factory.airflow\_dag\_factory module
+--------------------------------------------------
+
+.. automodule:: dbt_airflow_factory.airflow_dag_factory
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+dbt\_airflow\_factory.builder module
+------------------------------------
+
+.. automodule:: dbt_airflow_factory.builder
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+dbt\_airflow\_factory.builder\_factory module
+---------------------------------------------
+
+.. automodule:: dbt_airflow_factory.builder_factory
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+dbt\_airflow\_factory.config\_utils module
+------------------------------------------
+
+.. automodule:: dbt_airflow_factory.config_utils
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+dbt\_airflow\_factory.dbt\_parameters module
+--------------------------------------------
+
+.. automodule:: dbt_airflow_factory.dbt_parameters
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+dbt\_airflow\_factory.k8s\_parameters module
+--------------------------------------------
+
+.. automodule:: dbt_airflow_factory.k8s_parameters
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+dbt\_airflow\_factory.operator module
+-------------------------------------
+
+.. automodule:: dbt_airflow_factory.operator
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+dbt\_airflow\_factory.tasks module
+----------------------------------
+
+.. automodule:: dbt_airflow_factory.tasks
+   :members:
+   :undoc-members:
+   :show-inheritance:
diff --git a/docs/usage.rst b/docs/usage.rst
new file mode 100644
index 0000000..56256d1
--- /dev/null
+++ b/docs/usage.rst
@@ -0,0 +1,59 @@
+Usage
+-----
+
+The library is expected to be used inside an Airflow environment with a Kubernetes image referencing **dbt**.
+
+**dbt-airflow-factory**'s main task is to parse ``manifest.json`` and create Airflow DAG out of it. It also reads config
+files from ``config`` directory and therefore is highly customizable (e.g., user can set path to ``manifest.json``).
+
+To start, create a directory with a following structure, where ``manifest.json`` is a file generated by **dbt**:
+
+.. code-block:: bash
+
+ .
+ ├── config
+ │   ├── base
+ │   │   ├── airflow.yml
+ │   │   ├── dbt.yml
+ │   │   └── k8s.yml
+ │   └── dev
+ │       └── dbt.yml
+ ├── dag.py
+ └── manifest.json
+
+Then, put the following code into ``dag.py``:
+
+.. code-block:: python
+
+ from dbt_airflow_factory.airflow_dag_factory import AirflowDagFactory
+ from os import path
+
+ dag = AirflowDagFactory(path.dirname(path.abspath(__file__)), "dev").create()
+
+When uploaded to Airflow DAGs directory, it will get picked up by Airflow, parse ``manifest.json`` and prepare a DAG to run.
+
+Configuration files
++++++++++++++++++++
+
+It is best to look up the example configuration files in
+`tests directory <https://github.com/getindata/dbt-airflow-factory/tree/develop/tests/config>`_ to get a glimpse
+of correct configs.
+
+You can use `Airflow template variables <https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#variables>`_
+in your ``dbt.yml`` and ``k8s.yml`` files, as long as they are inside quotation marks:
+
+.. code-block:: yaml
+
+ target: "{{ var.value.env }}"
+ some_other_field: "{{ ds_nodash }}"
+
+Analogously, you can use ``"{{ var.value.VARIABLE_NAME }}"`` in ``airflow.yml``, but only the Airflow variable getter.
+Any other Airflow template variables will not work in ``airflow.yml``.
+
+
+Creation of the directory with data-pipelines-cli
++++++++++++++++++++++++++++++++++++++++++++++++++
+
+**DBT Airflow Factory** works best in tandem with `data-pipelines-cli <https://pypi.org/project/data-pipelines-cli/>`_
+tool. **dp** not only prepares directory for the library to digest, but also automates Docker image building and pushes
+generated directory to the cloud storage of your choice.
diff --git a/setup.cfg b/setup.cfg
index 7c87f3c..e3c166e 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,6 +1,25 @@
 [bumpversion]
-current_version = 0.17.0
+current_version = 0.18.0
 
 [bumpversion:file:setup.py]
 
-[bumpversion:file:dbt_airflow_manifest_parser/__init__.py]
+[bumpversion:file:dbt_airflow_factory/__init__.py]
+
+[flake8]
+exclude = .git,__pycache__,build,dist,docs/source/conf.py
+max-line-length = 88
+extend-ignore = E203
+
+[mypy]
+ignore_missing_imports = True
+follow_imports = silent
+strict_optional = True
+no_implicit_optional = True
+warn_redundant_casts = True
+warn_unused_ignores = True
+check_untyped_defs = True
+no_implicit_reexport = True
+disallow_untyped_defs = True
+
+[mypy-tests.*]
+ignore_errors = True
diff --git a/setup.py b/setup.py
index 2500283..7c4d61d 100644
--- a/setup.py
+++ b/setup.py
@@ -1,4 +1,4 @@
-"""kedro_airflow_k8s module."""
+"""dbt_airflow_factory module."""
 
 from setuptools import find_packages, setup
 
@@ -19,15 +19,17 @@
         "apache-airflow[kubernetes]==2.2.0",
     ],
     "docs": [
-        "sphinx==3.4.2",
-        "recommonmark==0.7.1",
-        "sphinx_rtd_theme==0.5.2",
+        "sphinx==4.3.1",
+        "sphinx-rtd-theme==1.0.0",
+        "sphinx-click>=3.0,<3.1",
+        "myst-parser>=0.16, <0.17",
+        "docutils>=0.17,<0.18",
     ],
 }
 
 setup(
-    name="dbt-airflow-manifest-parser",
-    version="0.17.0",
+    name="dbt-airflow-factory",
+    version="0.18.0",
     description="Library to convert DBT manifest metadata to Airflow tasks",
     long_description=README,
     long_description_content_type="text/markdown",
@@ -41,7 +43,7 @@
     keywords="dbt airflow manifest parser python",
     author=u"Piotr Pekala",
     author_email="piotr.pekala@getindata.com",
-    url="https://github.com/getindata/dbt-airflow-manifest-parser/",
+    url="https://github.com/getindata/dbt-airflow-factory/",
     packages=find_packages(exclude=["ez_setup", "examples", "tests", "docs"]),
     include_package_data=True,
     zip_safe=False,
diff --git a/tox.ini b/tox.ini
index bdfa482..20a0f06 100644
--- a/tox.ini
+++ b/tox.ini
@@ -5,12 +5,13 @@ envlist = py37,py39
 extras =
     tests
 commands=
-    python -m pytest --cov dbt_airflow_manifest_parser --cov-report xml --cov-report term-missing --ignore=venv
+    python -m pytest --cov dbt_airflow_factory --cov-report xml --cov-report term-missing --ignore=venv
 
 # Lint
 [flake8]
 exclude = .git,__pycache__,docs/source/conf.py,old,build,dist
-max-line-length = 120
+max-line-length = 88
+extend-ignore = E203
 
 [mypy]
 no_strict_optional = True


Test patch:
diff --git a/tests/config/airflow_vars/airflow.yml b/tests/config/airflow_vars/airflow.yml
new file mode 100644
index 0000000..20cb0f3
--- /dev/null
+++ b/tests/config/airflow_vars/airflow.yml
@@ -0,0 +1,12 @@
+dags_path: "{{ var.value.dags_path }}"
+
+default_args:
+  owner: "{{ var.value.dags_owner }}"
+  email:
+    - "{{ var.value.email_owner }}"
+  depends_on_past: False
+  start_date: 2021-10-20T00:00:00
+  email_on_failure: False
+  email_on_retry: False
+  retries: 0
+  retry_delay: 5m
diff --git a/tests/config/base/airflow.yml b/tests/config/base/airflow.yml
index 4b452a6..1018f47 100644
--- a/tests/config/base/airflow.yml
+++ b/tests/config/base/airflow.yml
@@ -19,3 +19,4 @@ dag:
 seed_task: True
 manifest_file_name: ../tests/manifest.json
 use_task_group: True
+dags_path: "gs://example-bucket/dags/experimental-sandbox"
diff --git a/tests/test_airflow_vars.py b/tests/test_airflow_vars.py
new file mode 100644
index 0000000..92cc14d
--- /dev/null
+++ b/tests/test_airflow_vars.py
@@ -0,0 +1,36 @@
+from os import path
+from unittest import mock
+
+from dbt_airflow_factory.airflow_dag_factory import AirflowDagFactory
+
+from .utils import IS_FIRST_AIRFLOW_VERSION
+
+
+@mock.patch("airflow.models.variable.Variable")
+def test_airflow_vars_parsing(variable_mock):
+    vars_dict = {
+        "dags_path": "gs://some-interesting-bucket/dags/path",
+        "dags_owner": "Some Guy",
+        "email_owner": "dags.owner@example.com",
+    }
+
+    variable_mock.configure_mock(
+        **{"get_variable_from_secrets": lambda key: vars_dict.get(key)}
+    )
+
+    # given
+    factory = AirflowDagFactory(path.dirname(path.abspath(__file__)), "airflow_vars")
+
+    # when
+    if IS_FIRST_AIRFLOW_VERSION:
+        with mock.patch(
+            "airflow.models.variable.get_variable", lambda key: vars_dict.get(key)
+        ):
+            config = factory.read_config()
+    else:
+        config = factory.read_config()
+
+    # then
+    assert config["dags_path"] == vars_dict["dags_path"]
+    assert config["default_args"]["owner"] == vars_dict["dags_owner"]
+    assert config["default_args"]["email"] == [vars_dict["email_owner"]]
diff --git a/tests/test_config_propagation.py b/tests/test_config_propagation.py
index dbde82e..f0edbc3 100644
--- a/tests/test_config_propagation.py
+++ b/tests/test_config_propagation.py
@@ -27,10 +27,10 @@ def test_configuration():
         assert run_task.tolerations[0]["operator"] == "Equal"
         assert run_task.tolerations[0]["value"] == "data-processing"
         assert run_task.tolerations[0]["effect"] == "NoSchedule"
-        assert run_task.k8s_resources[0].limit_memory == "2048M"
-        assert run_task.k8s_resources[0].limit_cpu == "2"
-        assert run_task.k8s_resources[0].request_memory == "1024M"
-        assert run_task.k8s_resources[0].request_cpu == "1"
+        assert run_task.resources[0].limit_memory == "2048M"
+        assert run_task.resources[0].limit_cpu == "2"
+        assert run_task.resources[0].request_memory == "1024M"
+        assert run_task.resources[0].request_cpu == "1"
     else:
         assert run_task.node_selector == {"group": "data-processing"}
         assert run_task.tolerations[0].key == "group"
diff --git a/tests/test_dag_factory.py b/tests/test_dag_factory.py
index e9befae..2efab88 100644
--- a/tests/test_dag_factory.py
+++ b/tests/test_dag_factory.py
@@ -1,7 +1,7 @@
 from datetime import datetime, timezone
 from os import path
 
-from dbt_airflow_manifest_parser.airflow_dag_factory import AirflowDagFactory
+from dbt_airflow_factory.airflow_dag_factory import AirflowDagFactory
 
 from .utils import IS_FIRST_AIRFLOW_VERSION
 
diff --git a/tests/test_ephemeral_operator.py b/tests/test_ephemeral_operator.py
index 3c5ea95..0267e83 100644
--- a/tests/test_ephemeral_operator.py
+++ b/tests/test_ephemeral_operator.py
@@ -1,7 +1,7 @@
 from os import path
 
-from dbt_airflow_manifest_parser.airflow_dag_factory import AirflowDagFactory
-from dbt_airflow_manifest_parser.operator import EphemeralOperator
+from dbt_airflow_factory.airflow_dag_factory import AirflowDagFactory
+from dbt_airflow_factory.operator import EphemeralOperator
 from tests.utils import task_group_prefix_builder, test_dag
 
 
diff --git a/tests/test_tasks.py b/tests/test_tasks.py
index 8b22f20..80bfa3b 100644
--- a/tests/test_tasks.py
+++ b/tests/test_tasks.py
@@ -1,6 +1,6 @@
 from os import path
 
-from dbt_airflow_manifest_parser.builder_factory import DbtAirflowTasksBuilderFactory
+from dbt_airflow_factory.builder_factory import DbtAirflowTasksBuilderFactory
 
 from .utils import (
     IS_FIRST_AIRFLOW_VERSION,
diff --git a/tests/utils.py b/tests/utils.py
index 18ab96e..338eb44 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -6,7 +6,7 @@
 import airflow
 from airflow import DAG
 
-from dbt_airflow_manifest_parser.builder_factory import DbtAirflowTasksBuilderFactory
+from dbt_airflow_factory.builder_factory import DbtAirflowTasksBuilderFactory
 
 
 def manifest_file_with_models(nodes_with_dependencies):