Skip to content

resources

An I/O Manager for handling the connection to Snowpark.

SnowparkResource

Bases: ConfigurableResource

I/O Manager Resource class for managing Snowpark sessions

Source code in data_science/defs/snowpark/resources.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class SnowparkResource(dg.ConfigurableResource):
    """I/O Manager Resource class for managing Snowpark sessions"""

    def __init__(self, **kwargs) -> None:     
        super().__init__(**kwargs)
        self._session = None

    def get_session(self, database: str ="analytics",
                    schema: str | None = None,
                    warehouse: str|None = None) -> "snowflake.snowpark.Session":  # type: ignore # noqa
        """Create a session with snowpark to allow for control of the remote execution
        environment.

        Args:
            database: Specify a specific database to use to remove requiring referencing
                assets by their fully qualified name.
            schema: Specify a specific schema to use to remove requiring referencing
                assets by their fully qualified name.
            warehouse: The compute warehouse to run on so that the appropriate resources
                are utilized.

        Returns:
            snowflake.snowpark.Session: A session which will allow for remote code
                execution on a snowflake warehouse.
        """
        import sys

        from data_platform_utils.helpers import get_database_name, get_schema_name
        from snowflake.snowpark import Session

        if sys.platform == "win32": # pragma: no coverage
            # hotfix to prevent path conversion issue on windows installations
            # should be able to remove once resolved in snowflake.snowpark
            import pathlib
            pathlib.PosixPath = pathlib.PurePosixPath

        if schema:
            schema = get_schema_name(schema)
        else:
            schema = get_secret_value("DESTINATION__SNOWFLAKE__USER")


        if not warehouse:
            warehouse = get_secret_value("DESTINATION__SNOWFLAKE__WAREHOUSE")

        self._session = (
            Session.builder.configs({ 
                "database":  get_database_name(database),
                "account":   get_secret_value("DESTINATION__SNOWFLAKE__HOST"),
                "user":      get_secret_value("DESTINATION__SNOWFLAKE__USER"),
                "password":  get_secret_value("DESTINATION__SNOWFLAKE__PASSWORD"),
                "role":      get_secret_value("DESTINATION__SNOWFLAKE__ROLE"),
                "warehouse": warehouse
            })
            .create()
        )

        try:
            self._session.use_schema(schema)
        except Exception:
            self._session.sql(f"create schema if not exists {schema}")
            self._session.use_schema(schema)

        return self._session

get_session(database='analytics', schema=None, warehouse=None)

Create a session with snowpark to allow for control of the remote execution environment.

Parameters:

Name Type Description Default
database str

Specify a specific database to use to remove requiring referencing assets by their fully qualified name.

 'analytics'
schema str | None

Specify a specific schema to use to remove requiring referencing assets by their fully qualified name.

 None
warehouse str | None

The compute warehouse to run on so that the appropriate resources are utilized.

 None

Returns:

Type Description
Session

snowflake.snowpark.Session: A session which will allow for remote code execution on a snowflake warehouse.
Source code in data_science/defs/snowpark/resources.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def get_session(self, database: str ="analytics",
                schema: str | None = None,
                warehouse: str|None = None) -> "snowflake.snowpark.Session":  # type: ignore # noqa
    """Create a session with snowpark to allow for control of the remote execution
    environment.

    Args:
        database: Specify a specific database to use to remove requiring referencing
            assets by their fully qualified name.
        schema: Specify a specific schema to use to remove requiring referencing
            assets by their fully qualified name.
        warehouse: The compute warehouse to run on so that the appropriate resources
            are utilized.

    Returns:
        snowflake.snowpark.Session: A session which will allow for remote code
            execution on a snowflake warehouse.
    """
    import sys

    from data_platform_utils.helpers import get_database_name, get_schema_name
    from snowflake.snowpark import Session

    if sys.platform == "win32": # pragma: no coverage
        # hotfix to prevent path conversion issue on windows installations
        # should be able to remove once resolved in snowflake.snowpark
        import pathlib
        pathlib.PosixPath = pathlib.PurePosixPath

    if schema:
        schema = get_schema_name(schema)
    else:
        schema = get_secret_value("DESTINATION__SNOWFLAKE__USER")


    if not warehouse:
        warehouse = get_secret_value("DESTINATION__SNOWFLAKE__WAREHOUSE")

    self._session = (
        Session.builder.configs({ 
            "database":  get_database_name(database),
            "account":   get_secret_value("DESTINATION__SNOWFLAKE__HOST"),
            "user":      get_secret_value("DESTINATION__SNOWFLAKE__USER"),
            "password":  get_secret_value("DESTINATION__SNOWFLAKE__PASSWORD"),
            "role":      get_secret_value("DESTINATION__SNOWFLAKE__ROLE"),
            "warehouse": warehouse
        })
        .create()
    )

    try:
        self._session.use_schema(schema)
    except Exception:
        self._session.sql(f"create schema if not exists {schema}")
        self._session.use_schema(schema)

    return self._session