Snowflake connection helper functions
Project description
Snowflake connection helper functions
A Python library to simplify connecting to Snowflake databases
Notes
- This is a major version upgrade and breaks compatibility with the previous versions (<
0.3.0
).sfconn
now relies onsnowflake-python-connector
for accessing named connections (e.g.connections.toml
). If you still need the ability use named connections from~/.snowflake/config
, please continue using older version (0.2.x
). sfconn
modifies the wayprivate_key_file
connection option is evaluated in the following ways:- with
--keyfile-anchor
option (default$SFCONN_KEYFILE_ANCHOR
): uses the supplied value to anchor any relative paths (v/s anchoring paths relative to current working directory of the running process) - with
--keyfile-strip-pfx
option (default$SFCONN_KEYFILE_STRIP_PFX
): strips the specified prefix from the path values (so--keyfile-anchor
option can be used)
- with
Installation
Use Python's standard pip
utility for installation:
pip install --upgrade sfconn
Usage
getconn()
Usage:
def getconn(connection_name: str | None, **overrides: dict[str, Any]) -> Connection
getconn
accepts a connection name and returns a connection object with modified behavior as noted above.
Example:
from sfconn import getconn
# assuming 'dev' is a connection defined in ~/.snowflake/config
with getconn('dev', schema='PUBLIC') as cnx:
with cnx.cursor() as csr:
csr.execute('SELECT CURRENT_USER()')
print("Hello " + csr.fetchone()[0])
conn_opts()
Usage:
def conn_opts(name: str | None , **overrides: dict[str, Any]) -> dict[str, Any]
conn_opts
, returns a Python dict
object populated with options and values. This can be useful passing as an argument to snowflake.snowpark.Session.builder.configs()
method.
Example:
from sfconn import conn_opts
from snowflake.snowpark import Session
# assuming 'dev' is a connection defined in ~/.snowflake/config
session = Session.builder.configs(conn_opts('dev')).create()
run_query*()
Cursor objects add a family of few convenience methods that return an Iterable
of values instead of generic tuple
or dict
.
<cursor>.run_query(<callable>|<class>, <sql> [,<params>])
: Returns an Iterable of values.<callable>
is a mapping function that shall accept all column values of a row as individual arguments, in order, and returns a value that will be used forIterable
.<class>
is any Python class whose attribute names, after upper-casing, are treated as column names from the result set.<class>
can include only a subset of a all available column from the query result as attributes and in a different order than the query.
<cursor>.run_query1(<callable>|<class>, <sql> [,<params>])
: Similar torun_query
, except returns a single value. Note, if at least one value is not available, raisesProgrammingError
exception.<cursor>.run_query1_opt(<callable>|<class>, <sql> [,<params>])
: Similar torun_query1
, except instead of raising an exception, the method returnsNone
.
Examples:
import datetime as dt
from collections import namedtuple
Result = namedtuple("Result", ["user", "date"])
def mkResult(x: str, y: dt.date) -> Result:
return Result(x, y)
with getconn() as cnx, cnx.cursor() as csr:
result = csr.run_query1(mkResult, "select current_user() as user, current_date() as date")
import datetime as dt
from dataclasses import dataclass
@dataclass
class Result:
date: dt.date
user: str
with getconn() as cnx, cnx.cursor() as csr:
result = csr.run_query1(
Result,
"select current_user() as user, current_date() as date, current_warehouse() as wh_name"
)
Decorator functions
Python scripts that accept command-line parameters and use argparse
library, can use decorator functions to further reduce boilerplate code needed for setting up common Snowflake connection options as command-line arguments
def with_connection_args(doc: str | None) -> Callable[[argparse.ArgumentParser], None]:
def with_connection(logger = None) -> Callable[[Connection, ...], None]:
def with_connection_option(logger = None) => Callable([dict[str, Any, ...]])
with_connection_args()
decorator function:
- builds an
ArgumentParser
object - adds common Snowflake connection options as arguments that allow overriding values specified in
~/.snowsql/config
- calls the decorated function with the parser object to allow adding any script specific options
with_connection()
decorator function:
- consumes standard Snowflake connection options (specified with
args()
) - creates a connection object
- calls the decorated function with a connection object as first parameter and any other script specific options that were specified on command line
with_connection_option()
decorator function:
- Similar to
entry()
but passes adict
of options as the first parameter. This is useful for passing options to thesnowflake.snowpark.Session.builder.configs()
method
Example:
from sfconn import args, entry
@entry
def main(con, show_account: bool):
with con.cursor() as csr:
csr.execute('SELECT CURRENT_USER()')
print("Hello " + csr.fetchone()[0])
if show_account:
csr.execute("SELECT CURRENT_ACCOUNT()")
print("You are connected to account: " + csr.fetchone()[0])
@args("Sample application that greets the current Snowflake user")
def getargs(parser):
parser.add_argument("-a", "--show-account", action='store_true', help="show snowflake account name")
if __name__ == '__main__':
main(**vars(getargs()))
get_token()
Function sfconn.get_token()
returns a JWT token for connections that use private_key_path
option. An optional lifetime value can be specified (default 54 minutes)
Example:
from sfconn import get_token
# assuming 'dev' is a connection defined in ~/.snowflake/config and uses key-pair authentication
jwt_token = get_token('dev', 120) # get a token valid for 120 minutes
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.