Python API Integration: Connect Any System to Anything
Modern businesses run on dozens of systems that need to talk to each other. Python makes API integration straightforward—whether connecting your CRM to accounting software or building custom integrations for proprietary systems.
API Integration Fundamentals
HTTP Methods
- GET: Retrieve data
- POST: Create new records
- PUT/PATCH: Update existing records
- DELETE: Remove records
Authentication Types
- API Key: Simple key in header or URL
- Bearer Token: Token in Authorization header
- OAuth 2.0: For user-authorized access
- Basic Auth: Username/password encoded
Making API Requests with Python
Basic GET Request
import requests
response = requests.get(
'https://api.example.com/users',
headers={'Authorization': 'Bearer your-api-key'}
)
if response.status_code == 200:
data = response.json()
print(data)
POST Request with Data
response = requests.post(
'https://api.example.com/users',
headers={
'Authorization': 'Bearer your-api-key',
'Content-Type': 'application/json'
},
json={
'name': 'John Doe',
'email': 'john@example.com'
}
)
if response.status_code == 201:
new_user = response.json()
Building a Reusable API Client
class APIClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
def get(self, endpoint, params=None):
response = self.session.get(
f'{self.base_url}/{endpoint}',
params=params
)
response.raise_for_status()
return response.json()
def post(self, endpoint, data):
response = self.session.post(
f'{self.base_url}/{endpoint}',
json=data
)
response.raise_for_status()
return response.json()
def put(self, endpoint, data):
response = self.session.put(
f'{self.base_url}/{endpoint}',
json=data
)
response.raise_for_status()
return response.json()
# Usage
client = APIClient('https://api.example.com', 'your-api-key')
users = client.get('users')
new_user = client.post('users', {'name': 'Jane'})
Common Integration Patterns
1. Data Sync Between Systems
def sync_customers():
# Get customers from source
crm_client = APIClient(CRM_URL, CRM_KEY)
customers = crm_client.get('customers', {'updated_since': last_sync})
# Update in destination
accounting_client = APIClient(ACCOUNTING_URL, ACCOUNTING_KEY)
for customer in customers:
accounting_client.post('contacts', {
'name': customer['name'],
'email': customer['email'],
'external_id': customer['id']
})
2. Webhook Handler
from flask import Flask, request
app = Flask(__name__)
@app.route('/webhook', methods=['POST'])
def handle_webhook():
data = request.json
if data['event'] == 'order.created':
# Process new order
create_invoice(data['order'])
notify_warehouse(data['order'])
return {'status': 'ok'}
3. Batch Processing
def batch_update_inventory():
products = get_all_products()
for batch in chunks(products, 100):
updates = []
for product in batch:
stock = check_warehouse_stock(product['sku'])
updates.append({
'id': product['id'],
'stock_quantity': stock
})
api_client.post('products/batch-update', {'products': updates})
time.sleep(1) # Rate limiting
Error Handling
def robust_api_call(func, max_retries=3):
for attempt in range(max_retries):
try:
return func()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429: # Rate limited
wait_time = int(e.response.headers.get('Retry-After', 60))
time.sleep(wait_time)
elif e.response.status_code >= 500: # Server error
time.sleep(2 ** attempt) # Exponential backoff
else:
raise
except requests.exceptions.ConnectionError:
time.sleep(2 ** attempt)
raise Exception(f"Failed after {max_retries} retries")
OAuth 2.0 Integration
from requests_oauthlib import OAuth2Session
client_id = 'your-client-id'
client_secret = 'your-client-secret'
authorization_base_url = 'https://api.example.com/oauth/authorize'
token_url = 'https://api.example.com/oauth/token'
# Step 1: Get authorization URL
oauth = OAuth2Session(client_id, redirect_uri='https://yourapp.com/callback')
authorization_url, state = oauth.authorization_url(authorization_base_url)
# Step 2: User authorizes (redirect to authorization_url)
# Step 3: Get token with authorization response
token = oauth.fetch_token(token_url, client_secret=client_secret,
authorization_response=callback_url)
# Step 4: Make authenticated requests
response = oauth.get('https://api.example.com/data')
Best Practices
- Use sessions: Reuse connections for performance
- Handle rate limits: Respect API limits
- Implement retries: Handle transient failures
- Log everything: Track all API calls for debugging
- Secure credentials: Use environment variables
- Validate data: Check responses before processing
Need Custom API Integration?
Our Python team builds robust API integrations connecting your business systems.
From simple data syncs to complex multi-system orchestration, we'll design and implement integrations that work reliably. Contact us to discuss your integration needs.
0 comments