Реализация REST API для управления ботами-парсерами

Реализация REST API для управления ботами-парсерами

Когда парсеров несколько и ими нужно управлять программно — запускать, останавливать, менять конфигурацию, получать статус — создаётся REST API. Это позволяет интегрировать управление парсерами в любые внешние системы: внутренний дашборд, CI/CD, сторонние сервисы.

Структура API

POST   /api/v1/scrapers              — создать новый скрапер
GET    /api/v1/scrapers              — список скраперов
GET    /api/v1/scrapers/{id}         — конфигурация скрапера
PATCH  /api/v1/scrapers/{id}         — обновить конфигурацию
DELETE /api/v1/scrapers/{id}         — удалить скрапер

POST   /api/v1/scrapers/{id}/run     — запустить немедленно
POST   /api/v1/scrapers/{id}/stop    — остановить запущенный
GET    /api/v1/scrapers/{id}/status  — текущий статус

GET    /api/v1/scrapers/{id}/runs    — история запусков
GET    /api/v1/scrapers/{id}/runs/{runId} — детали запуска

GET    /api/v1/scrapers/{id}/results — результаты парсинга

Реализация на FastAPI

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class ScraperConfig(BaseModel):
    name:        str
    url:         str
    schedule:    Optional[str] = None  # cron expression
    proxy_pool:  Optional[str] = None
    rate_limit:  int = 5  # req/sec
    headers:     dict = {}

@app.post('/api/v1/scrapers', status_code=201)
async def create_scraper(config: ScraperConfig):
    scraper = await ScraperRepository.create(config.dict())
    if config.schedule:
        await Scheduler.register(scraper.id, config.schedule)
    return scraper

@app.post('/api/v1/scrapers/{scraper_id}/run')
async def run_scraper(scraper_id: int, background_tasks: BackgroundTasks):
    scraper = await ScraperRepository.get_or_404(scraper_id)

    if scraper.status == 'running':
        raise HTTPException(409, 'Scraper is already running')

    run = await ScraperRun.create(scraper_id=scraper_id, status='pending')
    background_tasks.add_task(execute_scraper, scraper, run.id)

    return {'run_id': run.id, 'status': 'started'}

@app.get('/api/v1/scrapers/{scraper_id}/status')
async def get_status(scraper_id: int):
    scraper  = await ScraperRepository.get_or_404(scraper_id)
    last_run = await ScraperRun.get_latest(scraper_id)

    return {
        'id':          scraper_id,
        'status':      last_run.status if last_run else 'idle',
        'last_run':    last_run.started_at if last_run else None,
        'items_count': last_run.items_collected if last_run else 0,
    }

Аутентификация

API-ключи с уровнями доступа: read, write, admin. Ключи хранятся как хеши (bcrypt), передаются в заголовке Authorization: Bearer {key}.

Webhooks

Подписка на события: завершение запуска, ошибки, новые данные:

@app.post('/api/v1/webhooks')
async def create_webhook(url: str, events: list[str]):
    # events: ['run.completed', 'run.failed', 'data.new']
    return await WebhookRepository.create(url=url, events=events)

Сроки

REST API для управления парсерами с аутентификацией и webhooks: 5–8 рабочих дней.