mkt-connectors 1.1.9
Marketing connector
Requires Python: >=3.7,<4.0

![version 0.9.9](https://img.shields.io/badge/version-0.9.9-blue.svg "Version 0.9.9")
![python 3.7|3.8|3.9](https://img.shields.io/badge/python-3.7|3.8|3.9-yellowgreen.svg "Python 3.7|3.8|3.9")

# Marketing Connectors (Snowflake & Monitor Logs)

Marketing Connector is built for Marketing Engineers helps connect to Snowflake system. In addition, Jobs/Steps can be
logged utilizing Monitor Class.

## Setup

### Installation

There are a couple of way of installing the package. Installing from our private pypi is the preferred way.

If you have sure your `~/.pip/pip.conf` setup to use our private pypi server: http://10.112.60.35/simple/ simply:

```bash
$ pip install mkt-connectors
```

If you don't have a `~/.pip/pip.conf` use:

```bash
$ pip install mkt-connectors --extra-index-url=http://10.112.60.35/simple/ --trusted-host=10.112.60.35
```

In both case make sure that your either using `vpn2` or `AppGate`. The private PyPi is only reachable within chewy's network.

### Environments

The applications comes with 2 different environments and each environment connects to a different `snowflake` database:

- `dev` --> `EDLDB_DEV`
- `prd` --> `EDLDB`


### Usage Example

Several connector classes are provided which receive information parameters from the source 
modules and build a configuration for the job/task. Steps are as follows: 

```python
class SnowflakeDatabase:
    def __init__(self, config):
        self.config = config

        self.snowflake_engine = create_engine(
            f"snowflake://{self.config.snowflake_username}:{self.config.snowflake_password}"
            f"@{self.config.snowflake_account}/"
            f"{self.config.snowflake_database}&role={self.config.snowflake_role}"
        )
```

Above example takes the information from source object and connects to snowflake database for ETL process.

### A Simple Example

```python
from mkt_connectors.secret import get_secret
from mkt_connectors.snowflake import SnowflakeDatabase
from mkt_connectors.monitor import Monitor
from mkt_connectors.config import configure

snowflake_database = "snowflake_database" 
get_secret("secret_name")
config = configure(conf_options={})
snowy = SnowflakeDatabase(config)
monitoring = Monitor(config)
```

```python
class Monitor:
    def __init__(self, config):
        self.config = config
        self.snowflake_database = SnowflakeDatabase(config)
        self.rec_list_success = f"{','.join(config.REC_LIST)},{str(config.SLACK_SUCCESS)}"
        self.rec_list_failure = f"{','.join(config.REC_LIST)},{str(config.SLACK_FAILURE)}"

    def insert_step_metadata(self, step_start_date, step_start_time, process_name, step_name):
        self.snowflake_database.insert_step_metadata(
            step_start_date=step_start_date, step_start_time=step_start_time, process_name=process_name,
            user_name=self.config.snowflake_username, step_name=step_name,
        )

    def update_step_metadata(self, step_start_date, step_start_time, process_name, step_name,status,
                             step_end_time="null", time_taken="null", notes=json.dumps("None")):
        self.snowflake_database.update_step_metadata(
            step_start_date=step_start_date,
            step_start_time=step_start_time, process_name=process_name,
            step_name=step_name, step_end_time=step_end_time, status=status, time_taken=time_taken,
            notes = notes

        )
        if status == 'Failed':
            self.step_failure_notification(process_name,step_name,notes)

    def step_failure_notification(self, process_name, step_name, error):
        """
        Update metadata table with a success for the particular step within the pipeline
        Args:
            process_name: Name of the process
            step_name: Name of the Step withing the process
            error: Error seen in the step/process
            Returns: None. Sends out notification and updates the metadata table within Snowflake with the error data.
        """

        error = str(error)
        body = f'''
                    Process Name  --> {process_name}
                    Step Name      --> {step_name}
                    Execution Time --> {datetime.now()}
                    Error          --> {error}
               '''
        resp = requests.post(
            self.config.notification_url,
            data={"service": "email",
                  "type": "info",
                  "recipients": f"{self.config.SLACK_FAILURE}",
                  "subject": f'Failure in running the step {step_name} in the {process_name} process at {datetime.now()}',
                  "body": f'{body}'}
        )

    def insert_process_metadata(self, process_start_date, process_start_time, process_name):
        self.snowflake_database.insert_process_metadata(process_start_date=process_start_date,
                                                        process_start_time=process_start_time,
                                                        process_name=process_name,)

    def update_process_metadata(self, process_start_date, process_start_time, process_name, status,
                                process_end_time="null", notes=json.dumps("None"), time_taken="null",
                                attachment=None,attachment_name=None):
        self.snowflake_database.update_process_metadata(
            process_start_date=process_start_date, process_start_time=process_start_time,
            process_name=process_name, status=status, process_end_time=process_end_time, time_taken=time_taken,
            notes=notes

        )

        if status == 'Failed':
            self.process_failure_alert_notification(process_name, notes)

        else:
            self.process_success_alert_notification(process_name,notes,attachment,attachment_name)

    def process_failure_alert_notification(self, process_name, error):
        """
        Args:
            notification_url: API notification URL
            rec_list: List of users getting the notification.
            pipeline_name: Pipeline Name
            step_name: Step Name
            error: Error

        Returns:
            None. Sends notification and updates the metadata table on snowflake.

        """
        error = str(error)
        body = f'''
                    Pipeline Name  --> {process_name}
                    Execution Time --> {datetime.now()}
                    Error          --> 
                    The process {process_name} failed with the error --> {error}
               '''
        resp = requests.post(
            self.config.NOTIFICATION_URL,
            data={"service": "email",
                  "type": "info",
                  "recipients": f"{self.config.SLACK_FAILURE}",
                  "subject": f'Failure in running the process_name {process_name} at {datetime.now()}',
                  "body": f'{body}'
                  })

    def process_success_alert_notification(self, process_name, notes, attachment=None, attachment_name=None,):
        message_body = f'''
            Process Name    --> {process_name}
            Execution Time  --> {datetime.now()}
            Message         --> {notes}
        '''
        if attachment is not None:
            files = {"attachment": (attachment_name, attachment)}

            resp = requests.post(
                self.config.NOTIFICATION_URL,
                data={"service": "email",
                      "type": "info",
                      "recipients": f"{self.config.SLACK_SUCCESS}",
                      "subject": f'Process {process_name} ran successfully',
                      "body": f'{message_body}'},
                files = files
            )
        else:
            resp = requests.post(
                self.config.NOTIFICATION_URL,
                data={"service": "email",
                      "type": "info",
                      "recipients": f"{self.config.rec_list_success}",
                      "subject": f'Process {process_name} ran successfully',
                      "body": f'{message_body}'})

    ''' below functions are checking process status and data validations'''

    def process_start_check(self, notification_url, rec_list, process_name, process_start_time):
        body = f'''
            Process Name    --> {process_name}
            Process Expected Start Time   --> {process_start_time}
        '''
        resp = requests.post(
            notification_url,
            data={"service": "email",
                  "type": "info",
                  "recipients": f"{rec_list}",
                  "subject": f'Process {process_name} did not start at its scheduled time',
                  "body": f'{body}'}
        )

    def process_overtime_check(self, notification_url, rec_list, process_name, process_start_time):
        body = f'''
            Process Name    --> {process_name}
            Process Start Time --> {process_start_time}
        '''
        resp = requests.post(
            notification_url,
            data={"service": "email",
                  "type": "info",
                  "recipients": f"{rec_list}",
                  "subject": f'Process {process_name} is taking more time than usual',
                  "body": f'{body}'}
        )

    def process_multiple_runs_check(self,notification_url, rec_list, process_name):
        body = f'''
            Process Name    --> {process_name}
        '''
        resp = requests.post(notification_url, data={"service": "email",
                                                     "type": "info",
                                                     "recipients": f"{rec_list}", "subject": f'Process {process_name} has more than once instance running',
                                                     "body": f'{body}'}
        )
```

The above class Monitor methods can be utilized to keep track of Jobs and steps ran in the ETL process.