Python - Working with APIs for Automation

Working with APIs for Automation in Python

Modern software frequently relies on APIs—Application Programming Interfaces—to integrate services, retrieve data, and automate workflows. In Python, working with APIs is both accessible and powerful. Whether you want to pull weather data, automate deployments, interact with web services, or manage your own resources, Python provides robust tools like requests, httpx, and libraries for handling authentication, parsing data, and scheduling tasks.

This document provides an in-depth guide to using Python for API-driven automation. It covers:

• Understanding APIs and HTTP basics
• Working with GET, POST, PUT, DELETE
• Authentication strategies (API keys, OAuth2)
• Error handling and retries
• Pagination and rate limits
• Parsing JSON/XML
• Data storage and automation pipelines
• Scripting and scheduling for real-world automation

1. Introduction to APIs and HTTP

1.1 What is an API?

An API allows two systems to communicate via a well-defined interface. Web APIs typically use HTTP, enabling access to data and functionality over the internet. Common examples include:

• Weather data providers
• Social media integration
• Cloud infrastructure control
• Payment gateways

1.2 HTTP Methods Explained

• GET: Retrieve resources
• POST: Create or trigger actions
• PUT/PATCH: Update resources
• DELETE: Remove resources

1.3 Status Codes Overview

Key HTTP status codes:

• 200 Range: Success
• 201 Created
• 204 No Content
• 400 Bad Request
• 401 Unauthorized
• 403 Forbidden
• 404 Not Found
• 429 Too Many Requests
• 500–599 Server errors

2. Setting Up Your Python Environment

2.1 Installing the Requests Library

pip install requests

2.2 Basic GET Request Example

import requests

response = requests.get('https://api.example.com/data')
print(response.status_code)
print(response.json())

2.3 Handling Query Parameters

params = {'q': 'python', 'limit': 10}
response = requests.get('https://api.example.com/search', params=params)
print(response.url)
print(response.json())

3. Working with POST, PUT, and DELETE

3.1 POSTing JSON Data

payload = {'name': 'Alice', 'age': 30}
response = requests.post('https://api.example.com/user', json=payload)
print(response.status_code, response.json())

3.2 PUT for Updating Resources

update = {'age': 31}
response = requests.put('https://api.example.com/user/123', json=update)
print(response.status_code)

3.3 DELETE to Remove Resources

response = requests.delete('https://api.example.com/user/123')
print(response.status_code)

4. Handling Authentication

4.1 API Key in Headers or Query

API_KEY = 'your_api_key_here'
headers = {'Authorization': f'Bearer {API_KEY}'}
response = requests.get('https://api.example.com/protected', headers=headers)
print(response.status_code, response.json())

4.2 Basic Auth

from requests.auth import HTTPBasicAuth

response = requests.get('https://api.example.com/private', auth=HTTPBasicAuth('user', 'pass'))
print(response.status_code)

4.3 OAuth2 Authentication Flow

# Simplified example using client credentials grant
import requests

token_url = 'https://api.example.com/oauth2/token'
data = {'grant_type': 'client_credentials', 'client_id': 'xxx', 'client_secret': 'yyy'}
token_resp = requests.post(token_url, data=data)
token = token_resp.json()['access_token']

headers = {'Authorization': f'Bearer {token}'}
resp = requests.get('https://api.example.com/data', headers=headers)
print(resp.json())

5. Pagination and Rate Limiting

5.1 Handling Page-Based APIs

def fetch_all():
    url = 'https://api.example.com/items'
    params = {'page': 1, 'per_page': 100}
    items = []

    while True:
        resp = requests.get(url, params=params)
        data = resp.json()
        items.extend(data['items'])
        if not data.get('next_page'):
            break
        params['page'] += 1

    return items

5.2 Cursor-Based Pagination

def fetch_cursor():
    url = 'https://api.example.com/stream'
    cursor = None
    items = []

    while True:
        resp = requests.get(url, params={'cursor': cursor})
        data = resp.json()
        items.extend(data['items'])
        cursor = data.get('next_cursor')
        if cursor is None:
            break

    return items

5.3 Respecting Rate Limits

import time

response = requests.get(url)
reset = int(response.headers.get('X-Rate-Limit-Reset'))
remaining = int(response.headers.get('X-Rate-Limit-Remaining'))

if remaining == 0:
    wait = reset - time.time()
    time.sleep(max(wait, 0))

6. Error Handling and Retries

6.1 Basic Status Checking

resp = requests.get(url)
if resp.status_code != 200:
    raise Exception(f'Error: {resp.status_code}')

6.2 Automatic Retries with urllib3

from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[429, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))

resp = session.get(url)
print(resp.status_code)

7. Parsing JSON and XML

7.1 Navigating JSON Data

resp = requests.get(url)
data = resp.json()
for item in data['items']:
    print(item['id'], item['name'])

7.2 Parsing XML with ElementTree

import xml.etree.ElementTree as ET

resp = requests.get('https://api.example.com/data.xml')
root = ET.fromstring(resp.content)
for elem in root.findall('.//item'):
    print(elem.find('id').text, elem.find('name').text)

7.3 Handling Nested JSON

resp = requests.get(url)
data = resp.json()['results']
nested = [u['profile']['email'] for u in data]
print(nested)

8. Data Transformation and Storage

8.1 Transform to DataFrame

import pandas as pd

resp = requests.get(url)
entries = resp.json()['items']
df = pd.DataFrame(entries)
print(df.head())

8.2 Export to CSV or JSON

df.to_csv('output.csv', index=False)
df.to_json('output.json', orient='records')

8.3 Store in SQLite

import sqlite3

conn = sqlite3.connect('data.db')
df.to_sql('api_data', conn, if_exists='replace', index=False)
conn.close()

9. Scheduling and Automation

9.1 Using Cron Jobs (Linux/Mac)

# cron entry to run every hour
0 * * * * /usr/bin/python3 /path/to/script.py

9.2 Windows Task Scheduler

Use Task Scheduler to run the Python script at set intervals.

9.3 Using APScheduler for In-Process Scheduling

from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job('interval', minutes=60)
def fetch_and_store():
    resp = requests.get(url)
    df = pd.DataFrame(resp.json()['items'])
    df.to_csv('hourly.csv', index=False)

scheduler.start()

10. Real-World Automation Examples

10.1 Automated Weather Alerts

import requests

API_KEY = 'your_api_key'
url = 'https://api.weather.com/v3/wx/forecast/daily/5day'
params = {'postalKey': '12345:US', 'format': 'json', 'apiKey': API_KEY}

resp = requests.get(url, params=params)
forecast = resp.json()['temperatureMax']
for day, temp in enumerate(forecast):
    if temp > 35:
        print(f'Alert: Day {day} will be over 35°C: {temp}°C')

10.2 Posting to Slack

import requests

SLACK_WEBHOOK = 'https://hooks.slack.com/services/...'

def send_slack(message):
    payload = {'text': message}
    requests.post(SLACK_WEBHOOK, json=payload)

send_slack('Automation script completed successfully!')

10.3 GitHub Issues Reporter

import requests

token = 'ghp_xxx'
headers = {'Authorization': f'token {token}'}

issues = requests.get('https://api.github.com/repos/user/repo/issues', headers=headers).json()
for issue in issues:
    if issue['state'] == 'open':
        print(issue['number'], issue['title'])

11. Testing and Debugging

11.1 Using Mock for Unit Tests

from unittest.mock import patch
import requests

@patch('requests.get')
def test_fetch(mock_get):
    mock_get.return_value.status_code = 200
    mock_get.return_value.json.return_value = {'items': [1, 2]}
    result = fetch_all()  # function from earlier
    assert result == [1, 2]

11.2 Logging and Debug Mode

import logging

logging.basicConfig(level=logging.INFO)
logging.info('Starting API automation')
resp = requests.get(url)
logging.debug(resp.text)

12. Best Practices

• Always handle errors and non-200 status codes
• Securely store API keys (environment variables or vaults)
• Respect rate limits, use retries/backoff
• Parse JSON/XML safely and handle unexpected structures
• Write tests and use mocks for predictable, offline testing
• Modularize your logic into functions and classes
• Schedule your scripts thoughtfully and handle failures gracefully

Python offers a rich ecosystem for automating tasks using APIs, from simple GET requests to full OAuth2 flows and scheduled pipelines. With libraries like requests, httpx, pandas, and APScheduler, you can build robust, scalable automation that interacts with services, handles data, and runs reliably over time.

This guide covered all major aspects required for API-based automation: HTTP mechanics, authentication, pagination, error handling, scheduling, real-world examples, and best practices. Whether you're monitoring systems, integrating services, or building data pipelines, mastering API workflows is a powerful skill that elevates your Python development.