Build a City Status Monitor: Python Web Scraper & Slack Notifier

Before You Begin

Python 3.10+ installed (check with python3 --version)
pipenv or venv for virtual environments
A Slack workspace (free tier is fine) with an incoming webhook URL
A target website that publishes city weather data (we'll use wttr.in for simplicity — no API key needed)

Throughout this tutorial we'll build a City Status Monitor: a script that periodically fetches weather data for cities like toronto, bangalore, and berlin, detects changes in conditions (e.g., rain started, temperature dropped), and pushes a notification to a Slack channel.

1. Project Setup: The Skeleton

Create a project directory and virtual environment:

mkdir city_monitor && cd city_monitor
python3 -m venv .venv && source .venv/bin/activate

Install dependencies:

pip install requests beautifulsoup4 schedule

Create an initial file structure:

city_monitor/
├── scraper.py
├── notifier.py
├── storage.py
└── monitor.py

2. Building the Scraper: Fetching City Weather

We'll use the console-oriented weather service wttr.in. It returns plain text and HTML representations. Beautiful Soup will parse the HTML response. A common pitfall reported on StackOverflow is forgetting to set a User-Agent header; many sites (including wttr.in) block default Python-requests headers.

# scraper.py
from dataclasses import dataclass
from typing import Optional
import requests
from bs4 import BeautifulSoup

@dataclass
class CityWeather:
    city: str
    temperature: str
    condition: str
    humidity: str

def fetch_weather(city: str) -> Optional[CityWeather]:
    """
    Scrapes current weather data from wttr.in for the given city.
    Returns None if the city is invalid or network fails.
    """
    url = f"https://wttr.in/{city}?format=%C|%t|%h"
    headers = {"User-Agent": "curl/7.68.0"}  # mimic common CLI client

    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
    except requests.RequestException as e:
        print(f"Network error for {city}: {e}")
        return None

    # wttr.in returns a pipe-delimited string: condition|temp|humidity%
    parts = response.text.strip().split("|")
    if len(parts) != 3:
        print(f"Unexpected format for {city}: {response.text}")
        return None

    return CityWeather(
        city=city,
        condition=parts[0],
        temperature=parts[1],
        humidity=parts[2]
    )

Why this design? The function is testable in isolation, returns a dedicated data class (idiomatic Python), and handles common failure modes (network timeouts, malformed responses). The engineering blog at Real Python recommends exactly this pattern for web scraping: wrap the request in a try/except and return None to signal failure rather than raising exceptions externally.

3. Storing City Data: JSON-Based Database

We need to keep track of previous weather states to detect changes. A flat JSON file is sufficient for our lightweight service. The official Python documentation for json module advises using json.dump with indent for human readability, but in production you'd want sort_keys=True as well.

# storage.py
import json
from pathlib import Path
from typing import Dict, List
from scraper import CityWeather

DB_PATH = Path("city_data.json")

def load_cities() -> Dict[str, dict]:
    if DB_PATH.exists():
        with open(DB_PATH, "r") as f:
            return json.load(f)
    return {}

def save_cities(records: Dict[str, dict]) -> None:
    with open(DB_PATH, "w") as f:
        json.dump(records, f, indent=2, sort_keys=True)

def check_for_changes(weather: CityWeather) -> List[str]:
    """
    Compares current weather to stored state.
    Returns a list of change descriptions (empty if none).
    """
    records = load_cities()
    changes: List[str] = []

    previous = records.get(weather.city, {})
    if not previous:
        changes.append(f"{weather.city}: New city added to monitoring.")
    else:
        if previous.get("condition") != weather.condition:
            changes.append(f"{weather.city}: Condition changed from {previous['condition']} to {weather.condition}")
        if previous.get("temperature") != weather.temperature:
            changes.append(f"{weather.city}: Temperature changed from {previous['temperature']} to {weather.temperature}")

    # Update the stored record
    records[weather.city] = weather.__dict__
    save_cities(records)
    return changes

4. Sending Notifications via Slack Webhooks

A widely-used pattern in production Python services on GitHub is to treat Slack notifications as an async afterthought: send them in a separate thread or use a queue. For simplicity, we'll send synchronously but wrap it clearly.

# notifier.py
import json
import requests
from typing import List

def send_slack_notification(webhook_url: str, messages: List[str]) -> None:
    """
    Sends a list of change messages to a Slack channel via incoming webhook.
    """
    if not messages:
        return

    payload = {
        "text": "*City Status Monitor Alert* :rotating_light:",
        "attachments": [
            {
                "color": "#FF0000" if "Condition changed" in msg else "#36a64f",
                "text": msg
            }
            for msg in messages
        ]
    }

    try:
        response = requests.post(
            webhook_url,
            data=json.dumps(payload),
            headers={"Content-Type": "application/json"},
            timeout=10
        )
        response.raise_for_status()
    except requests.RequestException as e:
        print(f"Failed to send notification: {e}")

5. Orchestrating the Monitor: Schedule & Loop

The schedule library (used in many automated scraper projects on GitLab) lets us run the check loop every 10 minutes without blocking.

Relevant Resource➜

# monitor.py
from scraper import fetch_weather
from storage import check_for_changes
from notifier import send_slack_notification
import schedule
import time

# Config
WEBHOOK_URL = "https://hooks.slack.com/services/YOUR/TEAM/TOKEN"
CITIES = ["toronto", "bangalore", "berlin", "tokyo"]

def check_all_cities() -> None:
    all_changes: list[str] = []
    for city in CITIES:
        weather = fetch_weather(city)
        if weather:
            changes = check_for_changes(weather)
            all_changes.extend(changes)
        else:
            print(f"Skipping {city} due to fetch error.")

    if all_changes:
        send_slack_notification(WEBHOOK_URL, all_changes)
    else:
        print("No changes detected.")

if __name__ == "__main__":
    # Run immediately on start, then every 10 minutes
    check_all_cities()
    schedule.every(10).minutes.do(check_all_cities)

    print("City Monitor started. Press Ctrl+C to exit.")
    while True:
        schedule.run_pending()
        time.sleep(1)

6. Running and Extending

First, set your Slack webhook URL as an environment variable (never hardcode in production):

export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/..."
python monitor.py

Potential extensions (what a senior dev would think of next):

Add argparse so users can pass city names from the command line.
Replace the JSON file with SQLite for better concurrent access.
Use asyncio and aiohttp to fetch all cities concurrently (official Python docs have a great example of this pattern).

Common Issues & Solutions

Issue 1: requests.exceptions.ConnectionError: Max retries exceeded

Cause: wttr.in rate-limits frequent requests. Fix: increase the interval to 20 minutes or use a proper API with an API key.

Issue 2: JSONDecodeError when reading the database

If the JSON file gets corrupted (e.g., script crashes mid-write), wrap the load in a try/except and fall back to an empty dict. The pathlib library's open can cause issues if the file is empty; check DBPATH.stat().stsize > 0 before reading.

Issue 3: Slack webhook returning 404

Ensure your webhook URL is correct and the channel exists. Test with a simple curl command first: curl -X POST -H 'Content-type: application/json' --data '{"text":"Hello from Python"}' YOURWEBHOOKURL

Issue 4: ImportError: No module named 'schedule'

You forgot to activate your virtual environment: source .venv/bin/activate. Double-check with pip list | grep schedule.

Build a City Status Monitor: Web Scraper & Notification Service in Python