The Snowflake Ingest Service SDK allows users to ingest files into their Snowflake data warehouse in a programmatic fashion via key-pair authentication.
The Snowflake Ingest SDK requires Python 3.4 or above. Backwards compatibility with older versions of Python 3 or any versions of Python 2 is not planned at this time.
Snowflake Authentication for the Ingest Service requires creating a 2048 bit RSA key pair and, registering the public key with Snowflake. For detailed instructions, please visit the relevant Snowflake Documentation Page.
Internally, the Snowflake Ingest SDK makes use of Furl, PyJWT, and Requests. In addition, the cryptography is used with PyJWT to sign JWT tokens.
If you would like to use this sdk, you can install it using python setuptools.
pip install snowflake-ingest
Here is a simple "hello world" example for using ingest sdk.
from logging import getLogger
from snowflake.ingest import SimpleIngestManager
from snowflake.ingest import StagedFile
import time
from requests import HTTPError
logger = getLogger(__name__)
# assume public key has been registered in Snowflake
# private key in pem format
private_key="""-----BEGIN PRIVATE KEY-----
abc...
-----END PRIVATE KEY-----"""
# list of files in the stage specified in the pipe definition
file_list=['a.csv', 'b.csv']
ingest_manager = SimpleIngestManager(account='testaccount',
host='testaccount.snowflakecomputing.com',
user='ingest_user',
pipe='TESTDB.TESTSCHEMA.TESTPIPE',
private_key=private_key)
# list of files, but wrapped into a class
staged_file_list = []
for file_name in file_list:
staged_file_list.append(StagedFile(file_name, None))
try:
resp = ingest_manager.ingest_files(staged_file_list)
except IngestResponseError as e:
logger.error(e)
exit(1)
# This means Snowflake has received file and will start loading
assert(resp['responseCode'] == 'SUCCESS')
# Needs to wait for a while to get result in history
while True:
history_resp = ingest_manager.get_history()
if len(history_resp['files']) == 2:
print('Ingest Report:\n')
print(history_resp)
break
else:
# wait for 20 seconds
time.sleep(20)
# Valid ISO 8601 format requires Z at the end
hour = timedelta(hours=1)
date = datetime.datetime.utcnow() - hour
history_range_resp = ingest_manager.get_history_range(date.isoformat() + 'Z')
print('\nHistory scan report: \n')
print(history_range_resp)