Python Topic
Web Scraping & APIs
Introduction to Web Scraping & APIs
Web scraping is the process of extracting data from websites, while APIs (Application Programming Interfaces) provide structured access to data. Both are essential for data collection in modern applications.
Web Scraping with Requests & BeautifulSoup
1. Setting Up the Environment
python
# Install required libraries
# pip install requests beautifulsoup4 lxml
import requests
from bs4 import BeautifulSoup
import pandas as pd
import time
import json
import csv
from urllib.parse import urljoin, urlparse
2. Basic Web Scraping
python
def basic_web_scraping():
"""Basic web scraping example"""
# Target URL
url = "http://books.toscrape.com/"
try:
# Send GET request
response = requests.get(url)
response.raise_for_status() # Raise exception for bad status codes
# Parse HTML content
soup = BeautifulSoup(response.content, 'html.parser')
# Extract page title
title = soup.find('title').get_text()
print(f"Page Title: {title}")
# Extract all book titles and prices
books = soup.find_all('article', class_='product_pod')
book_data = []
for book in books:
title = book.h3.a['title']
price = book.find('p', class_='price_color').get_text()
availability = book.find('p', class_='instock').get_text().strip()
book_data.append({
'title': title,
'price': price,
'availability': availability
})
# Display results
print(f"\nFound {len(book_data)} books:")
for book in book_data[:5]: # Show first 5
print(f"Title: {book['title']}")
print(f"Price: {book['price']}")
print(f"Availability: {book['availability']}")
print("-" * 50)
return book_data
except requests.RequestException as e:
print(f"Error fetching the page: {e}")
return []
# Run basic scraping
book_data = basic_web_scraping()
3. Advanced Web Scraping with Error Handling
python
class WebScraper:
def __init__(self, delay=1):
self.delay = delay # Delay between requests
self.session = requests.Session()
# Set headers to mimic a real browser
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
def fetch_page(self, url, params=None):
"""Fetch webpage with error handling"""
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
print(f"Error fetching {url}: {e}")
return None
def scrape_quotes(self, base_url="http://quotes.toscrape.com"):
"""Scrape quotes from quotes.toscrape.com"""
quotes_data = []
page = 1
while True:
url = f"{base_url}/page/{page}/"
print(f"Scraping page {page}: {url}")
response = self.fetch_page(url)
if not response:
break
soup = BeautifulSoup(response.content, 'html.parser')
quotes = soup.find_all('div', class_='quote')
# If no quotes found, we've reached the last page
if not quotes:
print("No more quotes found. Ending scraping.")
break
for quote in quotes:
text = quote.find('span', class_='text').get_text()
author = quote.find('small', class_='author').get_text()
tags = [tag.get_text() for tag in quote.find_all('a', class_='tag')]
quotes_data.append({
'quote': text,
'author': author,
'tags': ', '.join(tags),
'page': page
})
# Check for next page
next_button = soup.find('li', class_='next')
if not next_button:
break
page += 1
time.sleep(self.delay) # Be polite to the server
print(f"Scraped {len(quotes_data)} quotes from {page-1} pages")
return quotes_data
def save_to_csv(self, data, filename):
"""Save scraped data to CSV"""
if not data:
print("No data to save!")
return
with open(filename, 'w', newline='', encoding='utf-8') as file:
writer = csv.DictWriter(file, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
print(f"Data saved to {filename}")
def save_to_json(self, data, filename):
"""Save scraped data to JSON"""
with open(filename, 'w', encoding='utf-8') as file:
json.dump(data, file, indent=2, ensure_ascii=False)
print(f"Data saved to {filename}")
# Usage
scraper = WebScraper(delay=1)
quotes_data = scraper.scrape_quotes()
if quotes_data:
scraper.save_to_csv(quotes_data, 'quotes.csv')
scraper.save_to_json(quotes_data, 'quotes.json')
Working with APIs
1. Basic API Requests
python
class APIClient:
def __init__(self, base_url=None):
self.session = requests.Session()
self.base_url = base_url
self.session.headers.update({
'Content-Type': 'application/json',
'User-Agent': 'Python-API-Client/1.0'
})
def make_request(self, method, endpoint, **kwargs):
"""Make API request with error handling"""
url = urljoin(self.base_url, endpoint) if self.base_url else endpoint
try:
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
return response
except requests.exceptions.HTTPError as e:
print(f"HTTP Error: {e}")
print(f"Response: {response.text}")
except requests.exceptions.RequestException as e:
print(f"Request Error: {e}")
return None
def demonstrate_basic_apis():
"""Demonstrate working with various public APIs"""
client = APIClient()
# Example 1: JSONPlaceholder API (Free fake API)
print("=== JSONPlaceholder API ===")
response = client.make_request('GET', 'https://jsonplaceholder.typicode.com/posts')
if response:
posts = response.json()
print(f"Retrieved {len(posts)} posts")
print("First post:")
print(f"Title: {posts[0]['title']}")
print(f"Body: {posts[0]['body'][:100]}...")
# Example 2: Cat Facts API
print("\n=== Cat Facts API ===")
response = client.make_request('GET', 'https://cat-fact.herokuapp.com/facts')
if response:
cat_facts = response.json()
print("Random cat facts:")
for fact in cat_facts[:3]:
print(f"- {fact['text']}")
# Example 3: Weather API (OpenWeatherMap example structure)
print("\n=== Weather API Example ===")
# Note: You need to sign up for an API key at openweathermap.org
# api_key = "your_api_key_here"
# city = "London"
# weather_url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"
# response = client.make_request('GET', weather_url)
# demonstrate_basic_apis()
2. Advanced API Integration with Authentication
python
class GitHubAPI:
def __init__(self, token=None):
self.base_url = "https://api.github.com"
self.session = requests.Session()
self.session.headers.update({
'Accept': 'application/vnd.github.v3+json',
'User-Agent': 'Python-GitHub-Client'
})
if token:
self.session.headers['Authorization'] = f'token {token}'
def get_user_repos(self, username):
"""Get repositories for a user"""
endpoint = f"/users/{username}/repos"
response = self.session.get(urljoin(self.base_url, endpoint))
if response.status_code == 200:
repos = response.json()
return repos
else:
print(f"Error: {response.status_code} - {response.text}")
return None
def get_repo_info(self, owner, repo):
"""Get detailed information about a repository"""
endpoint = f"/repos/{owner}/{repo}"
response = self.session.get(urljoin(self.base_url, endpoint))
if response.status_code == 200:
return response.json()
return None
def search_repositories(self, query, sort='stars', order='desc'):
"""Search for repositories"""
endpoint = "/search/repositories"
params = {
'q': query,
'sort': sort,
'order': order
}
response = self.session.get(urljoin(self.base_url, endpoint), params=params)
if response.status_code == 200:
return response.json()['items']
return None
def demonstrate_github_api():
"""Demonstrate GitHub API usage"""
# Note: For private repos or higher rate limits, use a personal access token
github = GitHubAPI() # Without token - limited to 60 requests/hour
# Get user repositories
print("=== User Repositories ===")
repos = github.get_user_repos("torvalds") # Linus Torvalds
if repos:
print(f"Found {len(repos)} repositories for torvalds")
for repo in repos[:5]:
print(f"- {repo['name']}: {repo['description']}")
# Search repositories
print("\n=== Repository Search ===")
python_repos = github.search_repositories("python", sort="stars", order="desc")
if python_repos:
print("Top Python repositories:")
for repo in python_repos[:5]:
print(f"- {repo['name']} ({repo['stargazers_count']} stars)")
print(f" {repo['description']}")
print(f" URL: {repo['html_url']}")
# demonstrate_github_api()
Real-World Project: Job Scraper
python
class JobScraper:
def __init__(self):
self.scraper = WebScraper(delay=2)
def scrape_indeed_jobs(self, position, location, max_pages=3):
"""Scrape job listings from Indeed (example structure)"""
# Note: This is a simplified example. Real Indeed scraping requires handling their structure
base_url = "https://www.indeed.com/jobs"
all_jobs = []
for page in range(max_pages):
params = {
'q': position,
'l': location,
'start': page * 10
}
response = self.scraper.fetch_page(base_url, params=params)
if not response:
break
soup = BeautifulSoup(response.content, 'html.parser')
# This selector would need to be adjusted for the actual Indeed structure
job_cards = soup.find_all('div', class_='job_seen_beacon')
for card in job_cards:
try:
title_elem = card.find('h2', class_='jobTitle')
company_elem = card.find('span', class_='companyName')
location_elem = card.find('div', class_='companyLocation')
if title_elem and company_elem:
job = {
'title': title_elem.get_text().strip(),
'company': company_elem.get_text().strip(),
'location': location_elem.get_text().strip() if location_elem else 'N/A',
'date_scraped': pd.Timestamp.now().strftime('%Y-%m-%d')
}
all_jobs.append(job)
except Exception as e:
print(f"Error parsing job card: {e}")
continue
print(f"Page {page + 1}: Found {len(job_cards)} jobs")
time.sleep(2) # Be respectful
return all_jobs
def analyze_jobs(self, jobs):
"""Analyze scraped job data"""
if not jobs:
print("No jobs to analyze")
return
df = pd.DataFrame(jobs)
print("\n=== JOB ANALYSIS ===")
print(f"Total jobs found: {len(df)}")
print(f"Unique companies: {df['company'].nunique()}")
# Company frequency
company_counts = df['company'].value_counts()
print("\nTop companies hiring:")
for company, count in company_counts.head(5).items():
print(f" {company}: {count} jobs")
# Location analysis
location_counts = df['location'].value_counts()
print("\nTop locations:")
for location, count in location_counts.head(5).items():
print(f" {location}: {count} jobs")
return df
# Example usage (commented out to avoid actual scraping)
# job_scraper = JobScraper()
# python_jobs = job_scraper.scrape_indeed_jobs("python developer", "New York", max_pages=2)
# if python_jobs:
# df = job_scraper.analyze_jobs(python_jobs)
# df.to_csv('python_jobs.csv', index=False)
API Integration with Web Scraping Fallback
python
class HybridDataCollector:
"""Combine API data with web scraping for comprehensive data collection"""
def __init__(self):
self.api_client = APIClient()
self.scraper = WebScraper()
def get_company_info(self, company_name):
"""Get company information using multiple sources"""
data_sources = {}
# Try LinkedIn (example - would need proper selectors)
linkedin_data = self._scrape_linkedin(company_name)
if linkedin_data:
data_sources['linkedin'] = linkedin_data
# Try Glassdoor (example - would need proper selectors)
glassdoor_data = self._scrape_glassdoor(company_name)
if glassdoor_data:
data_sources['glassdoor'] = glassdoor_data
# Try public APIs
api_data = self._get_company_from_api(company_name)
if api_data:
data_sources['api'] = api_data
return self._merge_company_data(data_sources)
def _scrape_linkedin(self, company_name):
"""Scrape company info from LinkedIn (example structure)"""
# This is a simplified example - real LinkedIn scraping is complex
# and may violate their terms of service
try:
search_url = f"https://www.linkedin.com/company/{company_name.lower().replace(' ', '-')}"
response = self.scraper.fetch_page(search_url)
if response:
soup = BeautifulSoup(response.content, 'html.parser')
# These selectors are hypothetical
name = soup.find('h1')
description = soup.find('meta', attrs={'name': 'description'})
return {
'name': name.get_text().strip() if name else company_name,
'description': description['content'] if description else '',
'source': 'linkedin'
}
except Exception as e:
print(f"LinkedIn scraping failed: {e}")
return None
def _scrape_glassdoor(self, company_name):
"""Scrape company info from Glassdoor (example structure)"""
# Similar structure to LinkedIn example
return None
def _get_company_from_api(self, company_name):
"""Get company info from public APIs"""
# Example using a hypothetical company API
try:
# This would be a real API endpoint
# response = self.api_client.make_request('GET', f'https://api.companyinfo.com/v1/companies?name={company_name}')
# return response.json() if response else None
pass
except Exception as e:
print(f"API call failed: {e}")
return None
def _merge_company_data(self, data_sources):
"""Merge data from multiple sources"""
merged_data = {
'company_name': '',
'description': '',
'employee_count': '',
'industry': '',
'sources': list(data_sources.keys())
}
# Simple merging logic - in reality, this would be more sophisticated
for source, data in data_sources.items():
if data:
for key, value in data.items():
if value and not merged_data.get(key):
merged_data[key] = value
return merged_data
Data Processing & Analysis
python
def process_and_analyze_scraped_data():
"""Process and analyze scraped data"""
# Sample data (in real scenario, this would come from scraping)
sample_jobs = [
{'title': 'Python Developer', 'company': 'Tech Corp', 'location': 'New York', 'salary': '100000'},
{'title': 'Data Scientist', 'company': 'Data Inc', 'location': 'San Francisco', 'salary': '120000'},
{'title': 'Backend Engineer', 'company': 'Web Solutions', 'location': 'New York', 'salary': '110000'},
{'title': 'Machine Learning Engineer', 'company': 'AI Ventures', 'location': 'Boston', 'salary': '130000'},
{'title': 'Python Developer', 'company': 'Startup XYZ', 'location': 'Austin', 'salary': '95000'},
]
df = pd.DataFrame(sample_jobs)
# Data cleaning
df['salary'] = pd.to_numeric(df['salary'], errors='coerce')
df = df.dropna()
# Analysis
print("=== JOB MARKET ANALYSIS ===")
print(f"Total jobs analyzed: {len(df)}")
print(f"Average salary: ${df['salary'].mean():,.2f}")
print(f"Highest salary: ${df['salary'].max():,.2f}")
print(f"Lowest salary: ${df['salary'].min():,.2f}")
# Group by location
location_stats = df.groupby('location')['salary'].agg(['mean', 'count']).round(2)
print("\nSalary by location:")
print(location_stats)
# Group by job title
title_stats = df.groupby('title')['salary'].agg(['mean', 'count']).round(2)
print("\nSalary by job title:")
print(title_stats)
# Visualization (if matplotlib is available)
try:
import matplotlib.pyplot as plt
plt.figure(figsize=(10, 6))
df.groupby('location')['salary'].mean().plot(kind='bar')
plt.title('Average Salary by Location')
plt.ylabel('Salary ($)')
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig('salary_by_location.png')
print("\nChart saved as 'salary_by_location.png'")
except ImportError:
print("\nMatplotlib not available for visualization")
# process_and_analyze_scraped_data()
Ethical Web Scraping & Best Practices
python
class EthicalScraper:
"""Demonstrate ethical web scraping practices"""
def __init__(self, delay=2, max_pages=10):
self.delay = delay
self.max_pages = max_pages
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'ResearchBot/1.0 (+http://example.com/bot)'
})
def check_robots_txt(self, base_url):
"""Check robots.txt before scraping"""
robots_url = urljoin(base_url, '/robots.txt')
response = self.session.get(robots_url)
if response.status_code == 200:
print("robots.txt found. Please review:")
print(response.text[:500]) # Show first 500 characters
else:
print("No robots.txt found or inaccessible")
return response.status_code == 200
def respectful_scraping(self, url, data_extractor):
"""Scrape respectfully with delays and limits"""
data = []
current_page = 1
while current_page <= self.max_pages:
print(f"Scraping page {current_page}...")
response = self.session.get(url)
if response.status_code != 200:
print(f"Failed to fetch page {current_page}")
break
# Extract data using provided function
page_data = data_extractor(response.content)
data.extend(page_data)
# Check if there's a next page
soup = BeautifulSoup(response.content, 'html.parser')
next_link = soup.find('a', text='Next') # Adjust selector as needed
if not next_link:
break
# Get next page URL
next_url = urljoin(url, next_link['href'])
url = next_url
current_page += 1
# Respectful delay
time.sleep(self.delay)
return data
def save_scraping_log(self, url, pages_scraped, data_points):
"""Maintain scraping log for transparency"""
log_entry = {
'timestamp': pd.Timestamp.now().isoformat(),
'url': url,
'pages_scraped': pages_scraped,
'data_points_collected': data_points,
'user_agent': self.session.headers['User-Agent']
}
try:
with open('scraping_log.json', 'a') as f:
f.write(json.dumps(log_entry) + '\n')
except Exception as e:
print(f"Could not save log: {e}")
# Example of ethical scraping
def ethical_scraping_example():
"""Demonstrate ethical scraping practices"""
scraper = EthicalScraper(delay=3, max_pages=2)
# Check robots.txt first
base_url = "http://books.toscrape.com"
scraper.check_robots_txt(base_url)
# Define data extractor function
def extract_books(html_content):
soup = BeautifulSoup(html_content, 'html.parser')
books = soup.find_all('article', class_='product_pod')
book_data = []
for book in books:
title = book.h3.a['title']
price = book.find('p', class_='price_color').get_text()
book_data.append({'title': title, 'price': price})
return book_data
# Scrape respectfully
books_data = scraper.respectful_scraping(
"http://books.toscrape.com/",
extract_books
)
# Log the activity
scraper.save_scraping_log(
"http://books.toscrape.com/",
len(books_data) // 20, # Assuming 20 books per page
len(books_data)
)
print(f"Ethically scraped {len(books_data)} books")
return books_data
# ethical_scraping_example()
API Best Practices:
-
Use API keys responsibly and keep them secure
-
Handle rate limiting with exponential backoff
-
Cache API responses to reduce calls
-
Use proper error handling
-
Respect API terms of service
-
Use pagination for large datasets
This extensive guide provides you with the foundation for web scraping and API integration in Python, along with best practices for ethical and efficient data collection.
Comments (0)
No comments yet. Be the first to comment!