Загрузить файлы в «api»

api/disk_client.py

@@ -0,0 +1,100 @@
+"""
+Модуль клиента для взаимодействия с 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)