Snowflake Adapter¶

Tip

turu-snowflake depends on snowflake-connector-python.

Installation¶

pip install "turu[snowflake]"

Warning

If you have any dependencies on pyarrow, please install the following stubs in your development environment to avoid errors due to Unknown types.

# For development
pip install pyarrow-stubs

Usage¶

Basic Usage¶

import pydantic
import turu.snowflake


class User(pydantic.BaseModel):
    id: int
    name: str


connection = turu.snowflake.connect_from_env()

with connection.cursor() as cursor:
    user = cursor.execute_map(
        User,
        "SELECT 1, 'taro'",
    ).fetchone()

    assert user == User(id=1, name="taro")

Parameters Usage¶

format style ¶

import pydantic
import turu.snowflake


class User(pydantic.BaseModel):
    id: int
    name: str


connection = turu.snowflake.connect_from_env()

with connection.cursor() as cursor:
    user = cursor.execute_map(
        User,
        "SELECT %s, %s",
        [1, "taro"],
    ).fetchone()

    assert user == User(id=1, name="taro")

Warning

The variables placeholder must always be a %s, even if a different placeholder (such as a %d for integers or %f for floats) may look more appropriate for the type.

Keyword Parameters Usage¶

pyformat style ¶

import pydantic
import turu.snowflake


class User(pydantic.BaseModel):
    id: int
    name: str


connection = turu.snowflake.connect_from_env()

with connection.cursor() as cursor:
    user = cursor.execute_map(
        User,
        "SELECT %(id)s, %(name)s",
        {"id": 1, "name": "taro"},
    ).fetchone()

    assert user == User(id=1, name="taro")

Use methods¶

turu.snowflake.Cursor supports use_* methods to set options.

use_warehouse
use_database
use_schema
use_role

import os

import pydantic
import turu.snowflake


class User(pydantic.BaseModel):
    id: int
    name: str


connection = turu.snowflake.connect_from_env()

with (
    connection.cursor()
    .use_warehouse(os.environ["SNOWFLAKE_WAREHOUSE"])
    .use_database(os.environ["SNOWFLAKE_DATABASE"])
) as cursor:
    user = cursor.execute_map(
        User,
        "SELECT 1, 'taro'",
    ).fetchone()

    assert user == User(id=1, name="taro")

Tip

use_* methods are not supported in turu.snowflake.Connection. Settings for connection creation should be specified in the constructor.

Tip

AsyncCursor.use_* methods are not async. This is intentionally done in sync because of the short processing time of those methods and the deteriorating readability caused by method chaining.

Pandas DataFrame Validation using Pandera¶

You can use Pandera to validate the pandas.DataFrame.

Note

Pandera is optional feature. If you want to use Pandera, you need to install pandera like this:

pip install pandera

import pandera as pa
import pytest
import turu.snowflake
from pandera.errors import SchemaError
from pandera.typing import DataFrame, Series


class User(pa.DataFrameModel):
    id: Series[pa.Int8] = pa.Field(ge=2, alias="ID")


connection = turu.snowflake.connect_from_env()

with pytest.raises(SchemaError):
    with connection.cursor() as cursor:
        df: DataFrame[User] = cursor.execute_map(
            User,
            """
            SELECT 1 AS id
            UNION ALL
            SELECT 2 AS id
            """,
        ).fetch_pandas_all()