blob: fad142aeffc0b6b3f6f3a882d47a15ff4198e0b2 (
plain) (
tree)
|
|
#!/usr/bin/env python3
""" Get exchange rates from nb """
import csv
import sys
import tempfile
from datetime import datetime, timedelta
from io import StringIO
import common
import requests
from tzlocal import get_localzone
# I'm not sure I understand Norges Banks json-model. It seems a lot easier to just get the CSV, and convert it to JSON.
apiUrl = "https://data.norges-bank.no/api/data/EXR/B.EUR.NOK.SP?format=csv&locale=en"
pg_db = common.env("el_pg_db")
pg_host = common.env("el_pg_host")
pg_table = "nbex"
startTime = datetime.now(get_localzone()) - timedelta(days=10)
startTime = startTime.strftime("%Y-%m-%d")
# startTime = '2023-05-23' # <- edit for manual starttime. Like when filling in missing info.
endTime = datetime.now(get_localzone()).strftime("%Y-%m-%d")
temp = tempfile.NamedTemporaryFile()
### Get the data
try:
url = f"{apiUrl}&startPeriod={startTime}&endPeriod={endTime}"
response = requests.get(url, timeout=10)
if response.status_code != 200:
print(f"Request Error, Status code: {response.status_code}")
response.raise_for_status()
data = StringIO(response.text)
except requests.exceptions.RequestException as e:
print("Error: %s", e)
sys.exit(e)
values = []
csvReader = csv.DictReader(data, delimiter=";")
for item in csvReader:
values.append(
(item["TIME_PERIOD"], item["BASE_CUR"], item["QUOTE_CUR"], item["OBS_VALUE"])
)
### insert data into database
# SQL
sql = """INSERT INTO nbex
VALUES(%s, %s, %s, %s)
ON CONFLICT (startdate,base_cur,quote_cur) DO NOTHING"""
common.dbi(sql, values, verbose=True)
|
