# Snowflake connector pattern
This example shows complete pattern (how to store and load SQL, how to validate and bind params to avoid SQL injections, and how to extend base class with the flow specific methods).

View the [documentation in notion](https://www.notion.so/Snowflake-7c88cc17b0034e7db669a88fd2962bab) for more info.

## Installation
Add the following lines: 
```
-i https://pypi.theorchard.io/pypi/
snowflake_connector_etl
```
to `requirements.txt`. Optionally you can fix the version.

## Usage
The `etl_connector` module provides you with two classes: `SnowflakeSQLExecutor` and `SQLLoader`. Also there is the `BaseValidator` class in `validator.py` module. 

### `SQLLoader`
`SQLLoader` is optional to use, it's just a convenient helper to store SQL queries as plain text `.sql` files, automatically load them, and access them via the get_query() method. 

*Please add meaningful comment/docstring in the beginning of each .sql file!*

SQLLoader enforces usage of `/queries` directory for `.sql` files.
```
/sample_swf_workflow/
    /queries/
        sf_count_before_start_period.sql
    snowflake_executor.py  # place sql_loader = SQLLoader(__file__)) there 
```
```python
# place this to the snowflake_executor.py file
sql_loader = SQLLoader(__file__))

# get query by its filename
sql_loader.load_query('sf_count_before_start_period')
```

#### Data binding and SQL injections
Please use proper binding (e.g., `%(param_name)s`) for params that can be binded:
```
SELECT COUNT(*)
FROM {db}.{schema}.{table}
WHERE {date_col} < %(start_period)s;
# these params will be binded via the cursor.execute() method, 
so it is not necessary to validate the parameters
params_to_bind = {'start_period': start_period}
self.fetchone(partially_parametrized_sql_template, params_to_bind)
```
Unfortunately, we can't use binding for db identifiers, like db, schema, table, column names, etc., because binding adds single quotes around a value, and this is an invalid syntax for identifiers. So we must hardcode them, or use standard string intepolation (`.format()` method). Because of that we strongly recommend to use custom **VALIDATION for all the identifiers** that we can't bind in order to avoid SQL injection. In order to do so you can use methods of `BaseValidator` class, or create a custom validator class which inherits from `BaseValidator` and extends it with some specific methods.

##### MUST BE BINDED/VALIDATED
* Any SQL parameter
* Any string used in SQL that is passed into a SWF workflow
** ex. COPY INTO command options
* Any string used in SQL that is submitted via an API
** POST/PUT body
** URL params
* Any string used in SQL that is submitted via a user form

##### OPTIONAL TO VALIDATE
* db identifiers that are hard coded in a config to a trusted string
 
We'll be monitoring this [issue](https://github.com/snowflakedb/snowflake-connector-python/issues/10) and update our connector if `snowflake-connector-python` dev team propose a solution. For now we can use custom validation: 
```python
# format SQL template with the 'table' identifier which was validated by the
# method of the custom validator instance
# (self.sf_config values were already checked for bad symbols within __init__()
# method of the `SnowflakeSQLExecutor` class)
sql_template = (
    sql_loader.get_query('sf_count_before_start_period').format(
        db=self.sf_config['db'], 
        schema=self.sf_config['schema'], 
        table=validator.validate_table(table))
```

##### COPY statements
COPY INTO (table from location | location from table) SQL statements are extremely versatile. That's why it's diffucult to propose a generic solution for safe arguments binding and validation. The rule of thumb is following: use binding feature of .execute() method for all the strings, except of FILE_FORMAT params: they should be validated via `validate_file_format()` method of the `BaseValidator` class or its inheritor. Unfortunately, we can't bind integers, only %(name)s is working, but not %(name)d. So, e.g., you can put into the `\queries` directory an SQL file with the following template:
```sql
COPY INTO {db}.{schema}.{table}
FROM %(s3_location)s
CREDENTIALS = (AWS_KEY_ID = %(AWS_KEY_ID)s AWS_SECRET_KEY = %(AWS_SECRET_KEY)s 
  FILES = ( %(file_name1)s, %(file_name2)s )
  PATTERN = %(regex_pattern)s
  FILE_FORMAT = ( {{ TYPE = CSV {file_format} }} )
  ON ERROR = CONTINUE 
  TRUNCATECOLUMNS = FALSE
```
In the example above we hard code copy options (`ON ERROR = CONTINUE TRUNCATECOLUMNS = FALSE FORCE = TRUE`) to the template.

* Use string interpolation for params, which were validated by `Validator` class.
** Don't forget to use double curly braces, if you want them to stay after Python string interpolation (`FILE_FORMAT = ( {{ TYPE = CSV {file_format} }} )`) in the example above).
* Use data binding for all the stings (e.g., `%(s3_location)s`).
* Hard code options which are not strings. Probably they won't change ever anyway.


##### Before deploy to production
Warning! Don't forget to add `\queries` directories to setup.py:
```python
setup(
    # some lines...
    package_data={
        # include all the SQL files
        'feed_ingestion.flows': ['*/queries/*.sql'],
        'feed_ingestion.common.fact_analytics_sf': ['queries/*.sql'],
        'feed_ingestion.common.staging_raw_sf': ['queries/*.sql']
    },
    # some lines...
)
```

### `SnowflakeSQLExecutor` 
`SnowflakeSQLExecutor` is a base class, which can be used as a context manager. It performs basic validation of the sf_config dict values (which is a dict with credentials), manages connection and cursor, and implements the subset of cursor methods accordingly to PEP 249 (Python Database API Specification v2.0): `execute`, `fetchone`, `fetchall`, and `fetchmany` (fetch methods basically combine `execute` with a standard fetch* action). It can be used in a really simple way, e.g.:
```python
from snowflake_connector.etl_connector import SnowflakeSQLExecutor

with SnowflakeSQLExecutor(sf_config) as sf_executor:
    count = sf_executor.fetchone(sql, params=params)[0]
```

It also could become a base class for a workflow-specific executor. In this case it helps to encapsulate business logic to custom methods. First, import `SnowflakeSQLExecutor` in the `snowflake_executor.py` file, which should be placed on the same level as `flow.py`, `tasks.py`, and `/queries` directory.

E.g.:
```python
# snowflake_executor.py
from snowflake_connector.etl_connector import SnowflakeSQLExecutor
from snowflake_connector.etl_connector import SQLLoader

from sample_swf_workflows.flows.sample_swf_workflow import validators

# Load SQL templates
sql_loader = SQLLoader(__file__)


class SnowflakeSQLExecutorSSW(SnowflakeSQLExecutor):
    """Helper class to abstract Snowflake operations.

    'SSW' in 'SnowflakeSQLExecutorSSW' stands for 'sample_swf_workflow'.

    This class inherits from SnowflakeSQLExecutor class, which provides basic
    set of methods. This class extends SnowflakeSQLExecutor with some specific
    methods, which are useful to encapsulate some flow specific operations.
    """
    
    def __init__(self):
        self.validator = validators.SampleValidator()

    def get_row_count_before_start_period(self, table, date_col):
        """Get count of rows from the Snowflake table.

        This is count of rows which have timestamp (e.g. processeddaytime)
        before start_period.

        WARNING! Args table and date_col MUST BE VALIDATED to prevent SQL
        injections.

        Args:
            table (str): Table name in Snowflake.
            date_col (str): Name of the date_col of table used by incremental
                of date_range sync.

        Returns:
            int: Row count of the specified Snowflake table.
        """
        sql_template = sql_loader.load_query('sf_count_before_start_period')
        params = dict(
            db=self.sf_config['db'],
            schema=self.sf_config['schema'],
            table=table,
            date_col=date_col)
        sql, non_identifier_params = self.validator.format_identifiers(sql_template, params)
        return self.fetchone(sql_template, params=non_identifier_params)[0]     

# tasks.py
with snowflake_executor.SnowflakeSQLExecutorSSW(sf_config) as sf_executor:
    snowflake_row_count = (
        sf_executor.get_row_count_before_start_period(table, date_col))
```

In that example above we're using all the powers of new Snowflake connector pattern.

## Working locally with SSH Keys

Using a username/password combo locally requires performing MFA for each request.
This can be bypassed by using a known SSH key to access snowflake from your
local machine. Follow the instructions in the notion doc linked at the top of
this README to register your SSH if you haven't done so yet.

To connect via your ssh key, you'll need to pass the `private_key` option to
the connector. Below is an example that allows for ssh-key based access during
local development.

```python
from snowflake_connector.etl_connector import SnowflakeSQLExecutor
from snowflake_connector.private_key import get_private_key

SNOWFLAKE_OPTIONS = {
    'role': os.environ.get('SNOWFLAKE_ROLE'),
    'warehouse': os.environ.get('SNOWFLAKE_WAREHOUSE'),
    'user': os.environ.get('SNOWFLAKE_USERNAME'),
    'password': os.environ.get('SNOWFLAKE_PASSWORD'),
    'account': os.environ.get('SNOWFLAKE_ACCOUNT'),
    'db': 'dbname',
    'schema': 'schema_name,
    'private_key': None,
}

SNOWFLAKE_PRIVATE_KEY_PATH = os.environ.get('SNOWFLAKE_PRIVATE_KEY_PATH', '')
SNOWFLAKE_KEY_PASSPHRASE = os.environ.get('SNOWFLAKE_KEY_PASSPHRASE', '')
if ENVIRONMENT == 'dev' and SNOWFLAKE_KEY_PASSPHRASE:
    if not SNOWFLAKE_PRIVATE_KEY_PATH:
        SNOWFLAKE_PRIVATE_KEY_PATH = '{}/.ssh/snowflake/rsa_key.p8'.format(
            os.environ['HOME'])
    SNOWFLAKE_OPTIONS['private_key'] = get_private_key(
        SNOWFLAKE_PRIVATE_KEY_PATH,
        SNOWFLAKE_KEY_PASSPHRASE)

with SnowflakeSQLExecutor(config.SNOWFLAKE_OPTIONS) as sf_executor:
    # ... use the connector
```