Data Ingestion: How Python Reads Files and REST APIs

Data Sources: Why Diverse Ingestion Matters

Data for analysis rarely originates in a single, uniform format or from a sole location. Operational systems, third-party services, and internal applications each generate data optimized for their specific function. A transactional database produces structured records, while a web server logs access patterns as semi-structured text or JSON. An external vendor might expose data through a REST API.

This leads to a range of common formats. CSV files often contain tabular exports from legacy systems. JSON is prevalent for API responses, configuration, and document-oriented data. XML appears in older enterprise integrations. Modern analytical systems frequently use binary formats like Parquet or Avro for efficient storage and query performance. Each format has distinct parsing requirements and internal structures.

Beyond format, data resides in diverse locations. Files might be local, on network file systems, or in object storage services like AWS S3 or Azure Blob Storage. Data streams from message queues such as Apache Kafka. Real-time data access often occurs via HTTP/REST endpoints from web services. Direct database connections to relational (e.g., PostgreSQL, MySQL) and NoSQL stores (e.g., MongoDB, Cassandra) are also common sources.

A data ingestion system must not assume homogeneity. It requires the capability to connect to varied endpoints, authenticate against different services, and correctly interpret multiple data representations. Building a system that only processes local CSV files, for example, would exclude critical business intelligence from web APIs, operational databases, or real-time event streams.

Failing to accommodate this diversity results in incomplete datasets and limits the scope of analysis. Data engineers must design pipelines that adapt to the inherent heterogeneity of modern data landscapes, ensuring all relevant information can be brought into the analytical environment.

File Parsing: Python Tools for CSV, JSON, XML

Structured data often resides in common file formats, requiring specific Python tools for efficient ingestion. CSV, JSON, and XML files each have dedicated standard library modules and external packages for parsing their contents into Python objects, facilitating their transformation into usable data structures.

For Comma Separated Values (CSV) files, Python’s csv module provides direct, row-oriented access. The csv.reader object iterates over lines, treating each as a list of strings. When a header row is present, csv.DictReader is often preferred; it maps each subsequent row to a dictionary where column headers serve as keys, simplifying data access by name rather than index.

# data.csv
# name,age,city
# Alice,30,New York
# Bob,24,London

import csv

with open('data.csv', mode='r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row)

{'name': 'Alice', 'age': '30', 'city': 'New York'}
{'name': 'Bob', 'age': '24', 'city': 'London'}

For larger CSV datasets, the pandas library offers pandas.read_csv, which provides significant performance benefits and loads data directly into a DataFrame. This approach requires an additional dependency (pip install pandas) but provides robust type inference, handles missing values, and offers extensive data manipulation capabilities immediately after ingestion, which is often crucial for subsequent analysis.

import pandas as pd

df = pd.read_csv('data.csv')
print(df)

    name  age      city
0  Alice   30  New York
1    Bob   24    London

JSON (JavaScript Object Notation) files map directly to Python dictionaries and lists, making their parsing straightforward. The standard library json module handles this serialization and deserialization. json.load() reads a JSON document from a file-like object and parses it into a native Python object, preserving its hierarchical structure and data types.

# data.json
# [
#   {"name": "Alice", "age": 30, "city": "New York"},
#   {"name": "Bob", "age": 24, "city": "London"}
# ]

import json

with open('data.json', mode='r', encoding='utf-8') as f:
    data = json.load(f)
    print(data)
    print(type(data))

[{'name': 'Alice', 'age': 30, 'city': 'New York'}, {'name': 'Bob', 'age': 24, 'city': 'London'}]
<class 'list'>

XML (Extensible Markup Language) parsing in Python commonly uses the xml.etree.ElementTree module. This module provides an ElementTree API, treating the XML document as a tree of elements. Accessing data involves navigating this tree structure using methods like find() and findall() to locate specific tags and extract their text content or attributes. This hierarchical nature can make XML parsing more verbose compared to flat CSV or simple JSON structures, often requiring more explicit pathing to target specific data points within the document.

# data.xml
# <people>
#     <person>
#         <name>Alice</name>
#         <age>30</age>
#         <city>New York</city>
#     </person>
#     <person>
#         <name>Bob</name>
#         <age>24</age>
#         <city>London</city>
#     </person>
# </people>

import xml.etree.ElementTree as ET

tree = ET.parse('data.xml')
root = tree.getroot()

for person in root.findall('person'):
    name = person.find('name').text
    age = person.find('age').text
    city = person.find('city').text
    print(f"Name: {name}, Age: {age}, City: {city}")

Name: Alice, Age: 30, City: New York
Name: Bob, Age: 24, City: London

REST APIs: Python Integration Patterns

REST APIs provide structured access to remote data over HTTP. Python interacts with these services primarily using the requests library, which simplifies HTTP operations compared to urllib. Install it with pip install requests.

To retrieve data, send an HTTP GET request to the API endpoint. The requests.get() function returns a Response object containing the server’s reply. This object holds the status code, headers, and the response body.

import requests

api_url = "https://jsonplaceholder.typicode.com/posts/1"
response = requests.get(api_url)

print(f"Status Code: {response.status_code}")
print(f"Content Type: {response.headers['Content-Type']}")

Status Code: 200
Content Type: application/json; charset=utf-8

Most REST APIs return data in JSON format. The Response object’s .json() method parses this content directly into a Python dictionary or list. This conversion handles common encoding issues and provides immediate access to the data structure.

import requests

api_url = "https://jsonplaceholder.typicode.com/posts"
response = requests.get(api_url)

if response.status_code == 200:
    posts = response.json()
    # Access the first post's title
    if posts:
        print(f"First post title: {posts[0]['title']}")
else:
    print(f"Error fetching data: {response.status_code}")

First post title: sunt aut facere repellat provident occaecati excepturi optio reprehenderit

API interactions often require parameters, such as filters or pagination settings. Pass these as a dictionary to the params argument in requests.get(). The library appends these as query string parameters to the URL.

import requests

api_url = "https://jsonplaceholder.typicode.com/comments"
params = {"postId": 1}
response = requests.get(api_url, params=params)

if response.status_code == 200:
    comments = response.json()
    print(f"Comments for postId 1: {len(comments)}")
    if comments:
        print(f"First comment email: {comments[0]['email']}")

Comments for postId 1: 5
First comment email: Eliseo@gardner.biz

Authentication for REST APIs typically uses API keys, tokens, or OAuth. For simple API key authentication, pass the key in the URL parameters or as a custom header. Tokens often go into the Authorization header with a Bearer prefix. Using headers is generally more secure than URL parameters as they are not logged by default.

import requests

# Example for an API key in headers (replace 'YOUR_API_KEY' with a real key)
# auth_headers = {"Authorization": "Bearer YOUR_API_KEY"}
# response = requests.get("https://api.example.com/data", headers=auth_headers)

# For APIs expecting a key in params
# auth_params = {"api_key": "YOUR_API_KEY"}
# response = requests.get("https://api.example.com/data", params=auth_params)

Error handling involves checking the response.status_code. A 200 indicates success. Codes in the 4xx range signify client errors (e.g., 401 Unauthorized, 404 Not Found), while 5xx codes indicate server errors. For more granular error checking, response.raise_for_status() raises an HTTPError for bad responses, simplifying error flow. This makes code more concise but loses the ability to handle specific status codes differently without a try-except block for HTTPError.

Ingestion Errors: What Breaks and How to Fix

Data ingestion processes fail due to external factors like network outages, incorrect file paths, or malformed data. Anticipating these failures and implementing specific error handling is key to building reliable data pipelines. Ignoring potential issues leads to silent failures or crashes, halting downstream processes.

File system operations commonly encounter FileNotFoundError if a path does not exist, or PermissionError if the process lacks read access to a file or directory. Incorrect file encoding can also cause UnicodeDecodeError when attempting to read the file content. Python’s try...except blocks handle these exceptions predictably.

import logging

logging.basicConfig(level=logging.WARNING, format='%(levelname)s: %(message)s')

def read_text_file(path: str) -> str | None:
    """Reads a text file and returns its content, handling common file errors."""
    try:
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        logging.warning(f"File not found: {path}. Skipping.")
        return None
    except PermissionError:
        logging.warning(f"Permission denied for file: {path}. Skipping.")
        return None
    except UnicodeDecodeError:
        logging.warning(f"Encoding error reading file: {path}. Expected UTF-8.")
        return None

# Example of expected output for a missing file:
# WARNING: File not found: non_existent.txt. Skipping.

When consuming REST APIs, network connectivity issues often manifest as requests.exceptions.ConnectionError or requests.exceptions.Timeout. The API server itself might return HTTP status codes indicating client errors (4xx) or server errors (5xx). The requests library can convert these HTTP errors into requests.exceptions.HTTPError using response.raise_for_status().

import requests
import logging

# Assumes logging.basicConfig from above is already set up

def fetch_json_data(url: str, timeout_sec: int = 10) -> dict | None:
    """Fetches JSON data from a URL, handling network and HTTP errors."""
    try:
        response = requests.get(url, timeout=timeout_sec)
        response.raise_for_status()  # Raises HTTPError for 4xx/5xx responses
        return response.json()
    except requests.exceptions.ConnectionError:
        logging.warning(f"Network error connecting to {url}. Check connectivity.")
        return None
    except requests.exceptions.Timeout:
        logging.warning(f"API request to {url} timed out after {timeout_sec}s.")
        return None
    except requests.exceptions.HTTPError as e:
        logging.warning(f"HTTP error {e.response.status_code} from {url}.")
        return None
    except ValueError: # If response.json() fails due to non-JSON content
        logging.warning(f"API response from {url} was not valid JSON.")
        return None
    except requests.exceptions.RequestException as e: # Catch other requests errors
        logging.warning(f"An unexpected request error occurred for {url}: {e}")
        return None

Transient errors like network glitches or temporary API unavailability can sometimes resolve themselves. For these cases, implementing a retry mechanism with exponential backoff can improve ingestion success rates. This adds complexity but increases overall system reliability.

Even after successful ingestion, the data itself might be malformed or incomplete. Implement post-ingestion validation steps to check data types, ranges, or schema conformity. This ensures that the data meets expectations before it moves to downstream processing.

Data Ingestion: Build a Multi-Source Pipeline

Real-world data ingestion rarely involves a single source. Combining information from local files, external APIs, and databases into a unified dataset is a common task. This section constructs a basic pipeline that integrates product details from a CSV file with live stock levels fetched from a REST API.

Consider a scenario where product identifiers and names reside in data/products.csv. Stock availability for each product is provided by an external service via a REST endpoint.

The data/products.csv file contains:

product_id,product_name
P001,Widget A
P002,Gadget B
P003,Thing C

An API endpoint, https://api.example.com/stock/{product_id}, returns JSON data for a given product ID. A request for P001 might yield:

{
  "product_id": "P001",
  "stock_level": 15,
  "last_updated": "2023-10-27T10:30:00Z"
}

The pipeline first reads the CSV file to obtain the base product information. For each product, it then makes an HTTP GET request to the stock API. Finally, it merges the API response into the product data.

import pandas as pd
import requests
import os

# Ensure data directory exists for the CSV
os.makedirs('data', exist_ok=True)

# Create a dummy CSV file for demonstration
csv_content = """product_id,product_name
P001,Widget A
P002,Gadget B
P003,Thing C
"""
with open('data/products.csv', 'w') as f:
    f.write(csv_content)

def fetch_stock_level(product_id: str) -> dict:
    """Fetches stock level for a given product_id from a mock API."""
    # In a real scenario, this would be requests.get(f"https://api.example.com/stock/{product_id}").json()
    # For demonstration, we simulate API responses.
    mock_stock_data = {
        "P001": {"stock_level": 15, "last_updated": "2023-10-27T10:30:00Z"},
        "P002": {"stock_level": 0, "last_updated": "2023-10-27T10:31:00Z"},
        "P003": {"stock_level": 7, "last_updated": "2023-10-27T10:32:00Z"},
    }
    return mock_stock_data.get(product_id, {"stock_level": -1, "last_updated": None})

# Ingest product data from CSV
products_df = pd.read_csv('data/products.csv')

# Ingest stock data from API and merge
product_data_list = products_df.to_dict(orient='records')
combined_data = []

for product in product_data_list:
    product_id = product['product_id']
    stock_info = fetch_stock_level(product_id)
    product.update(stock_info) # Merge stock information into the product dictionary
    combined_data.append(product)

# Convert back to DataFrame for easier manipulation/storage
final_df = pd.DataFrame(combined_data)
print(final_df)

  product_id product_name  stock_level         last_updated
0       P001    Widget A           15  2023-10-27T10:30:00Z
1       P002    Gadget B            0  2023-10-27T10:31:00Z
2       P003     Thing C            7  2023-10-27T10:32:00Z