Source code for yadisk.objects.operations

# -*- coding: utf-8 -*-

from functools import partial

from .yadisk_object import YaDiskObject
from .link_object import LinkObject
from ..common import str_or_error, dict_or_error

from typing import Any, Optional

__all__ = [
    "OperationStatusObject", "OperationLinkObject",
    "SyncOperationLinkObject", "AsyncOperationLinkObject"
]

[docs] class OperationStatusObject(YaDiskObject): """ Operation status object. :param operation_status: `dict` or `None` :param yadisk: :any:`YaDisk` or `None`, `YaDisk` object :ivar type: `str`, type of the operation :ivar status: `str`, status of the operation :ivar operation_id: `str`, ID of the operation :ivar link: :any:`LinkObject`, link to the operation :ivar data: `dict`, other information about the operation """ def __init__(self, operation_status: Optional[dict] = None, yadisk: Optional[Any] = None): YaDiskObject.__init__( self, {"type": str_or_error, "status": str_or_error, "operation_id": str_or_error, "link": partial(LinkObject, yadisk=yadisk), "data": dict_or_error}, yadisk) self.import_fields(operation_status)
[docs] class OperationLinkObject(LinkObject): """ Operation link object. :param link: `dict` or `None` :param yadisk: :any:`Client`/:any:`AsyncClient` or `None`, `YaDisk` object :ivar href: `str`, link URL :ivar method: `str`, HTTP method :ivar templated: `bool`, tells whether the URL is templated """ pass
[docs] class SyncOperationLinkObject(OperationLinkObject): """ Operation link object. :param link: `dict` or `None` :param yadisk: :any:`Client` or `None`, `YaDisk` object :ivar href: `str`, link URL :ivar method: `str`, HTTP method :ivar templated: `bool`, tells whether the URL is templated """
[docs] def get_status(self, **kwargs) -> str: """ Get operation status. :param fields: list of keys to be included in the response :param timeout: `float` or `tuple`, request timeout :param headers: `dict` or `None`, additional request headers :param n_retries: `int`, maximum number of retries :param retry_interval: delay between retries in seconds :raises OperationNotFoundError: requested operation was not found :returns: `str`, :code:`"in-progress"` indicates that the operation is currently running, :code:`"success"` indicates that the operation was successful, :code:`"failed"` means that the operation failed """ if self._yadisk is None: raise ValueError("This object is not bound to a YaDisk instance") if self.href is None: raise ValueError("OperationLinkObject has no link") return self._yadisk.get_operation_status(self.href, **kwargs)
[docs] class AsyncOperationLinkObject(OperationLinkObject): """ Operation link object. :param link: `dict` or `None` :param yadisk: :any:`AsyncClient` or `None`, `YaDisk` object :ivar href: `str`, link URL :ivar method: `str`, HTTP method :ivar templated: `bool`, tells whether the URL is templated """
[docs] async def get_status(self, **kwargs) -> str: """ Get operation status. :param fields: list of keys to be included in the response :param timeout: `float` or `tuple`, request timeout :param headers: `dict` or `None`, additional request headers :param n_retries: `int`, maximum number of retries :param retry_interval: delay between retries in seconds :raises OperationNotFoundError: requested operation was not found :returns: `str`, :code:`"in-progress"` indicates that the operation is currently running, :code:`"success"` indicates that the operation was successful, :code:`"failed"` means that the operation failed """ if self._yadisk is None: raise ValueError("This object is not bound to a YaDisk instance") if self.href is None: raise ValueError("OperationLinkObject has no link") return await self._yadisk.get_operation_status(self.href, **kwargs)