Python и Яндекс.Диск: полное руководство по интеграции
Основные способы интеграции с Яндекс.Диском
Как максимально эффективно работать с Яндекс.Диском через Python?
Наиболее удобный способ - использовать официальную библиотеку yadisk. Она абстрагирует детали HTTP запросов и предоставляет интуитивно понятные методы.
Установка и подготовка
pip install yadiskPython яндекс диск (работа с яндекс диском из python)
Для аутентификации требуется OAuth токен. Процесс получения описан в разделе вариантов.
Инициализация и проверка соединения
import yadisk
token = 'AgAAAA...' # Ваш токен
disk = yadisk.YaDisk(token)
print(disk.check_token())
True
После успешной проверки можно выполнять операции.
Примеры операций
Создание папки
disk.mkdir('disk:/example')
print('Папка создана')
Загрузка файла
disk.upload('C:/photo.jpg', 'disk:/example/photo.jpg')
Скачивание файла
disk.download('disk:/example/photo.jpg', 'C:/downloaded_photo.jpg')
Получение списка файлов
for item in disk.listdir('disk:/'):
print(item.name, item.type)
example dir photo.jpg file ...
Удаление файла или папки
disk.remove('disk:/example/photo.jpg')
Типичные проблемы и их решения
- Неверный токен: при вызове методов возникает исключение
yadisk.exceptions.AuthError. Проверьте срок действия токена и правильность его копирования. - Превышение квоты: исключение
yadisk.exceptions.TooManyRequestsError. Необходимо добавить задержки между запросами или уменьшить их частоту. - Сетевые ошибки: исключения
requests.exceptions.ConnectionError. Проверьте интернет-соединение и доступность серверов Яндекса. - Неверный путь: если папка не существует, метод
listdirвернёт пустой список или вызоветPathNotFoundError. Предварительно проверяйте существование черезdisk.exists(path).
Как работать с Яндекс.Диском без сторонних библиотек, используя только requests?
Этот вариант полезен, когда нужно минимизировать зависимости или требуется полный контроль над запросами. Используется REST API Яндекс.Диска напрямую.
Получение информации о диске
import requests
headers = {'Authorization': 'OAuth токен'}
response = requests.get('https://cloud-api.yandex.net/v1/disk', headers=headers)
print(response.json())
{'trash_size': 0, 'total_space': 107374182400, 'used_space': 2048, ...}
Создание папки
params = {'path': 'disk:/new_folder'}
response = requests.put('https://cloud-api.yandex.net/v1/disk/resources', headers=headers, params=params)
print(response.status_code)
201
Загрузка файла (два шага: получить URL для загрузки, затем загрузить)
# Получение ссылки для загрузки
params = {'path': 'disk:/new_folder/file.txt'}
response = requests.get('https://cloud-api.yandex.net/v1/disk/resources/upload', headers=headers, params=params)
url = response.json()['href']
# Загрузка файла
with open('local.txt', 'rb') as f:
upload_response = requests.put(url, files={'file': f})
print(upload_response.status_code)
201
Удаление файла
params = {'path': 'disk:/new_folder/file.txt'}
response = requests.delete('https://cloud-api.yandex.net/v1/disk/resources', headers=headers, params=params)
print(response.status_code)
204
Проблемы при работе с requests
- Ошибка авторизации: неверный формат заголовка
Authorization. Должно бытьOAuth ваш_токен(обязательно пробел после OAuth). - Загрузка больших файлов: может потребоваться разбивка на части или указание заголовка
Content-Length. Для простых случаев достаточноfiles. - Ограничение на количество запросов: API имеет лимиты (например, 20 запросов в секунду). При превышении возвращается 429 Too Many Requests. Следует добавить повторную попытку с ожиданием.
Как использовать протокол WebDAV для взаимодействия с Яндекс.Диском?
WebDAV предоставляет единый интерфейс для работы с файловыми хранилищами. Яндекс.Диск поддерживает WebDAV. Это удобно для интеграции с приложениями, где уже используется WebDAV.
Установка библиотеки
pip install webdavclient3
Подключение к Яндекс.Диску
from webdav3.client import Client
options = {
'webdav_hostname': 'https://webdav.yandex.ru',
'webdav_login': 'логин',
'webdav_password': 'пароль'
}
client = Client(options)
Внимание: использование пароля менее безопасно, чем OAuth. Рекомендуется применять только для тестов.
Примеры операций
# Список файлов корневого каталога
files = client.list()
print(files)
# Скачивание файла
client.download('remote/path/file.txt', 'local_file.txt')
# Загрузка
client.upload('local_file.txt', 'remote/path/file.txt')
# Создание папки
client.mkdir('new_folder')
Проблемы и ограничения WebDAV
- Не поддерживаются все методы API: например, невозможно получить информацию о диске (занятое место).
- Аутентификация: требуется пароль аккаунта Яндекса, что небезопасно. Лучше создать пароль приложения (настройки безопасности Яндекса).
- Кодировка путей: могут возникать проблемы с кириллическими названиями. Следует использовать URL-кодирование.
Как получить OAuth токен для доступа к API Яндекс.Диска?
Без токена невозможно использование REST API или библиотеки yadisk. Процесс состоит из регистрации приложения, получения кода авторизации и обмена его на токен.
Шаг 1: Регистрация приложения
Перейдите на https://oauth.yandex.ru/client/new, заполните поля (название, описание). В разделе Платформа выберите Веб-сервисы. Укажите Callback URL (например, https://example.com для тестов). После сохранения вы получите ID приложения и Пароль приложения.
Шаг 2: Получение кода авторизации
Сформируйте URL:
https://oauth.yandex.ru/authorize?response_type=code&client_id=ВАШ_ID&redirect_uri=ВАШ_REDIRECT_URI
Перейдите по нему, подтвердите доступ. Браузер перенаправит на redirect_uri с параметром code.
Шаг 3: Обмен кода на токен
import requests
data = {
'grant_type': 'authorization_code',
'code': 'полученный_код',
'client_id': 'ваш_id',
'client_secret': 'ваш_пароль'
}
response = requests.post('https://oauth.yandex.ru/token', data=data)
token = response.json().get('access_token')
print(token)
AgAAAA... (токен)
Токен действителен ограниченное время (обычно 1 год). Его можно сохранить и использовать повторно.
Распространённые ошибки
- Несовпадение redirect_uri: указанный в приложении и в запросе должны совпадать точно.
- Просроченный код: код авторизации действителен несколько минут. Если не обменять вовремя, нужно запрашивать новый.
- Недостаточно прав: при регистрации приложения нужно указать необходимые разрешения (например,
cloud_api:disk.readиcloud_api:disk.write).
Расширенные примеры работы с Яндекс.Диском
import yadisk, os
token = 'токен'
disk = yadisk.YaDisk(token)
# Рекурсивный обход всех папок и файлов
def list_all(disk, path='disk:/'):
for item in disk.listdir(path):
if item.type == 'dir':
list_all(disk, path + '/' + item.name)
else:
print(item.name)
list_all(disk)
file1.txt subfile.pdf ...
# Загрузка всей папки с локального диска
def upload_folder(local_path, remote_path, disk):
for root, dirs, files in os.walk(local_path):
for dir in dirs:
disk.mkdir(remote_path + '/' + dir)
for file in files:
local_file = os.path.join(root, file)
remote_file = remote_path + '/' + file
disk.upload(local_file, remote_file)
upload_folder('C:/my_files', 'disk:/backup', disk)
# Получение метаданных файла
info = disk.get_meta('disk:/photo.jpg')
print(info.name, info.size, info.created)
photo.jpg 102400 2023-05-10T12:34:56+00:00
# Работа с корзиной (список и восстановление)
trash_items = disk.trash_list()
for item in trash_items:
print(item.name, item.origin_path)
# Восстановить
disk.trash_restore('disk:/trash/photo.jpg')
# Использование повторных попыток с библиотекой requests
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
session.mount('http://', adapter)
headers = {'Authorization': 'OAuth токен'}
params = {'path': 'disk:/'}
response = session.get('https://cloud-api.yandex.net/v1/disk/resources', headers=headers, params=params)
print(response.json())
# Асинхронная загрузка (используя asyncio и aiohttp) - необязательно
# Покажем простую синхронную загрузку с прогресс-баром (tqdm)
import yadisk, tqdm
disk = yadisk.YaDisk(token)
with tqdm.tqdm(unit='B', unit_scale=True, desc='Загрузка') as pbar:
disk.upload('bigfile.iso', 'disk:/bigfile.iso', progress_callback=lambda sent, total: pbar.update(sent - pbar.n))