100 lines
4.5 KiB
Python
100 lines
4.5 KiB
Python
|
"""
|
|||
|
|
Модуль клиента для взаимодействия с REST API Яндекс Диска.
|
|||
|
|
Документация API: https://yandex.ru/dev/disk/api/concepts/about.html
|
|||
|
|
Полигон: https://cloud-api.yandex.net
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import requests
|
|||
|
|
from typing import Optional, Dict, Any
|
|||
|
|
|
|||
|
|
|
|||
|
|
class YandexDiskClient:
|
|||
|
|
"""
|
|||
|
|
HTTP-клиент-обёртка для выполнения запросов к API Яндекс Диска.
|
|||
|
|
Инкапсулирует базовые методы (GET, POST, PUT, DELETE) и авторизацию.
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
BASE_URL = "https://cloud-api.yandex.net/v1/disk/resources"
|
|||
|
|
|
|||
|
|
def __init__(self, token: str, timeout: int = 10):
|
|||
|
|
"""
|
|||
|
|
Инициализация сессии с предустановленными заголовками авторизации.
|
|||
|
|
|
|||
|
|
:param token: OAuth-токен сервиса Яндекс
|
|||
|
|
:param timeout: Дефолтный таймаут для HTTP-запросов в секундах
|
|||
|
|
"""
|
|||
|
|
self.timeout = timeout
|
|||
|
|
self.session = requests.Session()
|
|||
|
|
# Все запросы к API Яндекс Диска требуют заголовок Authorization с префиксом OAuth
|
|||
|
|
self.session.headers.update({
|
|||
|
|
"Authorization": f"OAuth {token}",
|
|||
|
|
"Accept": "application/json",
|
|||
|
|
"Content-Type": "application/json"
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def create_folder(self, path: str) -> requests.Response:
|
|||
|
|
"""
|
|||
|
|
PUT /v1/disk/resources
|
|||
|
|
Создание папки на Яндекс Диске.
|
|||
|
|
|
|||
|
|
:param path: Путь к создаваемой папке (например, 'test_dir' или 'disk:/test_dir')
|
|||
|
|
:return: Объект ответа requests.Response (201 Created при успехе)
|
|||
|
|
"""
|
|||
|
|
url = self.BASE_URL
|
|||
|
|
params = {"path": path}
|
|||
|
|
return self.session.put(url, params=params, timeout=self.timeout)
|
|||
|
|
|
|||
|
|
def get_resource_info(self, path: str, fields: Optional[str] = None) -> requests.Response:
|
|||
|
|
"""
|
|||
|
|
GET /v1/disk/resources
|
|||
|
|
Получение метаинформации о ресурсе (файле или папке).
|
|||
|
|
|
|||
|
|
:param path: Путь к ресурсу на Диске
|
|||
|
|
:param fields: Список возвращаемых полей через запятую (для фильтрации ответа)
|
|||
|
|
:return: Объект ответа requests.Response (200 OK при успехе)
|
|||
|
|
"""
|
|||
|
|
url = self.BASE_URL
|
|||
|
|
params = {"path": path}
|
|||
|
|
if fields:
|
|||
|
|
params["fields"] = fields
|
|||
|
|
return self.session.get(url, params=params, timeout=self.timeout)
|
|||
|
|
|
|||
|
|
def publish_resource(self, path: str) -> requests.Response:
|
|||
|
|
"""
|
|||
|
|
POST /v1/disk/resources/publish
|
|||
|
|
Публикация ресурса (создание публичной ссылки).
|
|||
|
|
|
|||
|
|
:param path: Путь к ресурсу, который необходимо опубликовать
|
|||
|
|
:return: Объект ответа requests.Response (200 OK при успехе)
|
|||
|
|
"""
|
|||
|
|
url = f"{self.BASE_URL}/publish"
|
|||
|
|
params = {"path": path}
|
|||
|
|
return self.session.post(url, params=params, timeout=self.timeout)
|
|||
|
|
|
|||
|
|
def unpublish_resource(self, path: str) -> requests.Response:
|
|||
|
|
"""
|
|||
|
|
PUT /v1/disk/resources/unpublish
|
|||
|
|
Закрытие публичного доступа к ресурсу.
|
|||
|
|
|
|||
|
|
:param path: Путь к ресурсу, доступ к которому закрывается
|
|||
|
|
:return: Объект ответа requests.Response (200 OK при успехе)
|
|||
|
|
"""
|
|||
|
|
url = f"{self.BASE_URL}/unpublish"
|
|||
|
|
params = {"path": path}
|
|||
|
|
return self.session.put(url, params=params, timeout=self.timeout)
|
|||
|
|
|
|||
|
|
def delete_resource(self, path: str, permanently: bool = True) -> requests.Response:
|
|||
|
|
"""
|
|||
|
|
DELETE /v1/disk/resources
|
|||
|
|
Удаление файла или папки.
|
|||
|
|
|
|||
|
|
:param path: Путь к удаляемому ресурсу
|
|||
|
|
:param permanently: True — удалить безвозвратно мимо Корзины; False — переместить в Корзину
|
|||
|
|
:return: Объект ответа requests.Response (204 No Content при успешном синхронном удалении)
|
|||
|
|
"""
|
|||
|
|
url = self.BASE_URL
|
|||
|
|
params = {
|
|||
|
|
"path": path,
|
|||
|
|
"permanently": str(permanently).lower()
|
|||
|
|
}
|
|||
|
|
return self.session.delete(url, params=params, timeout=self.timeout)
|