python - 如何使用 Python 将 blob 数据移动到 Snowflake

Question

我正在尝试将数据从 ADLS blob 移动到 Snowflake 表。

我可以对 UI 执行相同的操作。

UI 遵循的步骤:

生成了以下 SAS token :

sp=rl&st=2021-06-01T05:45:37Z&se=2021-06-01T13:45:37Z&spr=https&sv=2020-02-10&sr=c&sig=rYYY4o%2YY3jj%2XXXXXAB%2Bo8ygrtyAVCnPOxomlOc%3D

能够在 Snowflake Web UI 中使用上述 token 加载表:

copy into FIRST_LEVEL.MOVIES
  from 'azure://adlsedmadifpoc.blob.core.windows.net/airflow-dif/raw-area/'
  credentials=(azure_sas_token='sp=rl&st=2021-06-01T05:45:37Z&se=2021-06-01T13:45:37Z&spr=https&sv=2020-02-10&sr=c&sig=rYYY4o%2YY3jj%2XXXXXAB%2Bo8ygrtyAVCnPOxomlOc%3D')
   FORCE = TRUE file_format = (TYPE = CSV);

我正在尝试用 Python 做同样的事情:

from azure.storage.blob import BlobServiceClient,generate_blob_sas,BlobSasPermissions
from datetime import datetime,timedelta
import snowflake.connector

def generate_sas_token(file_name):

    sas = generate_blob_sas(account_name="xxxx",
account_key="p5V2GELxxxxQ4tVgLdj9inKwwYWlAnYpKtGHAg==", container_name="airflow-dif",blob_name=file_name,permission=BlobSasPermissions(read=True),
expiry=datetime.utcnow() + timedelta(hours=2))
    print (sas)
    return sas

sas = generate_sas_token("raw-area/moviesDB.csv")

# Connectio string

conn = snowflake.connector.connect(user='xx',password='xx@123',account='xx.southeast-asia.azure',database='xx')

# Create cursor

cur = conn.cursor()
cur.execute(
            f"copy into FIRST_LEVEL.MOVIES FROM  'azure://xxx.blob.core.windows.net/airflow-dif/raw-area/moviesDB.csv'   credentials=(azure_sas_token='{sas}')  file_format = (TYPE = CSV) ;")
cur.execute(f" Commit  ;")
# Execute SQL statement
cur.close()
conn.close()

代码中生成的 SAS token :

se=2021-06-01T07%3A42%3A11Z&sp=rt&sv=2020-06-12&sr=b&sig=ZhZMPSI%yyyyAPTqqE0%3D

通过 python 生成 sas token 时，我无法使用列表权限。

我遇到以下错误:

    cursor=cursor,
snowflake.connector.errors.ProgrammingError: 091003 (22000): Failure using stage area. Cause: [Server failed to authenticate the request. Make sure the value of Authorization header is formed correctly including the signature. (Status Code: 403; Error Code: AuthenticationFailed)]

我将来可能会在该文件夹中包含 csv 文件列表。

任何帮助表示赞赏。谢谢。

Answer 1

以下代码有效:

from azure.storage.blob import generate_container_sas, ContainerSasPermissions
from datetime import datetime,timedelta
import snowflake.connector

def get_sas_token():
    container_sas_token = generate_container_sas(
        account_name = 'XX',
        account_key = 'p5V2GEL3AqGuPMMYXXXQ4tVgLdj9inKwwYWlAnYpKtGHAg==',
        container_name = 'airflow-dif',
        permission=ContainerSasPermissions(read=True,list=True),
        expiry=datetime.utcnow() + timedelta(hours=1)
    )
    print (container_sas_token)
    return container_sas_token

sas = get_sas_token()

# Connectio string

conn = snowflake.connector.connect(user='XX',password='XX@123',account='XX.southeast-asia.azure',database='XX')

# Create cursor

cur = conn.cursor()
cur.execute(
            f"copy into FIRST_LEVEL.MOVIES FROM  'azure://XX.blob.core.windows.net/airflow-dif/raw-area/'   credentials=(azure_sas_token='{sas}')  FORCE = TRUE file_format = (TYPE = CSV) ;")

    
print (cur.fetchone())
cur.execute(f" Commit  ;")

# Execute SQL statement

cur.close()
conn.close()

感谢 Gaurav 的投入。