Repository navigation
Expand file tree
/
Copy pathconnector.py
More file actions
99 lines (91 loc) · 3.34 KB
/
Copy pathconnector.py
File metadata and controls
99 lines (91 loc) · 3.34 KB
1
2
3
4
5
6
7
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
import datetime
import os
import snowflake.connector
from snowflake.connector import DictCursor
from flask import Blueprint, request, abort, jsonify, make_response
# Make the Snowflake connection
def connect() -> snowflake.connector.SnowflakeConnection:
if os.path.isfile("/snowflake/session/token"):
creds = {
'host': os.getenv('SNOWFLAKE_HOST'),
'port': os.getenv('SNOWFLAKE_PORT'),
'protocol': "https",
'account': os.getenv('SNOWFLAKE_ACCOUNT'),
'authenticator': "oauth",
'token': open('/snowflake/session/token', 'r').read(),
'warehouse': os.getenv('SNOWFLAKE_WAREHOUSE'),
'database': os.getenv('SNOWFLAKE_DATABASE'),
'schema': os.getenv('SNOWFLAKE_SCHEMA'),
'client_session_keep_alive': True
}
else:
creds = {
'account': os.getenv('SNOWFLAKE_ACCOUNT'),
'user': os.getenv('SNOWFLAKE_USER'),
'password': os.getenv('SNOWFLAKE_PASSWORD'),
'warehouse': os.getenv('SNOWFLAKE_WAREHOUSE'),
'database': os.getenv('SNOWFLAKE_DATABASE'),
'schema': os.getenv('SNOWFLAKE_SCHEMA'),
'client_session_keep_alive': True
}
return snowflake.connector.connect(**creds)
conn = connect()
# Make the API endpoints
connector = Blueprint('connector', __name__)
## Top 10 customers in date range
dateformat = '%Y-%m-%d'
@connector.route('/customers/top10')
def customers_top10():
# Validate arguments
sdt_str = request.args.get('start_range') or '1995-01-01'
edt_str = request.args.get('end_range') or '1995-03-31'
try:
sdt = datetime.datetime.strptime(sdt_str, dateformat)
edt = datetime.datetime.strptime(edt_str, dateformat)
except:
abort(400, "Invalid start and/or end dates.")
sql_string = '''
SELECT
o_custkey
, SUM(o_totalprice) AS sum_totalprice
FROM snowflake_sample_data.tpch_sf10.orders
WHERE o_orderdate >= '{sdt}'
AND o_orderdate <= '{edt}'
GROUP BY o_custkey
ORDER BY sum_totalprice DESC
LIMIT 10
'''
sql = sql_string.format(sdt=sdt, edt=edt)
try:
res = conn.cursor(DictCursor).execute(sql)
return make_response(jsonify(res.fetchall()))
except:
abort(500, "Error reading from Snowflake. Check the logs for details.")
## Monthly sales for a clerk in a year
@connector.route('/clerk/<clerkid>/yearly_sales/<year>')
def clerk_montly_sales(clerkid, year):
# Validate arguments
try:
year_int = int(year)
except:
abort(400, "Invalid year.")
if not clerkid.isdigit():
abort(400, "Clerk ID can only contain numbers.")
clerkid_str = f"Clerk#{clerkid}"
sql_string = '''
SELECT
o_clerk
, Month(o_orderdate) AS month
, SUM(o_totalprice) AS sum_totalprice
FROM snowflake_sample_data.tpch_sf10.orders
WHERE Year(o_orderdate) = {year}
AND o_clerk = '{clerkid}'
GROUP BY o_clerk, month
ORDER BY o_clerk, month
'''
sql = sql_string.format(year=year_int, clerkid=clerkid_str)
try:
res = conn.cursor(DictCursor).execute(sql)
return make_response(jsonify(res.fetchall()))
except:
abort(500, "Error reading from Snowflake. Check the logs for details.")