pepkordna 0.111
Utils and airflow operators for Pepkor DNA data engineering team
Requires Python: >=3.7

## pepkordna

Utils and airflow operators for Pepkor DNA data engineering team. The Utils module can be used to standardize any frequently used code. 

## Installation

```
pip install pepkordna
```

## Usage

## Audit Operators

To create a new operator, use the [AuditBaseOperator](pepkordna/pepkordna/operators/audit_base_operator.py) class to inherit from. Look at example [AuditDuplicates](pepkordna/pepkordna/operators/audit_duplicates.py) operator which can be copied and renamed to create a new operator. The aud_sql variable and aud_sql string formattting variables can then just be changed to the new required sql.

## Audit Duplicates Operator

[AuditDuplicates](pepkordna/pepkordna/operators/audit_duplicates.py) 

This operator is used to check for duplication on a table based on the parameters passed in.
Below task will check for duplicates on table dna-data-prod.test.customers using customer_guid for date value 2022-05-31.

```
test_aud_dups = AuditDuplicates(
    task_id='test_aud_dups',
    results_table_project_id='gcp_project_id',
    bigquery_conn_id='default_conn_id',
    source_table='gcp_project_id.dataset_id.table_name',
    cols=['customer_id','snapshot_date'],
    partition_col='snapshot_date',
    partition_col_value_start='2022-01-31',
    partition_col_value_end='2022-05-31'
)
```

## Utils - Alerting

All alerting processes.

[Alerting - Gitlab](pepkordna/pepkordna/utils/utils.py) 

## Log to Gitlab

Log an incident to gitlab. 

```
Alerting().log_to_gitlab(
            git_api_token='123',
            airflow_git_api_key_conn_id='default_git_con_id',
            git_project_id='123',
            git_project_area='issues',
            git_issue_type='issue',
            git_request_label='Backlog',
            git_issue_description='Process failed with exception xyz',
            git_issue_title='Process that loads files failed')
```

## Send Google chat

Send a message to a google chatroom. 

See class Status for all possible statuses. 

```
Alerting().send_gchat(
            process_status=Status.FAILED,
            project_name='google-cloud-project-name',      
            extra_event_info='Process failed with exception xyz',
            process_name='File Loader',
            data_engineers=[{"user_name": "Name Surname","user_id": "1234567890"}],    
            webhook_url='https://chat.googleapis.com/',
            log_url='wwww.airflow_log_url_here.com')
```    

## Send Sengrid email

Send email using Sendgrid API.

```
Alerting().send_sendgrid_email(
            sendgrid_api_key='api_key',
            email_body='test',
            email_to_list=['user@email.com'],
            email_cc_list=['user@email.com'],
            email_subject='Test') 
```        

## Utils - Airflow Callbacks

Default standardized on_failure and on_success callbacks for airflow.

[Alerting - Gitlab](pepkordna/pepkordna/utils/utils.py) 

On FAILURE, the below example will send an email, google chat message and log a ticket to gitlab. Use whichever alerting needed.

```
def default_failure_callback(context):
    AirflowCallbacks(context).default_on_failure_callback(
        email_to_list=['user@pepkor-dna.com'],        
        email_cc_list=['user@pepkor-dna.com'],   
        data_engineers=[{"user_name": "Name Surname","user_id": "1234567890"}],               
        git_project_id='123456',
        airflow_gchat_webhook_conn_id='default_conn_id',
        airflow_git_api_key_conn_id='default_conn_id',
        airflow_sendgrid_api_key_conn_id='default_conn_id',
        send_gchat=True,
        send_gitlab=True,           
        send_email=True)

with models.DAG(dag_name...):

    test = PythonOperator(
        provide_context=True,        
        on_failure_callback=default_failure_callback,
        ...
    )          
```

On SUCCESS, the below example will send an email and send a google chat message. Use whichever alerting needed.

```
def default_success_callback(context):
    AirflowCallbacks(context).default_on_success_callback(
        email_to_list=['user@pepkor-dna.com'],        
        email_cc_list=['user@pepkor-dna.com'],
        data_engineers=[{"user_name": "Name Surname","user_id": "1234567890"}],        
        airflow_gchat_webhook_conn_id='default_conn_id',
        airflow_sendgrid_api_key_conn_id='default_conn_id',
        send_gchat=True,           
        send_email=True)

with models.DAG(dag_name...):

    test = PythonOperator(
        provide_context=True,        
        on_success_callback=default_success_callback,
        ...
    )          
```

## Contributors
Dean Lambrechts - deanl@pepkor-dna.com