Реалізація 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 робочих днів.