Python-Projekte/snowflake/snowflake_reisacher.py

import csv
import json
from datetime import datetime
from pathlib import Path

import requests

import snowflake.connector

url = "https://sq43047.eu-central-1.snowflakecomputing.com/oauth/token-request"
# User: REISACHER_GLOBALCUBE
# Password: ZXA9gmy4axa_wtd3vxu
# Login: https://sq43047.eu-central-1.snowflakecomputing.com/oauth/authorize?response_type=code&client_id=AxayPnlaAHtkZ9FuBbERcf0Wa8E%3D&redirect_uri=https%3A%2F%2Flocalhost.com
# clientid: AxayPnlaAHtkZ9FuBbERcf0Wa8E=
# clientSecret: RC18m4LHUwcDlxSInZ5FOlUOQ8J4ud5LMcq7ATZNgps=
# old code: A7FF59DAFCAE232C56413B78D0F5C48B33C6EE34
# code: 9FDA41370466E0E108F1712FAEF08297CB05E81B
login = {
    "user": "AxayPnlaAHtkZ9FuBbERcf0Wa8E=",
    "password": "RC18m4LHUwcDlxSInZ5FOlUOQ8J4ud5LMcq7ATZNgps=",
}

credentials = {
    "user": "1103",
    "account": "sq43047.eu-central-1",
    "authenticator": "oauth",
    "token": "256A1D8740C7AC1FE617BFF4B2C74E31789BE754",
    "database": "REISACHER",
    "schema": "EXTRAKTION",
}

sql = """
select
  *
from "01_EXTR_REISACHER_RSP_ACCOUNTCUSTOMERCARES" AS "T1"
inner join "01_EXTR_REISACHER_RSP_ACCOUNTS_LEGALENTITYINFORMATION" AS "T2" ON "T1"."accountId" = "T2"."accountId"

"""

# sql = "SHOW TABLES"


def get_auth_code():
    payload = {
        "grant_type": "authorization_code",
        "code": credentials["token"],
        "redirect_uri": "https://localhost.com",
    }
    r = requests.post(
        url=url,
        auth=(*login.values(),),
        data=payload,
    )
    token = r.json()
    return token


def get_access_token(token):
    payload = {
        "grant_type": "refresh_token",
        "refresh_token": token["refresh_token"],
        "redirect_uri": "https://localhost.com",
    }
    r = requests.post(
        url=url,
        auth=(*login.values(),),
        data=payload,
    )
    token = r.json()
    return token


def snowflake_to_csv(csv_export_file):
    token_json = Path("token.json")
    if token_json.exists():
        token = json.load(token_json.open("r"))
        if datetime.now().timestamp() > token_json.stat().st_mtime + token["expires_in"]:
            token = token.update(get_access_token(token))
            json.dump(token, token_json.open("w"), indent=2)
    else:
        token = get_auth_code()
        json.dump(token, token_json.open("w"), indent=2)
    credentials["token"] = token["access_token"]
    # get_refresh_token()

    with snowflake.connector.connect(**credentials) as conn:
        cur = conn.cursor()
        cur.execute(sql)

        with Path(csv_export_file).open("w", encoding="latin-1", newline="") as fwh:
            csv_export = csv.writer(fwh, delimiter=";")
            csv_export.writerow([i[0] for i in cur.description])
            csv_export.writerows(cur.fetchall())


if __name__ == "__main__":
    snowflake_to_csv(Path(__file__).parent / "01_EXTR_REISACHER_RSP_ACCOUNTCUSTOMERCARES.csv")
    # print(datetime.now().timestamp())