Learn how to consume REST APIs, handle HTTP requests, process JSON data, and build API clients using Python's requests library.
APIs (Application Programming Interfaces) are essential for modern applications, allowing different services to communicate. This lesson teaches you how to work with REST APIs in Python using the requests library, handle various response formats, and build robust API clients.
Understanding RESTful architecture and HTTP methods is fundamental to working with modern web services.
REST (Representational State Transfer) is an architectural style for designing networked applications. HTTP methods define the type of operation:
HTTP Methods:
1import requests
2
3# GET - Retrieve data
4response = requests.get('https://api.example.com/users')
5users = response.json()
6
7# POST - Create new resource
8new_user = {'name': 'John', 'email': '[email protected]'}
9response = requests.post('https://api.example.com/users', json=new_user)
10
11# PUT - Update existing resource
12updated_user = {'name': 'John Smith', 'email': '[email protected]'}
13response = requests.put('https://api.example.com/users/1', json=updated_user)
14
15# DELETE - Remove resource
16response = requests.delete('https://api.example.com/users/1')HTTP Status Codes:
Understanding these methods and status codes is crucial for effective API interaction.
The requests library is Python's most popular HTTP library, making API consumption straightforward and reliable.
Basic Request Operations:
1import requests
2
3# Simple GET request
4response = requests.get('https://api.github.com/users/octocat')
5print(response.status_code) # 200
6print(response.json()) # Parsed JSON data
7
8# Request with parameters
9params = {'q': 'python', 'sort': 'stars'}
10response = requests.get('https://api.github.com/search/repositories', params=params)
11
12# Request with headers
13headers = {
14 'Authorization': 'Bearer your_token_here',
15 'Accept': 'application/json'
16}
17response = requests.get('https://api.example.com/data', headers=headers)
18
19# POST request with JSON data
20data = {'name': 'New Repository', 'description': 'A test repo'}
21response = requests.post('https://api.github.com/user/repos',
22 json=data, headers=headers)Advanced Features:
The requests library handles complex networking tasks automatically, making it the preferred choice for API integration.
JSON (JavaScript Object Notation) is the standard format for API data exchange. Python provides excellent support for working with JSON data.
Working with JSON Responses:
1import requests
2import json
3
4# Get JSON data from API
5response = requests.get('https://jsonplaceholder.typicode.com/posts/1')
6data = response.json() # Automatically parse JSON
7
8print(data['title']) # Access nested data
9print(data['userId']) # Navigate JSON structure
10
11# Handle nested JSON structures
12user_response = requests.get('https://jsonplaceholder.typicode.com/users/1')
13user_data = user_response.json()
14
15# Safely access nested data
16address = user_data.get('address', {})
17city = address.get('city', 'Unknown')
18coordinates = address.get('geo', {})
19latitude = coordinates.get('lat', 0)
20
21# Convert Python objects to JSON
22user_info = {
23 'name': 'Alice Johnson',
24 'age': 30,
25 'skills': ['Python', 'JavaScript', 'SQL']
26}
27json_string = json.dumps(user_info, indent=2)
28print(json_string)
29
30# Parse JSON string back to Python object
31parsed_data = json.loads(json_string)Best Practices:
Understanding JSON navigation and safe access patterns is essential for robust API clients.
Proper API clients must handle various error scenarios gracefully and implement best practices for reliable operation.
Comprehensive Error Handling:
1import requests
2import time
3from requests.adapters import HTTPAdapter
4from urllib3.util.retry import Retry
5
6def make_api_request(url, max_retries=3):
7 # Configure retry strategy
8 retry_strategy = Retry(
9 total=max_retries,
10 backoff_factor=1,
11 status_forcelist=[429, 500, 502, 503, 504]
12 )
13
14 adapter = HTTPAdapter(max_retries=retry_strategy)
15 session = requests.Session()
16 session.mount("http://", adapter)
17 session.mount("https://", adapter)
18
19 try:
20 response = session.get(url, timeout=10)
21 response.raise_for_status() # Raise exception for bad status codes
22 return response.json()
23
24 except requests.exceptions.ConnectionError:
25 print("Failed to connect to the API")
26 return None
27 except requests.exceptions.Timeout:
28 print("Request timed out")
29 return None
30 except requests.exceptions.HTTPError as e:
31 if response.status_code == 401:
32 print("Authentication failed")
33 elif response.status_code == 429:
34 print("Rate limit exceeded")
35 else:
36 print(f"HTTP error: {e}")
37 return None
38 except requests.exceptions.RequestException as e:
39 print(f"Request failed: {e}")
40 return None
41
42# Rate limiting implementation
43class RateLimiter:
44 def __init__(self, max_requests, time_window):
45 self.max_requests = max_requests
46 self.time_window = time_window
47 self.requests = []
48
49 def wait_if_needed(self):
50 now = time.time()
51 # Remove old requests outside time window
52 self.requests = [req_time for req_time in self.requests
53 if now - req_time < self.time_window]
54
55 if len(self.requests) >= self.max_requests:
56 sleep_time = self.time_window - (now - self.requests[0])
57 if sleep_time > 0:
58 time.sleep(sleep_time)
59
60 self.requests.append(now)Best Practices:
Following these practices ensures your API clients are robust, respectful, and maintainable.
Build a comprehensive weather dashboard that consumes multiple APIs and handles various data formats
1import requests
2import json
3import time
4from datetime import datetime, timedelta
5from typing import Dict, List, Optional, Any
6
7class APIError(Exception):
8 """Custom exception for API-related errors"""
9 def __init__(self, message: str, status_code: Optional[int] = None):
10 self.message = message
11 self.status_code = status_code
12 super().__init__(self.message)
13
14class RateLimiter:
15 """Simple rate limiter for API requests"""
16 def __init__(self, max_requests: int, time_window: int):
17 self.max_requests = max_requests
18 self.time_window = time_window # seconds
19 self.requests = []
20
21 def wait_if_needed(self):
22 """Wait if rate limit would be exceeded"""
23 now = time.time()
24 # Remove old requests outside the time window
25 self.requests = [req_time for req_time in self.requests
26 if now - req_time < self.time_window]
27
28 if len(self.requests) >= self.max_requests:
29 sleep_time = self.time_window - (now - self.requests[0])
30 if sleep_time > 0:
31 print(f"Rate limit reached. Waiting {sleep_time:.1f} seconds...")
32 time.sleep(sleep_time)
33
34 self.requests.append(now)
35
36class WeatherAPI:
37 """Weather API client with comprehensive error handling"""
38
39 def __init__(self, api_key: str):
40 self.api_key = api_key
41 self.base_url = "https://api.openweathermap.org/data/2.5"
42 self.rate_limiter = RateLimiter(max_requests=10, time_window=60)
43 self.session = requests.Session()
44 # Set common headers
45 self.session.headers.update({
46 'User-Agent': 'WeatherDashboard/1.0',
47 'Accept': 'application/json'
48 })
49
50 def _make_request(self, endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
51 """Make HTTP request with error handling and rate limiting"""
52 self.rate_limiter.wait_if_needed()
53
54 # Add API key to parameters
55 params['appid'] = self.api_key
56 params['units'] = 'metric' # Use Celsius
57
58 url = f"{self.base_url}/{endpoint}"
59
60 try:
61 response = self.session.get(url, params=params, timeout=10)
62 response.raise_for_status() # Raise exception for bad status codes
63
64 return response.json()
65
66 except requests.exceptions.ConnectionError:
67 raise APIError("Failed to connect to weather service")
68 except requests.exceptions.Timeout:
69 raise APIError("Request timed out")
70 except requests.exceptions.HTTPError as e:
71 if response.status_code == 401:
72 raise APIError("Invalid API key", 401)
73 elif response.status_code == 404:
74 raise APIError("Location not found", 404)
75 elif response.status_code == 429:
76 raise APIError("Rate limit exceeded", 429)
77 else:
78 raise APIError(f"HTTP error: {e}", response.status_code)
79 except json.JSONDecodeError:
80 raise APIError("Invalid JSON response from server")
81 except Exception as e:
82 raise APIError(f"Unexpected error: {str(e)}")
83
84 def get_current_weather(self, city: str) -> Dict[str, Any]:
85 """Get current weather for a city"""
86 params = {'q': city}
87 data = self._make_request('weather', params)
88
89 return {
90 'city': data['name'],
91 'country': data['sys']['country'],
92 'temperature': round(data['main']['temp'], 1),
93 'feels_like': round(data['main']['feels_like'], 1),
94 'humidity': data['main']['humidity'],
95 'pressure': data['main']['pressure'],
96 'description': data['weather'][0]['description'].title(),
97 'wind_speed': data['wind']['speed'],
98 'wind_direction': data.get('wind', {}).get('deg', 0),
99 'visibility': data.get('visibility', 0) / 1000, # Convert to km
100 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
101 }
102
103 def get_forecast(self, city: str, days: int = 5) -> List[Dict[str, Any]]:
104 """Get weather forecast for specified days"""
105 params = {'q': city, 'cnt': days * 8} # 8 forecasts per day (3-hour intervals)
106 data = self._make_request('forecast', params)
107
108 daily_forecasts = {}
109 for item in data['list']:
110 date = datetime.fromtimestamp(item['dt']).date()
111
112 if date not in daily_forecasts:
113 daily_forecasts[date] = {
114 'date': date.strftime('%Y-%m-%d'),
115 'temperatures': [],
116 'descriptions': [],
117 'humidity_values': [],
118 'wind_speeds': []
119 }
120
121 daily_forecasts[date]['temperatures'].append(item['main']['temp'])
122 daily_forecasts[date]['descriptions'].append(item['weather'][0]['description'])
123 daily_forecasts[date]['humidity_values'].append(item['main']['humidity'])
124 daily_forecasts[date]['wind_speeds'].append(item['wind']['speed'])
125
126 # Process daily summaries
127 forecast_list = []
128 for date_data in list(daily_forecasts.values())[:days]:
129 temps = date_data['temperatures']
130 forecast_list.append({
131 'date': date_data['date'],
132 'temperature_min': round(min(temps), 1),
133 'temperature_max': round(max(temps), 1),
134 'temperature_avg': round(sum(temps) / len(temps), 1),
135 'description': max(set(date_data['descriptions']),
136 key=date_data['descriptions'].count).title(),
137 'humidity_avg': round(sum(date_data['humidity_values']) / len(date_data['humidity_values'])),
138 'wind_speed_avg': round(sum(date_data['wind_speeds']) / len(date_data['wind_speeds']), 1)
139 })
140
141 return forecast_list
142
143class WeatherDashboard:
144 """Weather dashboard combining multiple data sources"""
145
146 def __init__(self, api_key: str):
147 self.weather_api = WeatherAPI(api_key)
148 self.favorites = []
149 self.cache = {}
150 self.cache_duration = 600 # 10 minutes
151
152 def add_favorite_city(self, city: str):
153 """Add city to favorites list"""
154 if city not in self.favorites:
155 self.favorites.append(city)
156 print(f"Added {city} to favorites")
157
158 def remove_favorite_city(self, city: str):
159 """Remove city from favorites"""
160 if city in self.favorites:
161 self.favorites.remove(city)
162 print(f"Removed {city} from favorites")
163
164 def _is_cache_valid(self, cache_key: str) -> bool:
165 """Check if cached data is still valid"""
166 if cache_key not in self.cache:
167 return False
168
169 cache_time = self.cache[cache_key]['timestamp']
170 return (datetime.now() - cache_time).seconds < self.cache_duration
171
172 def get_weather_with_cache(self, city: str) -> Dict[str, Any]:
173 """Get weather data with caching"""
174 cache_key = f"current_{city}"
175
176 if self._is_cache_valid(cache_key):
177 print(f"Using cached data for {city}")
178 return self.cache[cache_key]['data']
179
180 try:
181 weather_data = self.weather_api.get_current_weather(city)
182 self.cache[cache_key] = {
183 'data': weather_data,
184 'timestamp': datetime.now()
185 }
186 return weather_data
187
188 except APIError as e:
189 print(f"Error getting weather for {city}: {e.message}")
190 return None
191
192 def display_current_weather(self, city: str):
193 """Display formatted current weather information"""
194 weather = self.get_weather_with_cache(city)
195 if not weather:
196 return
197
198 print(f"\n=== Current Weather in {weather['city']}, {weather['country']} ===")
199 print(f"Temperature: {weather['temperature']}°C (feels like {weather['feels_like']}°C)")
200 print(f"Conditions: {weather['description']}")
201 print(f"Humidity: {weather['humidity']}%")
202 print(f"Pressure: {weather['pressure']} hPa")
203 print(f"Wind: {weather['wind_speed']} m/s")
204 print(f"Visibility: {weather['visibility']} km")
205 print(f"Last updated: {weather['timestamp']}")
206
207 def display_forecast(self, city: str, days: int = 3):
208 """Display weather forecast"""
209 try:
210 forecast = self.weather_api.get_forecast(city, days)
211 print(f"\n=== {days}-Day Forecast for {city} ===")
212
213 for day in forecast:
214 print(f"\n{day['date']}:")
215 print(f" Temperature: {day['temperature_min']}°C - {day['temperature_max']}°C (avg: {day['temperature_avg']}°C)")
216 print(f" Conditions: {day['description']}")
217 print(f" Humidity: {day['humidity_avg']}%")
218 print(f" Wind: {day['wind_speed_avg']} m/s")
219
220 except APIError as e:
221 print(f"Error getting forecast for {city}: {e.message}")
222
223 def display_favorites_summary(self):
224 """Display weather summary for all favorite cities"""
225 if not self.favorites:
226 print("No favorite cities added yet.")
227 return
228
229 print("\n=== Favorites Weather Summary ===")
230 for city in self.favorites:
231 weather = self.get_weather_with_cache(city)
232 if weather:
233 print(f"{city}: {weather['temperature']}°C, {weather['description']}")
234 else:
235 print(f"{city}: Unable to fetch weather data")
236
237 def compare_cities(self, cities: List[str]):
238 """Compare weather across multiple cities"""
239 print("\n=== City Weather Comparison ===")
240 city_data = []
241
242 for city in cities:
243 weather = self.get_weather_with_cache(city)
244 if weather:
245 city_data.append(weather)
246
247 if not city_data:
248 print("No weather data available for comparison")
249 return
250
251 # Find extremes
252 hottest = max(city_data, key=lambda x: x['temperature'])
253 coldest = min(city_data, key=lambda x: x['temperature'])
254 most_humid = max(city_data, key=lambda x: x['humidity'])
255 windiest = max(city_data, key=lambda x: x['wind_speed'])
256
257 print(f"Hottest: {hottest['city']} ({hottest['temperature']}°C)")
258 print(f"Coldest: {coldest['city']} ({coldest['temperature']}°C)")
259 print(f"Most humid: {most_humid['city']} ({most_humid['humidity']}%)")
260 print(f"Windiest: {windiest['city']} ({windiest['wind_speed']} m/s)")
261
262# Example usage demonstrating comprehensive API integration
263def main():
264 """Main function demonstrating the weather dashboard"""
265 # Note: In a real application, use environment variables for API keys
266 api_key = "your_openweathermap_api_key_here"
267
268 try:
269 dashboard = WeatherDashboard(api_key)
270
271 # Add favorite cities
272 dashboard.add_favorite_city("London")
273 dashboard.add_favorite_city("Tokyo")
274 dashboard.add_favorite_city("New York")
275
276 # Display current weather
277 dashboard.display_current_weather("London")
278
279 # Display forecast
280 dashboard.display_forecast("London", 3)
281
282 # Show favorites summary
283 dashboard.display_favorites_summary()
284
285 # Compare cities
286 dashboard.compare_cities(["London", "Tokyo", "New York", "Sydney"])
287
288 print("\n=== API Integration Complete ===")
289 print("This example demonstrates:")
290 print("- HTTP request handling with error management")
291 print("- JSON data processing and navigation")
292 print("- Rate limiting and caching strategies")
293 print("- Object-oriented API client design")
294 print("- Comprehensive error handling patterns")
295
296 except Exception as e:
297 print(f"Application error: {e}")
298
299if __name__ == "__main__":
300 main()Your progress is saved in this browser, on this device only.