Skip to content

Instantly share code, notes, and snippets.

@dmussaku
Created February 17, 2025 08:55
Show Gist options
  • Save dmussaku/44b4d82978e9cf676012800f3677fb4a to your computer and use it in GitHub Desktop.
Save dmussaku/44b4d82978e9cf676012800f3677fb4a to your computer and use it in GitHub Desktop.
Data Engineering Code review test
"""
This is a code review exercise for Data Engineering candidates.
This code is not used in production in any way
"""
import requests
import time
import csv
import json
apiurl = "https://fake-api.com/data"
headers = {"Authorization": "Bearer token123"}
def getData():
try:
r = requests.get(apiurl, headers=headers)
if r.status_code == 200:
return r.json()
else:
print("Failed to fetch data:", r.status_code)
return None
except:
print("An error occurred")
return None
def TransformData(datalist=[]):
transformed_data = []
for i in range(len(datalist)):
item = datalist[i]
newobj = {}
newobj["id"] = item["id"]
newobj["username"] = item["name"].upper()
newobj["cost"] = item["price"] * 1.2
newobj["category"] = item["category"]
newobj["time"] = time.strftime("%Y-%m-%d %H:%M:%S")
transformed_data.append(newobj)
return transformed_data
def SaveToCSV(datalist, outputfile="data.csv"):
f = open(outputfile, "w")
writer = csv.writer(f)
writer.writerow(["id", "username", "cost", "category", "time"])
for item in datalist:
writer.writerow([item["id"], item["username"], item["cost"], item["category"], item["time"]])
print("File Saved")
f.close()
def main():
x = 0
while x < 3:
data = getData()
if data:
break
print("Retrying in 1 second")
time.sleep(1)
x += 1
final_data = TransformData(data)
SaveToCSV(final_data)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment