101 lines
4.5 KiB
Python
101 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)
|