Created
June 2, 2025 09:52
-
-
Save thoughtgeek/6c5df87003380f50d40c11bc91584dd8 to your computer and use it in GitHub Desktop.
Weather API calling and storing to Google Sheets
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
import json | |
from google.oauth2.service_account import Credentials | |
from googleapiclient.discovery import build | |
from datetime import datetime | |
class APIToSheetsWriter: | |
def __init__(self, credentials_file, spreadsheet_id): | |
""" | |
Initialize the Google Sheets API client | |
Args: | |
credentials_file (str): Path to your service account JSON file | |
spreadsheet_id (str): The ID of your Google Sheet | |
""" | |
self.spreadsheet_id = spreadsheet_id | |
# Set up Google Sheets API credentials | |
SCOPES = ['https://www.googleapis.com/auth/spreadsheets'] | |
self.creds = Credentials.from_service_account_file( | |
credentials_file, scopes=SCOPES | |
) | |
self.service = build('sheets', 'v4', credentials=self.creds) | |
self.sheet = self.service.spreadsheets() | |
def call_public_api(self, api_url, headers=None, params=None): | |
""" | |
Make a call to a public API | |
Args: | |
api_url (str): The API endpoint URL | |
headers (dict): Optional headers for the request | |
params (dict): Optional query parameters | |
Returns: | |
dict: JSON response from the API | |
""" | |
try: | |
response = requests.get(api_url, headers=headers, params=params) | |
response.raise_for_status() # Raises an HTTPError for bad responses | |
return response.json() | |
except requests.exceptions.RequestException as e: | |
print(f"Error calling API: {e}") | |
return None | |
def write_to_sheet(self, data, sheet_name="Sheet1", start_cell="A1"): | |
""" | |
Write data to Google Sheets | |
Args: | |
data (list): 2D list of data to write | |
sheet_name (str): Name of the sheet tab | |
start_cell (str): Starting cell (e.g., "A1") | |
""" | |
try: | |
# Clear existing data (optional) | |
range_name = f"{sheet_name}!{start_cell}:Z1000" | |
self.sheet.values().clear( | |
spreadsheetId=self.spreadsheet_id, | |
range=range_name | |
).execute() | |
# Write new data | |
body = { | |
'values': data | |
} | |
result = self.sheet.values().update( | |
spreadsheetId=self.spreadsheet_id, | |
range=f"{sheet_name}!{start_cell}", | |
valueInputOption='RAW', | |
body=body | |
).execute() | |
print(f"Successfully wrote {result.get('updatedCells')} cells to the sheet") | |
return True | |
except Exception as e: | |
print(f"Error writing to sheet: {e}") | |
return False | |
def format_weather_data(self, weather_data): | |
""" | |
Example function to format weather API data for sheets | |
Args: | |
weather_data (dict): Weather data from API | |
Returns: | |
list: Formatted data as 2D list | |
""" | |
if not weather_data: | |
return [] | |
# Create headers | |
headers = ["Timestamp", "City", "Temperature (°C)", "Description", "Humidity (%)", "Wind Speed (m/s)"] | |
# Extract data | |
city = weather_data.get('name', 'Unknown') | |
temp = weather_data.get('main', {}).get('temp', 0) - 273.15 # Convert from Kelvin | |
description = weather_data.get('weather', [{}])[0].get('description', 'N/A') | |
humidity = weather_data.get('main', {}).get('humidity', 0) | |
wind_speed = weather_data.get('wind', {}).get('speed', 0) | |
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
# Format as 2D list | |
data = [ | |
headers, | |
[timestamp, city, round(temp, 1), description, humidity, wind_speed] | |
] | |
return data | |
def format_crypto_data(self, crypto_data): | |
""" | |
Example function to format cryptocurrency API data for sheets | |
Args: | |
crypto_data (list): Crypto data from API | |
Returns: | |
list: Formatted data as 2D list | |
""" | |
if not crypto_data: | |
return [] | |
# Create headers | |
headers = ["Timestamp", "Symbol", "Name", "Price (USD)", "24h Change (%)", "Market Cap"] | |
# Format data | |
data = [headers] | |
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
for coin in crypto_data[:10]: # Limit to top 10 | |
row = [ | |
timestamp, | |
coin.get('symbol', ''), | |
coin.get('name', ''), | |
float(coin.get('current_price', 0)), | |
round(float(coin.get('price_change_percentage_24h', 0)), 2), | |
coin.get('market_cap', 0) | |
] | |
data.append(row) | |
return data | |
def main(): | |
# Configuration | |
CREDENTIALS_FILE = "path/to/your/service-account-key.json" | |
SPREADSHEET_ID = "your-google-sheet-id" | |
# Initialize the writer | |
writer = APIToSheetsWriter(CREDENTIALS_FILE, SPREADSHEET_ID) | |
# Example 1: Weather API (OpenWeatherMap) | |
print("Fetching weather data...") | |
weather_api_key = "your-openweather-api-key" | |
weather_url = f"http://api.openweathermap.org/data/2.5/weather" | |
weather_params = { | |
'q': 'London', | |
'appid': weather_api_key | |
} | |
weather_data = writer.call_public_api(weather_url, params=weather_params) | |
if weather_data: | |
formatted_weather = writer.format_weather_data(weather_data) | |
writer.write_to_sheet(formatted_weather, "Weather", "A1") | |
# Example 2: Cryptocurrency API (CoinGecko - no API key required) | |
print("Fetching cryptocurrency data...") | |
crypto_url = "https://api.coingecko.com/api/v3/coins/markets" | |
crypto_params = { | |
'vs_currency': 'usd', | |
'order': 'market_cap_desc', | |
'per_page': 10, | |
'page': 1 | |
} | |
crypto_data = writer.call_public_api(crypto_url, params=crypto_params) | |
if crypto_data: | |
formatted_crypto = writer.format_crypto_data(crypto_data) | |
writer.write_to_sheet(formatted_crypto, "Crypto", "A1") | |
# Example 3: JSONPlaceholder API (for testing) | |
print("Fetching test data...") | |
test_url = "https://jsonplaceholder.typicode.com/posts" | |
test_data = writer.call_public_api(test_url) | |
if test_data: | |
# Format the test data | |
headers = ["ID", "User ID", "Title", "Body Preview"] | |
formatted_test = [headers] | |
for post in test_data[:5]: # First 5 posts | |
row = [ | |
post.get('id', ''), | |
post.get('userId', ''), | |
post.get('title', '')[:50] + "..." if len(post.get('title', '')) > 50 else post.get('title', ''), | |
post.get('body', '')[:100] + "..." if len(post.get('body', '')) > 100 else post.get('body', '') | |
] | |
formatted_test.append(row) | |
writer.write_to_sheet(formatted_test, "TestData", "A1") | |
if __name__ == "__main__": | |
main() | |
# Installation requirements: | |
# pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib requests | |
# Setup Instructions: | |
""" | |
1. Enable Google Sheets API: | |
- Go to Google Cloud Console (console.cloud.google.com) | |
- Create a new project or select existing one | |
- Enable Google Sheets API | |
- Create service account credentials | |
- Download the JSON key file | |
2. Share your Google Sheet: | |
- Open your Google Sheet | |
- Click "Share" and add the service account email as an editor | |
- Copy the sheet ID from the URL | |
3. Install dependencies: | |
pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib requests | |
4. Update configuration: | |
- Replace CREDENTIALS_FILE with path to your JSON key | |
- Replace SPREADSHEET_ID with your sheet ID | |
- Add API keys for services that require them | |
""" |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment