IT/Python
[Python] 파이썬에서 Redash 리대시 쿼리 결과 값 가져오기
Life in the World
2025. 5. 9. 16:38
반응형
파이썬에서 Redash 리대시 쿼리 결과 값을 가져오는 방법에 대해 알아보도록 하겠습니다.
1. requests 라이브러리 설치
우선은 requests 라이브러리가 필요합니다.
설치가 되어 있지 않다면 pip 명령어를 통해 requests 라이브러리를 설치해주도록 합시다.
(python 2 버전은 pip 명령어 / python 3 버전은 pip3 명령어를 사용)
pip install requests
pip3 install requests
다음과 같이 import 후 호출했을 때, 설치된 경로가 출력된다면 라이브러리 설치는 끝났습니다.
2. API KEY 가져오기
redash 쿼리 결과 값을 출력하기 위해서는 사용자의 API KEY가 필요합니다.
redash 좌측 하단에 본인의 프로필 이미지가 있는데, 클릭 후 profile을 클릭해줍니다.
클릭했다면 아래와 같은 페이지가 보일겁니다. 거기서 API KEY를 복사해서 따로 저장해놓도록 합시다.
3. Redash 쿼리 결과 값 출력하기 (파라미터 X)
준비는 끝났으니 이제 Python에서 Redash 쿼리 결과 값을 가져와서 출력해보도록 하겠습니다.
쿼리는 Redash에서 아래와 같이 구성하였습니다.
WITH sample_date AS (
SELECT CAST('2025-05-07' AS DATE) AS dt UNION ALL
SELECT CAST('2025-05-08' AS DATE) UNION ALL
SELECT CAST('2025-05-09' AS DATE) UNION ALL
SELECT CAST('2025-05-10' AS DATE)
),
fruit AS (
SELECT 'apple' AS fruit UNION ALL
SELECT 'banana' AS fruit UNION ALL
SELECT 'grape' AS fruit UNION ALL
SELECT 'melon' AS fruit UNION ALL
SELECT 'peach' AS fruit UNION ALL
SELECT 'mango' AS fruit
)
SELECT sd.dt, f.fruit
FROM sample_date sd
CROSS JOIN fruit f
ORDER BY dt, fruit
파라미터가 없는 쿼리의 결과를 출력하는 소스 코드입니다.
import requests
import time
import json
def execute_redash():
# 쿼리 ID
query_id = 738 # 실행시키고자 하는 쿼리의 ID를 입력 (URL에서 확인 가능)
# URL 및 api key 설정
redash_url = "리대시 URL을 입력해주세요" # (ex) https://www.litw.tistory
api_key = "본인의 API 키를 입력해주세요" # (ex) wmfelmzef1fmlszdmf3a
# 쿼리 실행 요청
response = requests.post(
f'{redash_url}/api/queries/{query_id}/results',
headers={'Authorization': f'Key {api_key}'},
json={
# max_age를 0으로 설정해야 매번 쿼리를 재실행함
"max_age": 0}
)
try:
data = response.json()
except Exception as e:
raise Exception(f"응답을 JSON으로 변환할 수 없습니다: {response.text}") from e
# job 키가 없으면 상세 에러 출력
if 'job' not in data or 'id' not in data['job']:
raise Exception(f"'job' 또는 'id' 키가 응답에 없습니다: {data}")
job_id = data['job']['id']
# 쿼리 실행 완료까지 대기
while True:
job_response = requests.get(
f'{redash_url}/api/jobs/{job_id}',
headers={'Authorization': f'Key {api_key}'}
)
try:
job_data = job_response.json()
except Exception as e:
raise Exception(f"잡 상태 응답을 JSON으로 변환할 수 없습니다: {job_response.text}") from e
if 'job' not in job_data:
raise Exception(f"'job' 키가 잡 상태 응답에 없습니다: {job_data}")
status = job_data['job']['status']
if status == 3:
break
elif status == 4:
raise Exception('쿼리 실행 실패')
time.sleep(1)
query_result_id = job_data['job']['query_result_id']
result_response = requests.get(
f'{redash_url}/api/query_results/{query_result_id}.json',
headers={'Authorization': f'Key {api_key}'}
)
try:
result = result_response.json()['query_result']['data']['rows']
except Exception as e:
raise Exception(f"결과 응답에서 데이터를 가져올 수 없습니다: {result_response.text}") from e
print(json.dumps(result, indent=4))
def main():
execute_redash()
if __name__ == '__main__':
main()
출력 결과를 확인하기 전에 디버그를 통해 변수에 어떤 구조로 값이 저장되는지 확인해보겠습니다.
아래와 같이 리스트에 각 ROW가 딕셔너리 형태로 저장되고 있네요.
print로 출력한 결과도 한번 확인해보겠습니다.
문제없이 정상적으로 출력되는 모습입니다.
4. Redash 쿼리 결과 값 출력하기 (파라미터 O)
이번에는 파라미터 있는 쿼리의 결과 값을 가져와보도록 하겠습니다.
쿼리는 Redash에서 아래와 같이 구성하였습니다. (위에서 작성한 쿼리에서 where 절만 추가)
WITH sample_date AS (
SELECT CAST('2025-05-07' AS DATE) AS dt UNION ALL
SELECT CAST('2025-05-08' AS DATE) UNION ALL
SELECT CAST('2025-05-09' AS DATE) UNION ALL
SELECT CAST('2025-05-10' AS DATE)
),
fruit AS (
SELECT 'apple' AS fruit UNION ALL
SELECT 'banana' AS fruit UNION ALL
SELECT 'grape' AS fruit UNION ALL
SELECT 'melon' AS fruit UNION ALL
SELECT 'peach' AS fruit UNION ALL
SELECT 'mango' AS fruit
)
SELECT sd.dt, f.fruit
FROM sample_date sd
CROSS JOIN fruit f
where dt = cast('{{dt}}' as date)
ORDER BY dt, fruit
파라미터가 있는 쿼리의 결과를 출력하는 소스 코드입니다. (파라미터는 where 절에 있는 dt 입니다.)
파라미터에는 오늘 날짜를 넣어서 실행시켜보도록 하겠습니다.
import requests
import time
import json
from datetime import date
def execute_redash():
# 쿼리 ID
query_id = 738 # 실행시키고자 하는 쿼리의 ID를 입력 (URL에서 확인 가능)
# 파라미터에 오늘 날짜 설정
input_date = date.today().isoformat()
parameters = {"dt": input_date}
# URL 및 api key 설정
redash_url = "리대시 URL을 입력해주세요" # (ex) https://www.litw.tistory
api_key = "본인의 API 키를 입력해주세요" # (ex) wmfelmzef1fmlszdmf3a
# 쿼리 실행 요청
response = requests.post(
f'{redash_url}/api/queries/{query_id}/results',
headers={'Authorization': f'Key {api_key}'},
json={
# max_age를 0으로 설정해야 매번 쿼리를 재실행함
"max_age": 0,
# 파라미터 전달
"parameters": parameters}
)
try:
data = response.json()
except Exception as e:
raise Exception(f"응답을 JSON으로 변환할 수 없습니다: {response.text}") from e
# job 키가 없으면 상세 에러 출력
if 'job' not in data or 'id' not in data['job']:
raise Exception(f"'job' 또는 'id' 키가 응답에 없습니다: {data}")
job_id = data['job']['id']
# 쿼리 실행 완료까지 대기
while True:
job_response = requests.get(
f'{redash_url}/api/jobs/{job_id}',
headers={'Authorization': f'Key {api_key}'}
)
try:
job_data = job_response.json()
except Exception as e:
raise Exception(f"잡 상태 응답을 JSON으로 변환할 수 없습니다: {job_response.text}") from e
if 'job' not in job_data:
raise Exception(f"'job' 키가 잡 상태 응답에 없습니다: {job_data}")
status = job_data['job']['status']
if status == 3:
break
elif status == 4:
raise Exception('쿼리 실행 실패')
time.sleep(1)
query_result_id = job_data['job']['query_result_id']
result_response = requests.get(
f'{redash_url}/api/query_results/{query_result_id}.json',
headers={'Authorization': f'Key {api_key}'}
)
try:
result = result_response.json()['query_result']['data']['rows']
except Exception as e:
raise Exception(f"결과 응답에서 데이터를 가져올 수 없습니다: {result_response.text}") from e
print(json.dumps(result, indent=4))
def main():
execute_redash()
if __name__ == '__main__':
main()
실행 결과
파라미터가 필요한 쿼리의 경우에도 정상적으로 실행되는 모습입니다.
이상으로 Python에서 Redash 쿼리의 결과 값을 가져오는 방법에 대해 알아보았습니다.
반응형