Мини-проект: API погоды - IMI-DS

Привет! В этом проекте мы создадим API для получения погоды. Мы подключимся к внешнему сервису OpenWeatherMap, добавим кэширование, обработку ошибок и создадим удобные эндпоинты. Это отличный проект для портфолио и практики работы с внешними API.

В этом проекте мы создадим:

Интеграцию с OpenWeatherMap
Эндпоинты для текущей погоды и прогноза
Кэширование запросов
Обработку ошибок
Валидацию городов

Содержание

Что нужно знать перед началом
Основная часть
Установка зависимостей
Регистрация в OpenWeatherMap
Структура проекта
Конфигурация (config.py)
Модели данных (models.py)
Сервис погоды (services.py)
Кэширование (cache.py)
Основное приложение (main.py)
Тесты (tests/test_api.py)
Запуск и тестирование
Задачи для доработки
Шпаргалка
Заключение
Что дальше?

Что нужно знать перед началом

Для этого проекта тебе понадобится:

Установленный FastAPI и Uvicorn
API-ключ OpenWeatherMap (бесплатный)
Базовое понимание HTTP-запросов и асинхронности

Совет: OpenWeatherMap предоставляет бесплатный тариф с 60 запросами в минуту.

Основная часть

Установка зависимостей

pip install fastapi uvicorn httpx python-dotenv pytest

requirements.txt:

fastapi==0.104.1
uvicorn==0.24.0
httpx==0.25.1
python-dotenv==1.0.0
pytest==7.4.3

Регистрация в OpenWeatherMap

Перейди на openweathermap.org
Зарегистрируйся (бесплатно)
Получи API-ключ в разделе API Keys
Сохрани ключ в файле .env

.env:

OPENWEATHER_API_KEY=your_api_key_here
OPENWEATHER_BASE_URL=http://api.openweathermap.org/data/2.5

Структура проекта

weather_api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── services.py
│   ├── cache.py
│   └── config.py
├── tests/
│   └── test_api.py
├── .env
└── requirements.txt

Конфигурация (config.py)

import os
from dotenv import load_dotenv

load_dotenv()

class Settings:
    API_KEY: str = os.getenv("OPENWEATHER_API_KEY", "")
    BASE_URL: str = os.getenv("OPENWEATHER_BASE_URL", "http://api.openweathermap.org/data/2.5")
    CACHE_TTL: int = 300  # 5 минут

settings = Settings()

Модели данных (models.py)

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class WeatherMain(BaseModel):
    temp: float
    feels_like: float
    temp_min: float
    temp_max: float
    pressure: int
    humidity: int

class WeatherDescription(BaseModel):
    description: str
    icon: str

class WeatherResponse(BaseModel):
    city: str
    country: str
    main: WeatherMain
    weather: WeatherDescription
    wind_speed: float
    wind_deg: int
    timestamp: datetime
    sunrise: datetime
    sunset: datetime

class ForecastItem(BaseModel):
    dt: datetime
    temp: float
    feels_like: float
    temp_min: float
    temp_max: float
    pressure: int
    humidity: int
    description: str
    wind_speed: float

class ForecastResponse(BaseModel):
    city: str
    country: str
    items: list[ForecastItem]

Сервис погоды (services.py)

import httpx
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from config import settings

class WeatherService:
    def __init__(self):
        self.base_url = settings.BASE_URL
        self.api_key = settings.API_KEY

    async def get_current_weather(self, city: str) -> dict:
        """Получить текущую погоду для города"""
        url = f"{self.base_url}/weather"
        params = {
            "q": city,
            "appid": self.api_key,
            "units": "metric",
            "lang": "ru"
        }

        async with httpx.AsyncClient() as client:
            try:
                response = await client.get(url, params=params)
                response.raise_for_status()
                return response.json()
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 404:
                    raise HTTPException(status_code=404, detail="City not found")
                raise HTTPException(status_code=500, detail="Weather API error")
            except httpx.RequestError:
                raise HTTPException(status_code=500, detail="External API unavailable")

    async def get_forecast(self, city: str, days: int = 5) -> dict:
        """Получить прогноз погоды на указанное количество дней"""
        url = f"{self.base_url}/forecast"
        params = {
            "q": city,
            "appid": self.api_key,
            "units": "metric",
            "lang": "ru",
            "cnt": days * 8  # OpenWeatherMap даёт данные каждые 3 часа
        }

        async with httpx.AsyncClient() as client:
            try:
                response = await client.get(url, params=params)
                response.raise_for_status()
                return response.json()
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 404:
                    raise HTTPException(status_code=404, detail="City not found")
                raise HTTPException(status_code=500, detail="Weather API error")
            except httpx.RequestError:
                raise HTTPException(status_code=500, detail="External API unavailable")

    def parse_current_weather(self, data: dict) -> dict:
        """Парсинг текущей погоды из ответа OpenWeatherMap"""
        return {
            "city": data["name"],
            "country": data["sys"]["country"],
            "main": {
                "temp": data["main"]["temp"],
                "feels_like": data["main"]["feels_like"],
                "temp_min": data["main"]["temp_min"],
                "temp_max": data["main"]["temp_max"],
                "pressure": data["main"]["pressure"],
                "humidity": data["main"]["humidity"]
            },
            "weather": {
                "description": data["weather"][0]["description"],
                "icon": data["weather"][0]["icon"]
            },
            "wind_speed": data["wind"]["speed"],
            "wind_deg": data["wind"]["deg"],
            "timestamp": datetime.fromtimestamp(data["dt"]),
            "sunrise": datetime.fromtimestamp(data["sys"]["sunrise"]),
            "sunset": datetime.fromtimestamp(data["sys"]["sunset"])
        }

    def parse_forecast(self, data: dict) -> dict:
        """Парсинг прогноза из ответа OpenWeatherMap"""
        items = []
        for item in data["list"]:
            items.append({
                "dt": datetime.fromtimestamp(item["dt"]),
                "temp": item["main"]["temp"],
                "feels_like": item["main"]["feels_like"],
                "temp_min": item["main"]["temp_min"],
                "temp_max": item["main"]["temp_max"],
                "pressure": item["main"]["pressure"],
                "humidity": item["main"]["humidity"],
                "description": item["weather"][0]["description"],
                "wind_speed": item["wind"]["speed"]
            })

        return {
            "city": data["city"]["name"],
            "country": data["city"]["country"],
            "items": items
        }

weather_service = WeatherService()

Кэширование (cache.py)

from functools import lru_cache
from datetime import datetime, timedelta
from typing import Optional
import json

class Cache:
    def __init__(self):
        self._cache = {}

    def get(self, key: str) -> Optional[dict]:
        """Получить данные из кэша"""
        if key in self._cache:
            data, timestamp = self._cache[key]
            if datetime.now() - timestamp < timedelta(seconds=300):  # 5 минут
                return data
            else:
                del self._cache[key]
        return None

    def set(self, key: str, value: dict):
        """Сохранить данные в кэш"""
        self._cache[key] = (value, datetime.now())

    def clear(self):
        """Очистить кэш"""
        self._cache.clear()

cache = Cache()

Основное приложение (main.py)

from fastapi import FastAPI, HTTPException, Query
from typing import Optional
from models import WeatherResponse, ForecastResponse
from services import weather_service
from cache import cache

app = FastAPI(
    title="Weather API",
    description="API для получения погоды",
    version="1.0.0"
)

@app.get("/")
def root():
    return {
        "message": "Weather API",
        "endpoints": {
            "/weather/{city}": "Текущая погода",
            "/forecast/{city}": "Прогноз на 5 дней",
            "/weather/search": "Поиск погоды с параметрами"
        }
    }

@app.get("/weather/{city}", response_model=WeatherResponse)
async def get_weather(city: str, units: Optional[str] = Query("metric", regex="^(metric|imperial)$")):
    """
    Получить текущую погоду для города.
    """
    cache_key = f"weather_{city}_{units}"
    cached = cache.get(cache_key)

    if cached:
        return cached

    data = await weather_service.get_current_weather(city)
    parsed = weather_service.parse_current_weather(data)

    cache.set(cache_key, parsed)
    return parsed

@app.get("/forecast/{city}", response_model=ForecastResponse)
async def get_forecast(city: str, days: int = Query(5, ge=1, le=5)):
    """
    Получить прогноз погоды на указанное количество дней (от 1 до 5).
    """
    cache_key = f"forecast_{city}_{days}"
    cached = cache.get(cache_key)

    if cached:
        return cached

    data = await weather_service.get_forecast(city, days)
    parsed = weather_service.parse_forecast(data)

    cache.set(cache_key, parsed)
    return parsed

@app.get("/weather/search")
async def search_weather(
    q: str = Query(..., description="Название города"),
    units: Optional[str] = Query("metric", regex="^(metric|imperial)$")
):
    """
    Поиск погоды по городу.
    """
    return await get_weather(q, units)

@app.delete("/cache")
async def clear_cache():
    """Очистить кэш"""
    cache.clear()
    return {"message": "Cache cleared"}

Тесты (tests/test_api.py)

import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.cache import cache

client = TestClient(app)

class TestWeatherAPI:
    def setup_method(self):
        cache.clear()

    def test_root(self):
        response = client.get("/")
        assert response.status_code == 200
        assert "endpoints" in response.json()

    def test_get_weather(self):
        response = client.get("/weather/London")
        assert response.status_code in [200, 404, 500]
        if response.status_code == 200:
            data = response.json()
            assert "city" in data
            assert "main" in data
            assert "weather" in data

    def test_get_forecast(self):
        response = client.get("/forecast/London?days=3")
        assert response.status_code in [200, 404, 500]
        if response.status_code == 200:
            data = response.json()
            assert "city" in data
            assert "items" in data

    def test_clear_cache(self):
        response = client.delete("/cache")
        assert response.status_code == 200
        assert response.json()["message"] == "Cache cleared"

Запуск и тестирование

Запуск:

uvicorn app.main:app --reload

Документация:

Swagger: http://localhost:8000/docs

Примеры запросов:

# Текущая погода
GET /weather/Moscow
GET /weather/London?units=imperial

# Прогноз на 5 дней
GET /forecast/Moscow
GET /forecast/London?days=3

# Поиск
GET /weather/search?q=Paris

# Очистка кэша
DELETE /cache

Задачи для доработки

Задача 1. Добавь поддержку нескольких языков (помимо русского).

Задача 2. Добавь эндпоинт /weather/compare?city1=Moscow&city2=London.

Задача 3. Добавь в кэш TTL (Time To Live) для каждой записи.

Задача 4. Добавь лимит запросов (rate limiting) для защиты API.

Задача 5. Добавь историю запросов (последние 10 городов).

Ответы:

Задача 1.

# Добавь параметр lang в запрос к OpenWeatherMap
params = {"lang": "ru"}

Задача 2.

@app.get("/weather/compare")
async def compare_weather(city1: str, city2: str):
    weather1 = await get_weather(city1)
    weather2 = await get_weather(city2)
    return {"city1": weather1, "city2": weather2}

Задача 3.

# Используй timedelta для TTL в cache.py

Задача 4.

# Добавь SlowAPI для rate limiting

Задача 5.

# Храни список городов в памяти

Шпаргалка

Эндпоинт	Метод	Назначение
`/weather/{city}`	GET	Текущая погода
`/forecast/{city}`	GET	Прогноз на 5 дней
`/weather/search?q={city}`	GET	Поиск погоды
`/cache`	DELETE	Очистка кэша