Snowflake

Opinionated Snowflake cursor for logging Snowflake execution calls with parameter obfuscation options.

from snowflake.connector import connect
from prefecto.ext.snowflake import LogCursor

with connect(...).cursor(LogCursor) as cursor:
    cursor.execute(
        "SELECT * FROM table WHERE id = %(id)s AND secret = %(secret)s",
        params={"id": 123, "secret": "shhh"},
        obfuscate_params=["secret"],
        command_id="secret-selection",
    )

INFO - [secret-selection] Beginning command.
DEBUG - [secret-selection] Executing command:
SELECT * FROM table WHERE id = 123 AND secret = ****
INFO - [secret-selection] Command executed successfully.

CommandLogAdapter

Bases: LoggerAdapter

A logging adapter that prepends the command ID to the log messages.

LogCursor

Bases: SnowflakeCursor

A Snowflake cursor that logs command executions. The execute method has additional parameters for altering log behavior.

Example:

```python
from snowflake.connector import connect
from prefecto.ext.snowflake import LogCursor

c = connect(...).cursor(LogCursor)
c: LogCursor
c.execute("SELECT * FROM table")
```

Produces logs like

```txt
INFO - [rustic-human] Beginning command.
DEBUG - [rustic-human] Executing command:
SELECT * FROM table
INFO - [rustic-human] Command executed successfully.
```

execute(command, params=None, _bind_stage=None, timeout=None, _exec_async=False, _no_retry=False, _do_reset=True, _put_callback=None, _put_azure_callback=None, _put_callback_output_stream=sys.stdout, _get_callback=None, _get_azure_callback=None, _get_callback_output_stream=sys.stdout, _show_progress_bar=True, _statement_params=None, _is_internal=False, _describe_only=False, _no_results=False, _is_put_get=None, _raise_put_get_error=True, _force_put_overwrite=False, file_stream=None, *, logger=None, command_id=None, obfuscate_params=False, level=logging.DEBUG)

Executes a command on the Snowflake connection after logging it.

Parameters:
  • command (str) –

    The command to execute.

  • params (Sequence[Any] | dict[Any, Any] | None, default: None ) –

    The parameters to pass to the c.execute. Defaults to None.

  • _bind_stage (str | None, default: None ) –

    The stage to bind. Defaults to None.

  • timeout (int | None, default: None ) –

    The timeout to use. Defaults to None.

  • _exec_async (bool, default: False ) –

    Whether to execute asynchronously. Defaults to False.

  • _no_retry (bool, default: False ) –

    Whether to retry on failure. Defaults to False.

  • _do_reset (bool, default: True ) –

    Whether to reset the connection. Defaults to True.

  • _put_callback (SnowflakeProgressPercentage, default: None ) –

    The callback for PUT operations. Defaults to None.

  • _put_azure_callback (SnowflakeProgressPercentage, default: None ) –

    The callback for Azure PUT operations. Defaults to None.

  • _put_callback_output_stream (IO[str], default: stdout ) –

    The output stream for PUT callbacks. Defaults to sys.stdout.

  • _get_callback (SnowflakeProgressPercentage, default: None ) –

    The callback for GET operations. Defaults to None.

  • _get_azure_callback (SnowflakeProgressPercentage, default: None ) –

    The callback for Azure GET operations. Defaults to None.

  • _get_callback_output_stream (IO[str], default: stdout ) –

    The output stream for GET callbacks. Defaults to sys.stdout.

  • _show_progress_bar (bool, default: True ) –

    Whether to show a progress bar. Defaults to True.

  • _statement_params (dict[str, str] | None, default: None ) –

    The statement parameters. Defaults to None.

  • _is_internal (bool, default: False ) –

    Whether the command is internal. Defaults to False.

  • _describe_only (bool, default: False ) –

    Whether to describe only. Defaults to False.

  • _no_results (bool, default: False ) –

    Whether to return no results. Defaults to False.

  • _is_put_get (bool | None, default: None ) –

    Whether the command is a PUT or GET operation. Defaults to None.

  • _raise_put_get_error (bool, default: True ) –

    Whether to raise an error on PUT or GET failure. Defaults to True.

  • _force_put_overwrite (bool, default: False ) –

    Whether to force PUT overwrite. Defaults to False.

  • file_stream (IO[bytes] | None, default: None ) –

    The file stream to use. Defaults to None.

  • logger (Logger | None, default: None ) –

    The logger to use.

  • command_id (str, default: None ) –

    The ID to use in the logs. Generates a random one if not provided. Defaults to None.

  • obfuscate_params (bool | str | list[str], default: False ) –

    Whether to obfuscate the parameters in the logs. If True, obfuscates all parameters. If a string or list of strings, obfuscates only the specified parameters. Defaults to False.

  • level (int | str, default: DEBUG ) –

    The logging level to use. Defaults to logging.DEBUG.
Returns:
  • SnowflakeCursor | dict[str, Any] | None –

    SnowflakeCursor | dict[str, Any] | None: The result of the c.execute.

Example:

```python
import logging
from snowflake.connector import connect
from prefecto.ext.snowflake import LogCursor

c = connect(...).cursor(LogCursor)
c.execute("SELECT * FROM table")
```

Produces logs like

```txt
INFO - [rustic-human] Beginning command.
DEBUG - [rustic-human] Executing command:
SELECT * FROM table
INFO - [rustic-human] Command executed successfully.
```

And tracebacks like

```txt
ERROR - [rustic-human] Command failed.
Traceback (most recent call last):
...
snowflake.connector.errors.ProgrammingError: ...
[rustic-human] Command failed
SELECT * FROM table
```

You can also obfuscate parameters:

```python
c.execute("SELECT * FROM table WHERE id = %(id)s", params={"id": 123}, obfuscate_params=True)
c.execute("SELECT * FROM table WHERE id = %(id)s AND %(secret)s", params={"id": 123, "secret": "shhh"}, obfuscate_params=["secret"])
```

```txt
INFO - [delightful-octopus] Beginning command.
DEBUG - [delightful-octopus] Executing command:
SELECT * FROM table WHERE id = ****
INFO - [delightful-octopus] Command executed successfully.
INFO - [super-athlete] Beginning command.
DEBUG - [super-athlete] Executing command:
SELECT * FROM table WHERE id = 123 AND secret = ****
INFO - [super-athlete] Command executed successfully.
```

execute(c, command, params=None, _bind_stage=None, timeout=None, _exec_async=False, _no_retry=False, _do_reset=True, _put_callback=None, _put_azure_callback=None, _put_callback_output_stream=sys.stdout, _get_callback=None, _get_azure_callback=None, _get_callback_output_stream=sys.stdout, _show_progress_bar=True, _statement_params=None, _is_internal=False, _describe_only=False, _no_results=False, _is_put_get=None, _raise_put_get_error=True, _force_put_overwrite=False, file_stream=None, *, logger=None, command_id=None, obfuscate_params=False, level=logging.DEBUG)

Executes a command on the Snowflake connection after logging it.

Parameters:
  • c (SnowflakeCursor) –

    The Snowflake cursor or connection to use.

  • command (str) –

    The command to execute.

  • params (Sequence[Any] | dict[Any, Any] | None, default: None ) –

    The parameters to pass to the c.execute. Defaults to None.

  • _bind_stage (str | None, default: None ) –

    The stage to bind. Defaults to None.

  • timeout (int | None, default: None ) –

    The timeout to use. Defaults to None.

  • _exec_async (bool, default: False ) –

    Whether to execute asynchronously. Defaults to False.

  • _no_retry (bool, default: False ) –

    Whether to retry on failure. Defaults to False.

  • _do_reset (bool, default: True ) –

    Whether to reset the connection. Defaults to True.

  • _put_callback (SnowflakeProgressPercentage, default: None ) –

    The callback for PUT operations. Defaults to None.

  • _put_azure_callback (SnowflakeProgressPercentage, default: None ) –

    The callback for Azure PUT operations. Defaults to None.

  • _put_callback_output_stream (IO[str], default: stdout ) –

    The output stream for PUT callbacks. Defaults to sys.stdout.

  • _get_callback (SnowflakeProgressPercentage, default: None ) –

    The callback for GET operations. Defaults to None.

  • _get_azure_callback (SnowflakeProgressPercentage, default: None ) –

    The callback for Azure GET operations. Defaults to None.

  • _get_callback_output_stream (IO[str], default: stdout ) –

    The output stream for GET callbacks. Defaults to sys.stdout.

  • _show_progress_bar (bool, default: True ) –

    Whether to show a progress bar. Defaults to True.

  • _statement_params (dict[str, str] | None, default: None ) –

    The statement parameters. Defaults to None.

  • _is_internal (bool, default: False ) –

    Whether the command is internal. Defaults to False.

  • _describe_only (bool, default: False ) –

    Whether to describe only. Defaults to False.

  • _no_results (bool, default: False ) –

    Whether to return no results. Defaults to False.

  • _is_put_get (bool | None, default: None ) –

    Whether the command is a PUT or GET operation. Defaults to None.

  • _raise_put_get_error (bool, default: True ) –

    Whether to raise an error on PUT or GET failure. Defaults to True.

  • _force_put_overwrite (bool, default: False ) –

    Whether to force PUT overwrite. Defaults to False.

  • file_stream (IO[bytes] | None, default: None ) –

    The file stream to use. Defaults to None.

  • logger (Logger | None, default: None ) –

    The logger to use.

  • command_id (str, default: None ) –

    The ID to use in the logs. Generates a random one if not provided. Defaults to None.

  • obfuscate_params (bool | str | list[str], default: False ) –

    Whether to obfuscate the parameters in the logs. If True, obfuscates all parameters. If a string or list of strings, obfuscates only the specified parameters. Defaults to False.

  • level (int | str, default: DEBUG ) –

    The logging level to use. Defaults to logging.DEBUG.
Returns:
  • SnowflakeCursor | dict[str, Any] | None –

    SnowflakeCursor | dict[str, Any] | None: The result of the c.execute.

Example:

```python
import logging
from snowflake.connector import connect
from prefecto.ext.snowflake import execute

c = connect(...).cursor()
execute(c, "SELECT * FROM table")
```

Produces logs like

```txt
INFO - [rustic-human] Beginning command.
DEBUG - [rustic-human] Executing command:
SELECT * FROM table
INFO - [rustic-human] Command executed successfully.
```

And tracebacks like

```txt
ERROR - [rustic-human] Command failed.
Traceback (most recent call last):
...
snowflake.connector.errors.ProgrammingError: ...
[rustic-human] Command failed
SELECT * FROM table
```

You can also obfuscate parameters:

```python
execute(c, "SELECT * FROM table WHERE id = %(id)s", params={"id": 123}, obfuscate_params=True)
execute(c, "SELECT * FROM table WHERE id = %(id)s AND %(secret)s", params={"id": 123, "secret": "shhh"}, obfuscate_params=["secret"])
```

```txt
INFO - [delightful-octopus] Beginning command.
DEBUG - [delightful-octopus] Executing command:
SELECT * FROM table WHERE id = ****
INFO - [delightful-octopus] Command executed successfully.
INFO - [super-athlete] Beginning command.
DEBUG - [super-athlete] Executing command:
SELECT * FROM table WHERE id = 123 AND secret = ****
INFO - [super-athlete] Command executed successfully.
```