# Airflow

## Overview

Airflow manages the execution of **DAGS** (Directed acyclical graphs), meaning workflows executed as
branching steps on a graph that follow an order from top left to bottom right, without looping backward

![DAG](https://upload.wikimedia.org/wikipedia/commons/c/c6/Topological_Ordering.svg)

### DAG Structure

ABACUS related dags can be found in the root of the [dags](./dags) directory.

Each DAG consists of various **operators** that are executed in a particular order set by the developers.

Many of these operators are `PythonOperators`, which can execute python code. The python code for these operators are located in the [tasks](./dags/tasks) directory.

(The tasks directory has sub-directories that coincide with each DAG.)

For example, the [example_lambda](./dags/example_lambda.py) DAG has a python operator, `invoke_example_lambda_operator`. This operator calls a task located in the `example_lambda` [subdirectory](./dags/tasks/example_lambda/invoke_example_lambda).

### Hooks

This repo makes use of a few important hooks to allow airflow to interact with AWS (S3, Lambdas) and Snowflake.

#### OrchLambdaHook
The [OrchLambdaHook](./dags/hooks/lambda_hook.py) wraps the existing airflow `AWSLambdaHook` and is used to invoke lambdas within The Orchard's domain.

```python
hook = OrchLambdaHook('name-of-lambda')
response = hook.invoke_lambda({'json': true})
```

#### ArtRelationsMySqlHook
The [ArtRelationsMySqlHook](./dags/hooks/art_relations_mysql_hook.py) wraps the existing airflow `MySQLHook` and is used to interact with the `art_relations` database using the `ART_RELATIONS_CONNECTION_URI` by default. It is primarily used to read/write data to the `publishing_escrow` table.

```python
hook = ArtRelationsMySqlHook()
sql_statement = 'SELECT * FROM publishing_escrow;'
hook.run(sql_statement)
```

#### RoyaltySnowflakeHook
The [RoyaltySnowflakeHook](./dags/hooks/royalty_snowflake_hook.py) wraps the existing airflow `SnowflakeHook` and is used to interact with Snowflake. It is used to write/read Snowflake data.

```python
hook = RoyaltySnowflakeHook(snowflake_conn_id=constants.SNOWFLAKE_CONN_NAME)
sql_statement = 'sql statement to select, copy, or insert snowflake data'
hook.run(sql_statement, autocommit=True)
```

## Requirements
- Python 3.11
- python3-dev
- gcc
- libsasl2-dev

## Developing

Follow AWS Access guide [here](https://www.notion.so/AWS-Access-f841b9dd815d4443a80e96a86c92cd2f#f11f08d09f16493a8a3e9dbf38d87164), including configuration of the [ECR Docker Credential Helper](https://www.notion.so/AWS-Access-f841b9dd815d4443a80e96a86c92cd2f?pvs=4#c4ec404995784b5bb0f7f2c5b2f74886)

To run tests and integrate with your IDE locally, install dependencies locally with:

```bash
make dev_env

make test
```

(Note the pipeline and PR job run containerized linting & unit tests via `make lint_unit_dockerized`)

You'll have to do some setup prior to spinning up the docker containers.

Please note for `awsume` you need to set up your `~/.aws/config` to look similar to this:

```editorconfig
[profile prod]
region = us-east-1
output = json
mfa_serial = arn:aws:iam::<PROD_ACCOUNT_ID>:mfa/<USERNAME>
manager = awsume

[profile shared]
source_profile = prod
role_session_name = <USERNAME>
role_arn = arn:aws:iam::<SHARED_ACCOUNT_ID>:role/generic-engineer-role
manager = awsume
```
For more context - check out this [notion page](https://www.notion.so/AWS-Access-f841b9dd815d4443a80e96a86c92cd2f?pvs=4#2796c2b2b6a349d2b193941db9165932)

To set up the scheduler and webserver as Docker containers - do the following:
```bash
# Paste your credentails to your `.env` file from the output below
awsume prod -s 

# Gain access to the shared account to download the docker image

awsume shared

aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 086679231553.dkr.ecr.us-east-1.amazonaws.com

# start docker containers
make dev

# spin down docker containers
make docker_down

# rebuild docker containers
make docker_dev_build
```

> Mac OS X Note: Prior to running docker-compose, ensure the airflow home directory (or a parent directory)
> is added to your docker file sharing paths under Docker Preferences >> Resources >> File Sharing

After a few seconds you should see the airflow UI at [http://localhost:6103](http://localhost:6103).

**NOTE:** If the local airflow UI asks you to log in, the username and password are both `admin`.

From the UI you can trigger DAG runs and view logs.
Updates made inside of the `dags` directory will be picked up every few seconds by the scheduler.

### Snowflake

Snowflake developer accounts require MFA, which disrupts the flow of programmatic calls.
To get around this, we use whitelisted ssh keys while making calls.
Create your own key using the instructions [found here](https://docs.snowflake.net/manuals/user-guide/snowsql-start.html#using-key-pair-authentication) and then open a [DevOps Support](https://www.notion.so/DevOps-End-User-Support-Process-fae6dcaca43e48e28bf9f601bdaa7722) ticket for them to add it to your Snowflake user.

Make sure to uncomment the line in `docker-compose.dev.yaml` that will mount your snowflake private key in the container.

### Testing dags and operators locally (outside of Docker)

Initialize a local environment & database and run the task:

> Running the **airflow** command while `AIRFLOW_HOME` is not set will create a directory at `~/airflow` that should be deleted.

```bash
make dev_env

make airflow.db

# airflow tasks test <dag-name> <task-name> <start-time> -t {task configuration json}
env/bin/airflow tasks test preprocess start_preprocess_job_operator 2017-03-18T18:00:00.0 -t "$(< tests/mock_event.json)"
```

### Deleting DAG

- Delete the DAG file from `/dags` and related subtasks from `/tasks` directory.
- Delete the test-cases for DAG from `/tests` directory.
- Also delete DAG related variables from `constants`, `.env.shadow`, `.flake8` and `/lib/config` files.
- Once PR is merged, `aws s3 sync  --delete` command from [ows-royalties-workflows-airflow-deploy](https://pipeline.theorchard.io/job/ows-royalties-workflows-airflow-deploy/)job should delete files that exists in destination(i.e S3) but not in the [source code](https://github.com/theorchard/ows-royalties-workflows/tree/master/dags) directory.

(Note: In case DAG doesn't get deleted from the Airflow Webserver UI, we can delete it manually from the UI using the delete button (x in circle))