need help understanding streaming

[M2ATrail]

so I'm downloading certain data from an API, I'm going for streaming since their server cluster randomly closes connections.

this is just a sketch of what I'm doing, I plan on reworking it later for better logging and skipping downloaded files, but I want to test what happens if the connection fails for whatever reason, but i never used streaming before.

Process, three levels of loops, project, dates, endpoints.

inside those, I want to stream the call to those files, if I get 200 then just write.

if I get 429 sleep for 61 seconds and retry.

if 504 (connection closed at their end), sleep 61s, consume one retry

anything else, throw the exception, sleep 61s and consume one retry

I tried forcing 429 by calling that thing seven times (supposed to be 4 requests per minutes), but it isn't happening, and I need a sanity check.

I'd also probably need to async this at project level thing but that's a level of complexity that I don't need now (each project have its own different limit)

import time
import pandas as pd
import helpers
import httpx
import get_data

iterable_users_export_path = helpers.prep_dir(
    r"imsdatablob/Iterable Exports/data_csv/Iterable Users Export"
)
iterable_datacsv_endpoint_paths = {
    "emailSend": helpers.prep_dir(r"imsdatablob/Iterable Exports/data_csv/Iterable emailSend Export"),
    "emailOpen": helpers.prep_dir(r"imsdatablob/Iterable Exports/data_csv/Iterable emailOpen Export"),
    "emailClick": helpers.prep_dir(r"imsdatablob/Iterable Exports/data_csv/Iterable emailClick Export"),
    "hostedUnsubscribeClick": helpers.prep_dir(r"imsdatablob/Iterable Exports/data_csv/Iterable hostedUnsubscribeClick Export"),
    "emailComplaint": helpers.prep_dir(r"imsdatablob/Iterable Exports/data_csv/Iterable emailComplaint Export"),
    "emailBounce": helpers.prep_dir(r"imsdatablob/Iterable Exports/data_csv/Iterable emailBounce Export"),
    "emailSendSkip": helpers.prep_dir(r"imsdatablob/Iterable Exports/data_csv/Iterable emailSendSkip Export"),
}


start_date = "2025-04-01"
last_download_date = time.strftime("%Y-%m-%d", time.localtime(time.time() - 60*60*24*2))
date_range = pd.date_range(start=start_date, end=last_download_date)
date_range = date_range.strftime("%Y-%m-%d").tolist()


iterableProjects_list = get_data.get_iterableprojects_df().to_dict(orient="records")

with httpx.Client(timeout=150) as client:

    for project in iterableProjects_list:
        iterable_headers = {"api-key": project["projectKey"]}
        for d in date_range:
            end_date = (pd.to_datetime(d) + pd.DateOffset(days=1)).strftime("%Y-%m-%d")

            for e in iterable_datacsv_endpoint_paths:
                url = f"https://api.iterable.com/api/export/data.csv?dataTypeName={e}&range=All&delimiter=%2C&startDateTime={d}&endDateTime={end_date}"
                file = f"{iterable_datacsv_endpoint_paths[e]}/sfn_{project['projectName']}-d_{d}.csv"
                retries = 0
                max_retries = 10
                while retries < max_retries:
                    try:
                        with client.stream("GET", url, headers=iterable_headers, timeout=30) as r:
                            if r.status_code == 200:
                                with open(file, "w") as file:
                                    for chunk in r.iter_lines():
                                        file.write(chunk)
                                        file.write('\n')
                                break

                            elif r.status_code == 429:
                                time.sleep(61)
                                print(f"429 for {project['projectName']}-{e} -{start_date}")
                                continue
                            elif r.status_code == 504:
                                retries += 1
                                print(f"504 {project['projectName']}-{e} -{start_date}")
                                time.sleep(61)
                                continue
                    except Exception as excp:
                        retries += 1
                        print(f"{excp} {project['projectName']}-{e} -{start_date}")
                        time.sleep(61)
                        if retries == max_retries:
                            print(f"This was the last retry: {project['projectName']}-{e} -{start_date}")

[ehsandeep]

@M2ATrail you probbaly looking for https://github.com/encode/httpx

[M2ATrail]

oh lol sorry about that.
have a great weekend.