One Script, Any API: A Config-Driven Ingestion Pipeline
- Bill Donofrio
- 4 days ago
- 8 min read
Imagine you are responsible for building a new data lakehouse, and time is not on your side. Management wants the project finished yesterday, leaving little room for error. How do you bring in ten new data sources in the fastest, most efficient way possible, all while keeping the pipeline scalable and the data clean?
This article builds a single, config-driven script that makes your API calls efficient, configurable, and scalable across many sources at once. If you are new to APIs, start with the first article in this series, “Stop Fearing APIs: A Data Engineer’s Field Guide.” From here, it is time to make the impossible possible.
Getting Started
Importing Your Libraries
The first step is to import all the necessary libraries. If you are just testing this code in Google Colab or a Jupyter Notebook, the pieces that use Spark and Azure libraries are commented out below. The details will be discussed as you read on.
import requests import time # Only used for the retry delay import pandas as pd import csv # Only used for audit log import os # Only used for audit log from datetime import datetime, timezone # Only used for audit log
# from azure.identity import DefaultAzureCredential # Uncomment if you are using Fabric # from azure.keyvault.secrets import SecretClient # Uncomment if you are using Fabric # from delta.tables import DeltaTable # Uncomment to use Spark # from pyspark.sql import SparkSession # Uncomment to use Spark
# spark = SparkSession.builder.appName("SparkSession").getOrCreate() # Uncomment to use Spark |
Setting Up the Config Table
The next step is to build your configuration table. In a real production environment, you would load an Excel or CSV file instead. For simplicity, this example uses a plain DataFrame. As we work through the code, each parameter’s role in the pipeline will be explained.
config = [ { "base_url": "https://rickandmortyapi.com/api", "end_point": "character", "destination_database": "rick_and_morty", "destination_schema": "characters", "destination_table": "aliens", "write_type": "overwrite", "primary_key": "id", "columns_needed": "id, name, status, species, gender", "key_vault_url": None, "key_vault_secret": None, "retries_allowed": 3, "get_results": '["results"]', "pagination": '["info"]["next"]', "api_parameter_columns": "name, status, species, type, gender, dimension, episode", "name": None, "status": None, "species": "Alien", "type": None, "gender": None, "dimension": None, "episode": None, }, { "base_url": "https://rickandmortyapi.com/api", "end_point": "location", "destination_database": "rick_and_morty", "destination_schema": "locations", "destination_table": "planets", "write_type": "overwrite", "primary_key": "id", "columns_needed": "id, name, type, dimension", "key_vault_url": None, "key_vault_secret": None, "retries_allowed": 3, "get_results": '["results"]', "pagination": '["info"]["next"]', "api_parameter_columns": "name, status, species, type, gender, dimension, episode", "name": None, "status": None, "species": None, "type": "Planet", "gender": None, "dimension": None, "episode": None, }, { "base_url": "https://rickandmortyapi.com/api", "end_point": "episode", "destination_database": "rick_and_morty", "destination_schema": "episodes", "destination_table": "season", "write_type": "overwrite", "primary_key": "id", "columns_needed": "id, name, air_date, episode", "key_vault_url": None, "key_vault_secret": None, "retries_allowed": 3, "get_results": '["results"]', "pagination": '["info"]["next"]', "api_parameter_columns": "name, status, species, type, gender, dimension, episode", "name": None, "status": None, "species": None, "type": None, "gender": None, "dimension": None, "episode": "S01", }, ]
config_df = pd.DataFrame(config) display(config_df) |
Getting Your Header
The function below builds your header. Notice that it takes in the key_vault_url and key_vault_secret from the config DataFrame. For the Rick and Morty API, this part is ignored, since it does not need a special token or header to authenticate. This is a rare case, though. Normally, you have to get a token directly from the application, with permission to access some or all of its data. Since that token grants carte blanche authority to the application, it must be stored as a secret. In this example, the secret is retrieved from Azure Key Vault. You would need to modify this slightly if you are using AWS or GCP. Either way, the final result should be a header that looks like this.
headers = { "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c", "Content-Type": "application/json" } |
def get_auth_header(vault_url, secret_name): if vault_url is not None and vault_url != '': credential = DefaultAzureCredential() client = SecretClient(vault_url=vault_url, credential=credential)
token = client.get_secret(secret_name).value
headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } else: headers = {}
return headers |
Creating the Parameters Dictionary
This part builds the parameters dictionary for the API call. In this example, there are seven possible parameters in the config DataFrame: name, status, species, type, gender, dimension, and episode. There is also an api_parameter_columns field that stores that whole list. Since each row can be configured however the user sees fit, many of those fields will be blank. This step drops the blanks and keeps only the parameters that have a value. Below is an example of what the final result should look like.
params = { "species": "Alien" } |
def build_params(row, API_PARAMETER_COLUMNS): params = {} for column in API_PARAMETER_COLUMNS: value = row[column] if value is not None and value != "": params[column] = value return params |
Calling the API and Getting JSON Results
This function does the heavy lifting. Notice that the max_retries and pagination parameters from the config DataFrame are passed in, along with the params built in the previous step. The base_url and end_point parameters are combined to create the url you see as the first argument. Finally, the get_results parameter tells the function exactly which part of the response holds the records to keep. There are two while loops. The outer one handles pagination, and the inner one handles retries if the connection drops. Both loops check the status code and only keep a page’s results once it comes back as 200. There is a 60-second delay if a retry is needed. The extend method combines each page into one complete list of results, and the try and except blocks handle any request errors along the way.
def get_data(url, params, max_retries, pagination, get_results): all_results = [] page_number = 1
while url is not None: current_try = 1 error_message = None page_data = None
while current_try <= max_retries: print(f"Page {page_number}, attempt {current_try} of {max_retries}: calling {url}")
try: response = requests.get(url, params=params if page_number == 1 else None) status = response.status_code
if status == 429: return all_results, None elif status == 200: page_data = response.json() break
error_message = f"Received status code {status}: {response.text[:200]}" print(f" Failed with status {status}.")
except requests.exceptions.RequestException as e: error_message = str(e) print(f" Request error: {error_message}")
current_try += 1 if current_try <= max_retries: print("Waiting 60 seconds before retrying...") time.sleep(60)
if page_data is None: print(f" Giving up on page {page_number} after {max_retries} attempts.") return None, error_message
all_results.extend(eval(f"page_data{get_results}"))
url = eval(f"page_data{pagination}") # url = page_data["info"]["next"] page_number += 1
return all_results, None |
|
One note on that status == 429 check. A 429 means the API is rate-limiting you. Rather than burning through retries against a server that just asked you to back off, the function returns whatever results it already collected and stops. That is a deliberate design choice, but it does mean a run that hits a rate limit partway through will quietly return partial data with no error attached, so keep an eye on your row counts if you are pulling a lot of pages.
Identifying the Primary Key
This function assigns a primary key if one is not already available, based on the primary_key value from the config DataFrame. In this example, there is a simple “id” primary key for every row. If you ever need to combine several fields to create uniqueness, this is where you would do that. A primary key is required if you plan on using a MERGE command later on.
def add_primary_key(df, PRIMARY_KEY): current_columns = df.columns if PRIMARY_KEY not in current_columns: keys = PRIMARY_KEY.replace(" ", "").split(",")
df = df.copy() df["primary_key"] = df[keys].astype(str).agg("-".join, axis=1) return df |
Writing to the Data Lakehouse
This part stays commented out until you run the actual pipeline. The key takeaway is that the final DataFrame gets passed into this function, along with a write_type of overwrite, merge, or append. If you use the merge option, you will also need to supply a primary_key. The three-part namespace of database_name, schema_name, and table_name comes from the same config DataFrame used throughout this pipeline. One more point worth calling out: setting option("mergeSchema", "true") allows for schema evolution, the ability to automatically add new columns as your table changes over time.
df.write.format("delta").mode("overwrite").option("mergeSchema", "true").saveAsTable(full_table_name) |
def write_to_lakehouse(df, write_type, database_name, schema_name, table_name, merge_key=None): full_table_name = f"{database_name}.{schema_name}.{table_name}" write_type = write_type.lower().strip()
if write_type == "overwrite" or write_type == "append": try: df.write.format("delta").mode("overwrite").option("mergeSchema", "true").saveAsTable(full_table_name) error_message = "" row_count = df.count() except Exception as e: error_message = str(e) row_count = 0
elif write_type == "merge": if not merge_key: raise ValueError("merge_key is required when write_type is 'merge'.")
spark = df.sparkSession spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")
if not spark.catalog.tableExists(full_table_name): try: df.write.format("delta").mode("overwrite").option("mergeSchema", "true").saveAsTable(full_table_name) error_message = "" row_count = df.count() except Exception as e: error_message = str(e) row_count = 0 else: try: keys = [key.strip() for key in merge_key.split(",")] merge_condition = " AND ".join(f"target.{key} = source.{key}" for key in keys)
target = DeltaTable.forName(spark, full_table_name) target.alias("target").merge(df.alias("source"), merge_condition).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
error_message = "" row_count = df.count() except Exception as e: error_message = str(e) row_count = 0
else: raise ValueError( f"Unknown write_type: {write_type!r}. Expected 'overwrite', 'append', or 'merge'." ) return error_message, row_count |
Logging the Results
This last part logs the results. It is disabled in this example but necessary once you build this out for real work. You will also need to update the audit_log_path variable to the path of your choice. The error_message, if one exists, comes from the write_to_lakehouse function above.
def audit_log(destination_database, destination_schema, destination_table, watermark_start_time, error_message=None, row_count=None): audit_log_path = f"{destination_database}.logs.process_log"
if error_message == "" or error_message is None: status = "Success" else: status = "Failure"
log = { "destination_database": destination_database, "destination_schema": destination_schema, "destination_table": destination_table, "watermark_start_time": watermark_start_time, "watermark_end_time": datetime.now(timezone.utc).isoformat(), "status": status, "error_message": error_message, "row_count": row_count, }
df = spark.createDataFrame([log]) df.write.format("delta").mode("append").option("mergeSchema", "true").saveAsTable(audit_log_path) |
Putting It All Together
Now here is the code that puts it all together.
1. Loop through all the rows in the config DataFrame to get parameters.
2. Build the URL from the BASE_URL and END_POINT.
3. Create your parameters.
4. Create the header and get the token from the key vault (not applicable in this example).
5. Create the watermark_start_time for the audit log.
6. Collect the JSON data from the API call.
7. Use the json_normalize method to break apart the objects into their own columns.
8. Choose just the columns you want to store.
9. Add a primary key if needed.
10. Convert the data from a Pandas DataFrame to a Spark DataFrame (commented out).
11. Write the DataFrame to your lakehouse (commented out).
12. Log the results in your log table (commented out).
for index, row in config_df.iterrows(): BASE_URL = row["base_url"] END_POINT = row["end_point"] DESTINATION_DATABASE = row["destination_database"] DESTINATION_SCHEMA = row["destination_schema"] DESTINATION_TABLE = row["destination_table"] WRITE_TYPE = row["write_type"] PRIMARY_KEY = row["primary_key"] COLUMNS_NEEDED = row["columns_needed"].replace(" ", "").split(",") KEY_VAULT_URL = row["key_vault_url"] KEY_VAULT_SECRET = row["key_vault_secret"] RETRIES_ALLOWED = row["retries_allowed"] GET_RESULTS = row["get_results"] PAGINATION = row["pagination"] API_PARAMETER_COLUMNS = row["api_parameter_columns"].replace(" ", "").split(",") NAME = row["name"] STATUS = row["status"] SPECIES = row["species"] TYPE = row["type"] GENDER = row["gender"] DIMENSION = row["dimension"] EPISODE = row["episode"]
URL = f"{BASE_URL}/{END_POINT}" PARAMS = build_params(row, API_PARAMETER_COLUMNS)
headers = get_auth_header(KEY_VAULT_URL, KEY_VAULT_SECRET)
watermark_start_time = datetime.now(timezone.utc).isoformat() results, error_message = get_data(URL, PARAMS, RETRIES_ALLOWED, PAGINATION, GET_RESULTS)
if results is not None: df_original = pd.json_normalize(results) df_correct_columns = df_original[COLUMNS_NEEDED] df_primary_key = add_primary_key(df_correct_columns, PRIMARY_KEY) display(df_primary_key)
# spark_df = spark.createDataFrame(df_primary_key) # write_error, row_count = write_to_lakehouse(spark_df, WRITE_TYPE, DESTINATION_DATABASE, DESTINATION_SCHEMA, DESTINATION_TABLE, PRIMARY_KEY) # audit_log(DESTINATION_DATABASE, DESTINATION_SCHEMA, DESTINATION_TABLE, watermark_start_time, write_error, row_count) else: print(f"{DESTINATION_TABLE}: failed after {RETRIES_ALLOWED} attempts - {error_message}") # audit_log(DESTINATION_DATABASE, DESTINATION_SCHEMA, DESTINATION_TABLE, watermark_start_time, error_message, 0) |
Writer’s note: the initial pull happens through Pandas because a single API call is synchronous, one request and response at a time, so there is no distributed work for Spark to do yet. Pandas handles that scale easily. Once the data is collected, it gets converted into a Spark DataFrame so it can be written to your lakehouse.
Wrapping Up
This script did not come together overnight. It is the product of years of failed pulls, silent pagination bugs, and API calls that quietly returned half the data they should have. Copy it, paste it, and skip that whole process. Adjust it freely to fit your own sources, and treat the config table as the only place you should ever need to make changes. Keep in mind that every API is a little different, so some parts will need to be adjusted. This is a great starter template, not a one-size-fits-all script.
Comments