Как настроить автоматическую проверку доступности внешних URL из базы данных с помощью Python и библиотеки requests?

Автоматическая проверка доступности внешних URL из базы данных — типичная задача для мониторинга ссылок, SEO-аудита или контроля работоспособности партнёрских ресурсов. Рассмотрим полный подход с использованием Python и библиотеки requests.

## 1. Подготовка окружения

Установите необходимые библиотеки:

pip install requests sqlalchemy psycopg2-binary

Для работы с SQLite дополнительных зависимостей не нужно — она встроена в Python.

## 2. Подключение к базе данных и получение URL

Пример с SQLite через стандартный модуль sqlite3:
python
import sqlite3

def get_urls_from_db(db_path):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute(‘SELECT id, url FROM links WHERE is_active = 1’)
rows = cursor.fetchall()
conn.close()
return rows

## 3. Функция проверки доступности URL

python
import requests

def check_url(url, timeout=10):
try:
response = requests.head(url, timeout=timeout, allow_redirects=True)
return response.status_code, None
except requests.exceptions.ConnectionError:
return None, ‘ConnectionError’
except requests.exceptions.Timeout:
return None, ‘Timeout’
except requests.exceptions.RequestException as e:
return None, str(e)

Использование метода HEAD вместо GET снижает нагрузку на сервер и ускоряет проверку.

## 4. Асинхронная или параллельная проверка

Для большого количества URL используйте ThreadPoolExecutor:
python
from concurrent.futures import ThreadPoolExecutor, as_completed

def check_all_urls(urls, max_workers=10):
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_url = {executor.submit(check_url, url): (uid, url) for uid, url in urls}
for future in as_completed(future_to_url):
uid, url = future_to_url[future]
status_code, error = future.result()
results.append((uid, url, status_code, error))
return results

## 5. Сохранение результатов обратно в БД

python
from datetime import datetime

def save_results(db_path, results):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
for uid, url, status_code, error in results:
cursor.execute(
‘UPDATE links SET last_status=?, last_error=?, checked_at=? WHERE id=?’,
(status_code, error, datetime.now().isoformat(), uid)
)
conn.commit()
conn.close()

## 6. Автоматизация запуска

Для регулярного запуска используйте:
— **cron** (Linux/macOS): добавьте задачу через `crontab -e`
— **Task Scheduler** (Windows)
— **APScheduler** — библиотека для планировщика внутри Python-приложения

Пример cron-задачи каждые 6 часов:

0 */6 * * * /usr/bin/python3 /path/to/checker.py

## 7. Логирование и уведомления

Добавьте модуль logging для фиксации ошибок и настройте отправку уведомлений через smtplib или Telegram Bot API при обнаружении недоступных URL.

Такой подход обеспечивает надёжный, масштабируемый и легко поддерживаемый мониторинг внешних ссылок.