# -*- coding: utf-8 -*-
from functools import partial
from pathlib import PurePosixPath
from urllib.parse import urlencode, urlparse, parse_qs
from .yadisk_object import YaDiskObject
from .disk import UserPublicInfoObject
from ..common import (
typed_list, yandex_date, is_resource_link, is_public_resource_link,
ensure_path_has_schema, str_or_error, int_or_error, bool_or_error,
dict_or_error, str_or_dict_or_error, FileOrPath, FileOrPathDestination)
from typing import overload, Union, Protocol, Optional, TYPE_CHECKING
from ..compat import AsyncGenerator, List
if TYPE_CHECKING:
from ..yadisk import YaDisk
import datetime
__all__ = ["CommentIDsObject", "EXIFObject", "FilesResourceListObject",
"LastUploadedResourceListObject", "LinkObject", "OperationLinkObject",
"PublicResourcesListObject", "ResourceListObject", "ResourceObject",
"ResourceUploadLinkObject", "ShareInfoObject", "PublicResourceObject",
"PublicResourceListObject", "TrashResourceObject", "TrashResourceListObject",
"ResourceLinkObject", "PublicResourceLinkObject", "ResourceDownloadLinkObject"]
[документация]
class EXIFObject(YaDiskObject):
"""
EXIF metadata object.
:param exif: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar date_time: :any:`datetime.datetime`, capture date
"""
date_time: Optional["datetime.datetime"]
def __init__(self,
exif: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
YaDiskObject.__init__(self, {"date_time": yandex_date}, yadisk)
self.import_fields(exif)
[документация]
class FilesResourceListObject(YaDiskObject):
"""
Flat list of files.
:param files_resource_list: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar items: `list`, flat list of files (:any:`ResourceObject`)
:ivar limit: `int`, maximum number of elements in the list
:ivar offset: `int`, offset from the beginning of the list
"""
items: Optional[List["ResourceObject"]]
limit: Optional[int]
offset: Optional[int]
def __init__(self,
files_resource_list: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
YaDiskObject.__init__(
self,
{"items": typed_list(partial(ResourceObject, yadisk=yadisk)),
"limit": int_or_error,
"offset": int_or_error},
yadisk)
self.import_fields(files_resource_list)
[документация]
class LastUploadedResourceListObject(YaDiskObject):
"""
List of last uploaded resources.
:param last_uploaded_resources_list: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar items: `list`, list of resources (:any:`ResourceObject`)
:ivar limit: `int`, maximum number of elements in the list
"""
items: Optional[List["ResourceObject"]]
limit: Optional[int]
def __init__(self,
last_uploaded_resources_list: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
YaDiskObject.__init__(
self,
{"items": typed_list(partial(ResourceObject, yadisk=yadisk)),
"limit": int_or_error},
yadisk)
self.import_fields(last_uploaded_resources_list)
[документация]
class LinkObject(YaDiskObject):
"""
Link object.
:param link: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar href: `str`, link URL
:ivar method: `str`, HTTP method
:ivar templated: `bool`, tells whether the URL is templated
"""
href: Optional[str]
method: Optional[str]
templated: Optional[bool]
def __init__(self,
link: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
YaDiskObject.__init__(
self,
{"href": str_or_error,
"method": str_or_error,
"templated": bool_or_error},
yadisk)
self.import_fields(link)
[документация]
class OperationLinkObject(LinkObject):
"""
Operation link object.
:param link: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar href: `str`, link URL
:ivar method: `str`, HTTP method
:ivar templated: `bool`, tells whether the URL is templated
"""
[документация]
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 :any:`aiohttp.ClientTimeout`, 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`
"""
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)
[документация]
class PublicResourcesListObject(YaDiskObject):
"""
List of public resources.
:param public_resources_list: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar items: `list`, list of public resources (:any:`PublicResourceObject`)
:ivar type: `str`, resource type to filter by
:ivar limit: `int`, maximum number of elements in the list
:ivar offset: `int`, offset from the beginning of the list
"""
items: Optional[List["PublicResourceObject"]]
type: Optional[str]
limit: Optional[int]
offset: Optional[int]
def __init__(self,
public_resources_list: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
YaDiskObject.__init__(
self,
{"items": typed_list(partial(PublicResourceObject, yadisk=yadisk)),
"type": str_or_error,
"limit": int_or_error,
"offset": int_or_error},
yadisk)
self.import_fields(public_resources_list)
class ResourceProtocol(Protocol):
@property
def type(self) -> Optional[str]: ...
@property
def path(self) -> Optional[str]: ...
@property
def public_key(self) -> Optional[str]: ...
@property
def public_url(self) -> Optional[str]: ...
@property
def file(self) -> Optional[str]: ...
@property
def _yadisk(self) -> Optional["YaDisk"]: ...
class ResourceObjectMethodsMixin:
async def get_meta(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> "ResourceObject":
"""
Get meta information about a file/directory.
:param relative_path: `str` or `None`, relative path from resource
:param limit: number of children resources to be included in the response
:param offset: number of children resources to be skipped in the response
:param preview_size: size of the file preview
:param preview_crop: `bool`, cut the preview to the size specified in the `preview_size`
:param sort: `str`, field to be used as a key to sort children resources
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:returns: :any:`ResourceObject`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.get_meta(str(path), **kwargs)
async def get_public_meta(self: ResourceProtocol, **kwargs) -> "PublicResourceObject":
"""
Get meta-information about a public resource.
:param path: relative path to a resource in a public folder.
:param offset: offset from the beginning of the list of nested resources
:param limit: maximum number of nested elements to be included in the list
:param sort: `str`, field to be used as a key to sort children resources
:param preview_size: file preview size
:param preview_crop: `bool`, allow preview crop
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:returns: :any:`PublicResourceObject`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
public_key_or_url = self.public_key or self.public_url
if public_key_or_url is None:
raise ValueError("ResourceObject doesn't have a public_key/public_url")
return await self._yadisk.get_public_meta(public_key_or_url, **kwargs)
async def exists(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> bool:
"""
Check whether resource exists.
:param relative_path: `str` or `None`, relative path from the resource
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ForbiddenError: application doesn't have enough rights for this request
:returns: `bool`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.exists(str(path), **kwargs)
async def get_type(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> str:
"""
Get resource type.
:param relative_path: relative path from the resource
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:returns: "file" or "dir"
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.get_type(str(path), **kwargs)
async def is_dir(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> bool:
"""
Check whether resource is a directory.
:param relative_path: relative path from the resource
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ForbiddenError: application doesn't have enough rights for this request
:returns: `True` if `path` is a directory, `False` otherwise (even if it doesn't exist)
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.is_dir(str(path), **kwargs)
async def is_file(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> bool:
"""
Check whether resource is a file.
:param relative_path: relative path from the resource
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ForbiddenError: application doesn't have enough rights for this request
:returns: `True` if `path` is a file, `False` otherwise (even if it doesn't exist)
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.is_file(str(path), **kwargs)
async def listdir(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> AsyncGenerator["ResourceObject", None]:
"""
Get contents of the resource.
:param relative_path: relative path from resource
:param limit: number of children resources to be included in the response
:param offset: number of children resources to be skipped in the response
:param preview_size: size of the file preview
:param preview_crop: `bool`, cut the preview to the size specified in the `preview_size`
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises WrongResourceTypeError: resource is not a directory
:returns: generator of :any:`ResourceObject`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.listdir(str(path), **kwargs)
async def public_listdir(self: ResourceProtocol, **kwargs) -> AsyncGenerator["PublicResourceObject", None]:
"""
Get contents of a public directory.
:param path: relative path to the resource in the public folder.
:param limit: number of children resources to be included in the response
:param offset: number of children resources to be skipped in the response
:param preview_size: size of the file preview
:param preview_crop: `bool`, cut the preview to the size specified in the `preview_size`
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises WrongResourceTypeError: resource is not a directory
:returns: generator of :any:`PublicResourceObject`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
public_key_or_url = self.public_key or self.public_url
if public_key_or_url is None:
raise ValueError("ResourceObject doesn't have a public_key/public_url")
return await self._yadisk.public_listdir(public_key_or_url, **kwargs)
async def get_upload_link(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> str:
"""
Get a link to upload the file using the PUT request.
:param relative_path: `str` or `None`, relative path to the resource
:param overwrite: `bool`, determines whether to overwrite the destination
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ParentNotFoundError: parent directory doesn't exist
:raises PathExistsError: destination path already exists
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:raises InsufficientStorageError: cannot upload file due to lack of storage space
:raises UploadTrafficLimitExceededError: upload limit has been exceeded
:returns: `str`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.get_upload_link(str(path), **kwargs)
async def upload(self: ResourceProtocol,
path_or_file: FileOrPath,
relative_path: Optional[str] = None, /, **kwargs) -> "ResourceLinkObject":
"""
Upload a file to disk.
:param path_or_file: path, file-like object or an async generator function to be uploaded
:param relative_path: `str` or `None`, destination path relative to the resource
:param overwrite: if `True`, the resource will be overwritten if it already exists,
an error will be raised otherwise
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ParentNotFoundError: parent directory doesn't exist
:raises PathExistsError: destination path already exists
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:raises InsufficientStorageError: cannot upload file due to lack of storage space
:raises UploadTrafficLimitExceededError: upload limit has been exceeded
:returns: :any:`ResourceLinkObject`, link to the destination resource
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
dst_path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.upload(path_or_file, str(dst_path), **kwargs)
async def upload_url(self: ResourceProtocol,
url: str,
relative_path: Optional[str] = None, /, **kwargs) -> OperationLinkObject:
"""
Upload a file from URL.
:param url: source URL
:param relative_path: `str` or `None`, destination path relative to the resource
:param disable_redirects: `bool`, forbid redirects
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ParentNotFoundError: parent directory doesn't exist
:raises PathExistsError: destination path already exists
:raises InsufficientStorageError: cannot upload file due to lack of storage space
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:raises UploadTrafficLimitExceededError: upload limit has been exceeded
:returns: :any:`OperationLinkObject`, link to the asynchronous operation
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
dst_path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.upload_url(url, str(dst_path), **kwargs)
async def get_download_link(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> str:
"""
Get a download link for a file (or a directory).
:param relative_path: `str` or `None`, path relative to the resource
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: `str`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.get_download_link(str(path), **kwargs)
@overload
async def download(self: ResourceProtocol,
dst_path_or_file: FileOrPathDestination, /, **kwargs) -> "ResourceLinkObject":
pass
@overload
async def download(self: ResourceProtocol,
relative_path: Optional[str],
dst_path_or_file: FileOrPathDestination, /, **kwargs) -> "ResourceLinkObject":
pass
async def download(self: ResourceProtocol, *args, **kwargs) -> "ResourceLinkObject":
"""
Download the file. This method takes 1 or 2 positional arguments:
1. :code:`download(dst_path_or_file, /, **kwargs)`
2. :code:`download(relative_path, dst_path_or_file, /, **kwargs)`
If `relative_path` is empty or None (or not specified) this method
will try to use the `file` attribute as a download link.
:param relative_path: `str` or `None`, source path relative to the resource
:param dst_path_or_file: destination path or file-like object
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`ResourceLinkObject`, link to the source resource
"""
if len(args) == 1:
relative_path, dst_path_or_file = None, args[0]
elif len(args) == 2:
relative_path, dst_path_or_file = args
else:
raise TypeError("download() takes 1 or 2 positional arguments")
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if not relative_path and hasattr(self, "file") and self.file is not None:
await self._yadisk.download_by_link(self.file, dst_path_or_file, **kwargs)
return ResourceLinkObject.from_path(self.path, yadisk=self._yadisk)
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
src_path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.download(str(src_path), dst_path_or_file, **kwargs)
@overload
async def patch(self: ResourceProtocol, properties: dict, **kwargs) -> "ResourceObject":
pass
@overload
async def patch(self: ResourceProtocol,
relative_path: Union[str, None],
properties: dict, **kwargs) -> "ResourceObject":
pass
async def patch(self: ResourceProtocol, *args, **kwargs) -> "ResourceObject":
"""
Update custom properties of a resource.
This method takes 1 or 2 positional arguments:
1. :code:`patch(properties, /, **kwargs)`
2. :code:`patch(relative_path, properties, /, **kwargs)`
:param relative_path: `str` or `None`, path relative to the resource
:param properties: `dict`, custom properties to update
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`ResourceObject`
"""
if len(args) == 1:
relative_path, properties = None, args[0]
elif len(args) == 2:
relative_path, properties = args
else:
raise TypeError("patch() takes 1 or 2 positional arguments")
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.patch(str(path), properties, **kwargs)
async def publish(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> "ResourceLinkObject":
"""
Make a resource public.
:param relative_path: `str` or `None`, relative path to the resource to be published
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`ResourceLinkObject`, link to the resource
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.publish(str(path), **kwargs)
async def unpublish(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> "ResourceLinkObject":
"""
Make a public resource private.
:param relative_path: `str` or `None`, relative path to the resource to be unpublished
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`ResourceLinkObject`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.unpublish(str(path), **kwargs)
async def mkdir(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> "ResourceLinkObject":
"""
Create a new directory.
:param relative_path: `str` or `None`, relative path to the directory to be created
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ParentNotFoundError: parent directory doesn't exist
:raises DirectoryExistsError: destination path already exists
:raises InsufficientStorageError: cannot create directory due to lack of storage space
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`ResourceLinkObject`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.mkdir(str(path), **kwargs)
async def remove(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> Optional[OperationLinkObject]:
"""
Remove the resource.
:param relative_path: `str` or `None`, relative path to the resource to be removed
:param permanently: if `True`, the resource will be removed permanently,
otherwise, it will be just moved to the trash
:param md5: `str`, MD5 hash of the file to remove
:param force_async: forces the operation to be executed asynchronously
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises BadRequestError: MD5 check is only available for files
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`OperationLinkObject` if the operation is performed asynchronously, `None` otherwise
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.remove(str(path), **kwargs)
@overload
async def move(self: ResourceProtocol,
dst_path: str, /, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
pass
@overload
async def move(self: ResourceProtocol,
relative_path: Optional[str],
dst_path: str, /, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
pass
async def move(self: ResourceProtocol, *args, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
"""
Move resource to `dst_path`.
This method takes 1 or 2 positional arguments:
1. :code:`move(dst_path, /, **kwargs)`
2. :code:`move(relative_path, dst_path, /, **kwargs)`
:param relative_path: `str` or `None`, source path to be moved relative to the resource
:param dst_path: destination path
:param overwrite: `bool`, determines whether to overwrite the destination
:param force_async: forces the operation to be executed asynchronously
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises PathExistsError: destination path already exists
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`ResourceLinkObject` or :any:`OperationLinkObject`
"""
if len(args) == 1:
relative_path, dst_path = None, args[0]
elif len(args) == 2:
relative_path, dst_path = args
else:
raise TypeError("move() takes 1 or 2 positional arguments")
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
src_path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.move(str(src_path), dst_path, **kwargs)
@overload
async def rename(self: ResourceProtocol,
new_name: str, /, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
pass
@overload
async def rename(self: ResourceProtocol,
relative_path: Optional[str],
new_name: str, /, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
pass
async def rename(self: ResourceProtocol, *args, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
"""
Rename `src_path` to have filename `new_name`.
Does the same as `move()` but changes only the filename.
:param relative_path: `str` or `None`, source path to be renamed relative to the resource
:param new_name: target filename to rename to
:param overwrite: `bool`, determines whether to overwrite the destination
:param force_async: forces the operation to be executed asynchronously
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises PathExistsError: destination path already exists
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:raises ValueError: `new_name` is not a valid filename
:returns: :any:`ResourceLinkObject` or :any:`OperationLinkObject`
"""
if len(args) == 1:
relative_path, new_name = None, args[0]
elif len(args) == 2:
relative_path, new_name = args
else:
raise TypeError("rename() takes 1 or 2 positional arguments")
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.rename(str(path), new_name, **kwargs)
@overload
async def copy(self: ResourceProtocol,
dst_path: str, /, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
pass
@overload
async def copy(self: ResourceProtocol,
relative_path: Optional[str],
dst_path: str, /, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
pass
async def copy(self: ResourceProtocol, *args, **kwargs) -> Union["ResourceLinkObject", OperationLinkObject]:
"""
Copy resource to `dst_path`.
If the operation is performed asynchronously, returns the link to the operation,
otherwise, returns the link to the newly created resource.
This method takes 1 or 2 positional arguments:
1. :code:`copy(dst_path, /, **kwargs)`
2. :code:`copy(relative_path, dst_path, /, **kwargs)`
:param src_path: `str` or `None`, source path relative to the resource
:param dst_path: destination path
:param overwrite: if `True` the destination path can be overwritten,
otherwise, an error will be raised
:param force_async: forces the operation to be executed asynchronously
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises PathExistsError: destination path already exists
:raises ForbiddenError: application doesn't have enough rights for this request
:raises InsufficientStorageError: cannot complete request due to lack of storage space
:raises ResourceIsLockedError: resource is locked by another request
:raises UploadTrafficLimitExceededError: upload limit has been exceeded
:returns: :any:`ResourceLinkObject` or :any:`OperationLinkObject`
"""
if len(args) == 1:
relative_path, dst_path = None, args[0]
elif len(args) == 2:
relative_path, dst_path = args
else:
raise TypeError("copy() takes 1 or 2 positional arguments")
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("ResourceObject doesn't have a path")
src_path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.copy(str(src_path), dst_path, **kwargs)
[документация]
class ResourceObject(YaDiskObject, ResourceObjectMethodsMixin):
"""
Resource object.
:param resource: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar antivirus_status: `str`, antivirus check status
:ivar file: `str`, download URL
:ivar size: `int`, file size
:ivar public_key: `str`, public resource key
:ivar sha256: `str`, SHA256 hash
:ivar md5: `str`, MD5 hash
:ivar embedded: :any:`ResourceListObject`, list of nested resources
:ivar name: `str`, filename
:ivar exif: :any:`EXIFObject`, EXIF metadata
:ivar resource_id: `str`, resource ID
:ivar custom_properties: `dict`, custom resource properties
:ivar public_url: `str`, public URL
:ivar share: :any:`ShareInfoObject`, shared folder information
:ivar modified: :any:`datetime.datetime`, date of last modification
:ivar created: :any:`datetime.datetime`, date of creation
:ivar photoslice_time: :any:`datetime.datetime`, photo/video creation date
:ivar mime_type: `str`, MIME type
:ivar path: `str`, path to the resource
:ivar preview: `str`, file preview URL
:ivar comment_ids: :any:`CommentIDsObject`, comment IDs
:ivar type: `str`, type ("file" or "dir")
:ivar media_type: `str`, file type as determined by Yandex.Disk
:ivar revision: `int`, Yandex.Disk revision at the time of last modification
"""
antivirus_status: Optional[str]
file: Optional[str]
size: Optional[int]
public_key: Optional[str]
sha256: Optional[str]
embedded: Optional["ResourceListObject"]
_embedded: Optional["ResourceListObject"]
name: Optional[str]
exif: Optional[EXIFObject]
resource_id: Optional[str]
custom_properties: Optional[dict]
public_url: Optional[str]
share: Optional["ShareInfoObject"]
modified: Optional["datetime.datetime"]
created: Optional["datetime.datetime"]
photoslice_time: Optional["datetime.datetime"]
mime_type: Optional[str]
path: Optional[str]
preview: Optional[str]
comment_ids: Optional[CommentIDsObject]
type: Optional[str]
media_type: Optional[str]
md5: Optional[str]
revision: Optional[int]
def __init__(self, resource: Optional[dict] = None, yadisk: Optional["YaDisk"] = None):
YaDiskObject.__init__(
self,
{"antivirus_status": str_or_dict_or_error,
"file": str_or_error,
"size": int_or_error,
"public_key": str_or_error,
"sha256": str_or_error,
"embedded": partial(ResourceListObject, yadisk=yadisk),
"name": str_or_error,
"exif": partial(EXIFObject, yadisk=yadisk),
"resource_id": str_or_error,
"custom_properties": dict_or_error,
"public_url": str_or_error,
"share": partial(ShareInfoObject, yadisk=yadisk),
"modified": yandex_date,
"created": yandex_date,
"photoslice_time": yandex_date,
"mime_type": str_or_error,
"path": str_or_error,
"preview": str_or_error,
"comment_ids": partial(CommentIDsObject, yadisk=yadisk),
"type": str_or_error,
"media_type": str_or_error,
"md5": str_or_error,
"revision": int_or_error},
yadisk)
self.set_alias("_embedded", "embedded")
self.import_fields(resource)
[документация]
class ResourceLinkObject(LinkObject, ResourceObjectMethodsMixin):
"""
Resource link object.
:param link: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar href: `str`, link URL
:ivar method: `str`, HTTP method
:ivar templated: `bool`, tells whether the URL is templated
:ivar path: `str`, path to the resource
"""
path: Optional[str]
def __init__(self, link: Optional[dict] = None, yadisk: Optional["YaDisk"] = None):
LinkObject.__init__(self, link, yadisk)
self.set_field_type("path", str_or_error)
self.set_field_type("public_key", str_or_error)
self.set_field_type("public_url", str_or_error)
if self.href is not None and is_resource_link(self.href):
try:
self.path = ensure_path_has_schema(
parse_qs(urlparse(self.href).query).get("path", [])[0])
except IndexError:
pass
@staticmethod
def from_path(path: Optional[str], yadisk: Optional["YaDisk"] = None) -> "ResourceLinkObject":
if path is None:
return ResourceLinkObject(yadisk=yadisk)
path = ensure_path_has_schema(path)
return ResourceLinkObject(
{"method": "GET",
"href": "https://cloud-api.yandex.net/v1/disk/resources?" + urlencode({"path": path}),
"templated": False},
yadisk=yadisk)
[документация]
class PublicResourceLinkObject(LinkObject, ResourceObjectMethodsMixin):
"""
Public resource link object.
:param link: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar href: `str`, link URL
:ivar method: `str`, HTTP method
:ivar templated: `bool`, tells whether the URL is templated
:ivar public_key: `str`, public key of the resource
:ivar public_url: `str`, public URL of the resource
"""
public_key: Optional[str]
public_url: Optional[str]
def __init__(self, link: Optional[dict] = None, yadisk: Optional["YaDisk"] = None):
LinkObject.__init__(self, link, yadisk)
self.set_field_type("public_key", str_or_error)
self.set_field_type("public_url", str_or_error)
self.set_field_type("path", str_or_error)
if self.href is not None and is_public_resource_link(self.href):
try:
public_key_or_url = parse_qs(urlparse(self.href).query).get("public_key", [])[0]
except IndexError:
public_key_or_url = ""
if public_key_or_url.startswith("http://") or public_key_or_url.startswith("https://"):
self.public_url = public_key_or_url
else:
self.public_key = public_key_or_url
@staticmethod
def from_public_key(public_key: Optional[str], yadisk: Optional["YaDisk"] = None) -> "PublicResourceLinkObject":
if public_key is None:
return PublicResourceLinkObject(yadisk=yadisk)
return PublicResourceLinkObject(
{"method": "GET",
"href": "https://cloud-api.yandex.net/v1/disk/public/resources?" + urlencode({"public_key": public_key}),
"templated": False},
yadisk=yadisk)
[документация]
class ResourceListObject(YaDiskObject):
"""
List of resources.
:param resource_list: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar sort: `str`, sort type
:ivar items: `list`, list of resources (:any:`ResourceObject`)
:ivar limit: `int`, maximum number of elements in the list
:ivar offset: `int`, offset from the beginning of the list
:ivar path: `str`, path to the directory that contains the elements of the list
:ivar total: `int`, number of elements in the list
"""
sort: Optional[str]
items: Optional[List[ResourceObject]]
limit: Optional[int]
offset: Optional[int]
path: Optional[str]
total: Optional[int]
def __init__(self, resource_list: Optional[dict] = None, yadisk: Optional["YaDisk"] = None):
YaDiskObject.__init__(
self,
{"sort": str_or_error,
"items": typed_list(partial(ResourceObject, yadisk=yadisk)),
"limit": int_or_error,
"offset": int_or_error,
"path": str_or_error,
"total": int_or_error},
yadisk)
self.import_fields(resource_list)
[документация]
class ResourceUploadLinkObject(LinkObject):
"""
Resource upload link.
:param resource_upload_link: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar operation_id: `str`, ID of the upload operation
:ivar href: `str`, link URL
:ivar method: `str`, HTTP method
:ivar templated: `bool`, tells whether the URL is templated
"""
operation_id: Optional[str]
def __init__(self,
resource_upload_link: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
LinkObject.__init__(self, None, yadisk)
self.set_field_type("operation_id", str_or_error)
self.import_fields(resource_upload_link)
[документация]
class ResourceDownloadLinkObject(LinkObject):
"""
Resource download link.
:param link: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar href: `str`, link URL
:ivar method: `str`, HTTP method
:ivar templated: `bool`, tells whether the URL is templated
"""
pass
[документация]
class ShareInfoObject(YaDiskObject):
"""
Shared folder information object.
:param share_info: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar is_root: `bool`, tells whether the folder is root
:ivar is_owned: `bool`, tells whether the user is the owner of this directory
:ivar rights: `str`, access rights
"""
is_root: Optional[bool]
is_owned: Optional[bool]
rights: Optional[str]
def __init__(self, share_info: Optional[dict] = None, yadisk: Optional["YaDisk"] = None):
YaDiskObject.__init__(
self,
{"is_root": bool_or_error,
"is_owned": bool_or_error,
"rights": str_or_error},
yadisk)
self.import_fields(share_info)
[документация]
class PublicResourceObject(ResourceObject):
"""
Public resource object.
:param resource: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar antivirus_status: `str`, antivirus check status
:ivar file: `str`, download URL
:ivar size: `int`, file size
:ivar public_key: `str`, public resource key
:ivar sha256: `str`, SHA256 hash
:ivar md5: `str`, MD5 hash
:ivar embedded: :any:`PublicResourceObject`, list of nested resources
:ivar name: `str`, filename
:ivar exif: :any:`EXIFObject`, EXIF metadata
:ivar resource_id: `str`, resource ID
:ivar custom_properties: `dict`, custom resource properties
:ivar public_url: `str`, public URL
:ivar share: :any:`ShareInfoObject`, shared folder information
:ivar modified: :any:`datetime.datetime`, date of last modification
:ivar created: :any:`datetime.datetime`, date of creation
:ivar photoslice_time: :any:`datetime.datetime`, photo/video creation date
:ivar mime_type: `str`, MIME type
:ivar path: `str`, path to the resource
:ivar preview: `str`, file preview URL
:ivar comment_ids: :any:`CommentIDsObject`, comment IDs
:ivar type: `str`, type ("file" or "dir")
:ivar media_type: `str`, file type as determined by Yandex.Disk
:ivar revision: `int`, Yandex.Disk revision at the time of last modification
:ivar view_count: `int`, number of times the public resource was viewed
:ivar owner: :any:`UserPublicInfoObject`, owner of the public resource
"""
views_count: Optional[int]
view_count: Optional[int]
embedded: Optional["PublicResourceListObject"]
_embedded: Optional["PublicResourceListObject"]
owner: Optional[UserPublicInfoObject]
def __init__(self, public_resource=None, yadisk=None):
ResourceObject.__init__(self, None, yadisk)
self.set_field_type("views_count", int_or_error)
self.set_alias("view_count", "views_count")
self.set_field_type("embedded", partial(PublicResourceListObject, yadisk=yadisk))
self.set_field_type("owner", partial(UserPublicInfoObject, yadisk=yadisk))
self.import_fields(public_resource)
[документация]
class PublicResourceListObject(ResourceListObject):
"""
List of public resources.
:param public_resource_list: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar sort: `str`, sort type
:ivar items: `list`, list of resources (:any:`ResourceObject`)
:ivar limit: `int`, maximum number of elements in the list
:ivar offset: `int`, offset from the beginning of the list
:ivar path: `str`, path to the directory that contains the elements of the list
:ivar total: `int`, number of elements in the list
:ivar public_key: `str`, public key of the resource
"""
public_key: Optional[str]
items: Optional[List[PublicResourceObject]]
def __init__(self,
public_resource_list: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
ResourceListObject.__init__(self, None, yadisk)
self.set_field_type("public_key", str_or_error)
self.set_field_type("items", typed_list(partial(PublicResourceObject, yadisk=yadisk)))
self.import_fields(public_resource_list)
[документация]
class TrashResourceObject(ResourceObject):
"""
Trash resource object.
:param trash_resource: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar antivirus_status: `str`, antivirus check status
:ivar file: `str`, download URL
:ivar size: `int`, file size
:ivar public_key: `str`, public resource key
:ivar sha256: `str`, SHA256 hash
:ivar md5: `str`, MD5 hash
:ivar embedded: :any:`TrashResourceListObject`, list of nested resources
:ivar name: `str`, filename
:ivar exif: :any:`EXIFObject`, EXIF metadata
:ivar resource_id: `str`, resource ID
:ivar custom_properties: `dict`, custom resource properties
:ivar public_url: `str`, public URL
:ivar share: :any:`ShareInfoObject`, shared folder information
:ivar modified: :any:`datetime.datetime`, date of last modification
:ivar created: :any:`datetime.datetime`, date of creation
:ivar photoslice_time: :any:`datetime.datetime`, photo/video creation date
:ivar mime_type: `str`, MIME type
:ivar path: `str`, path to the resource
:ivar preview: `str`, file preview URL
:ivar comment_ids: :any:`CommentIDsObject`, comment IDs
:ivar type: `str`, type ("file" or "dir")
:ivar media_type: `str`, file type as determined by Yandex.Disk
:ivar revision: `int`, Yandex.Disk revision at the time of last modification
:ivar origin_path: `str`, original path
:ivar deleted: :any:`datetime.datetime`, date of deletion
"""
embedded: Optional["TrashResourceListObject"]
_embedded: Optional["TrashResourceListObject"]
origin_path: Optional[str]
deleted: Optional["datetime.datetime"]
def __init__(self,
trash_resource: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
ResourceObject.__init__(self, None, yadisk)
self.set_field_type("embedded", partial(TrashResourceListObject, yadisk=yadisk))
self.set_field_type("origin_path", str_or_error)
self.set_field_type("deleted", yandex_date)
self.import_fields(trash_resource)
[документация]
async def exists(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> bool:
"""
Check whether the trash resource exists.
:param relative_path: `str` or `None`, relative path to the trash resource
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ForbiddenError: application doesn't have enough rights for this request
:returns: `bool`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("TrashResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.trash_exists(str(path), **kwargs)
[документация]
async def get_type(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> str:
"""
Get trash resource type.
:param relative_path: `str` or `None`, relative path to the trash resource
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:returns: "file" or "dir"
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("TrashResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.get_trash_type(str(path), **kwargs)
[документация]
async def is_dir(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> bool:
"""
Check whether resource is a trash directory.
:param relative_path: `str` or `None`, relative path to the trash resource
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ForbiddenError: application doesn't have enough rights for this request
:returns: `True` if `path` is a directory, `False` otherwise (even if it doesn't exist)
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("TrashResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.is_trash_dir(str(path), **kwargs)
[документация]
async def is_file(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> bool:
"""
Check whether resource is a trash file.
:param relative_path: `str` or `None`, relative path to the trash resource
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 ForbiddenError: application doesn't have enough rights for this request
:returns: `True` if `path` is a file, `False` otherwise (even if it doesn't exist)
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("TrashResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.is_trash_file(str(path), **kwargs)
[документация]
async def listdir(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> AsyncGenerator["TrashResourceObject", None]:
"""
Get contents of a trash resource.
:param relative_path: `str` or `None`, relative path to the directory in the trash bin
:param limit: number of children resources to be included in the response
:param offset: number of children resources to be skipped in the response
:param preview_size: size of the file preview
:param preview_crop: `bool`, cut the preview to the size specified in the `preview_size`
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises WrongResourceTypeError: resource is not a directory
:returns: generator of :any:`TrashResourceObject`
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("TrashResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.trash_listdir(str(path), **kwargs)
[документация]
async def remove(self: ResourceProtocol,
relative_path: Optional[str] = None, /, **kwargs) -> Optional[OperationLinkObject]:
"""
Remove a trash resource.
:param relative_path: `str` or `None`, relative path to the trash resource to be deleted
:param force_async: forces the operation to be executed asynchronously
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`OperationLinkObject` if the operation is performed asynchronously, `None` otherwise
"""
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("TrashResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.remove_trash(str(path), **kwargs)
@overload
async def restore(self: ResourceProtocol,
dst_path: str, /, **kwargs) -> Union[ResourceLinkObject, OperationLinkObject]:
pass
@overload
async def restore(self: ResourceProtocol,
relative_path: Optional[str],
dst_path: str, /, **kwargs) -> Union[ResourceLinkObject, OperationLinkObject]:
pass
[документация]
async def restore(self: ResourceProtocol, *args, **kwargs) -> Union[ResourceLinkObject, OperationLinkObject]:
"""
Restore a trash resource.
Returns a link to the newly created resource or a link to the asynchronous operation.
This method takes 1 or 2 positional arguments:
1. :code:`restore(dst_path, /, **kwargs)`
2. :code:`restore(relative_path=None, dst_path, /, **kwargs)`
:param relative_path: `str` or `None`, relative path to the trash resource to be restored
:param dst_path: destination path
:param overwrite: `bool`, determines whether the destination can be overwritten
:param force_async: forces the operation to be executed asynchronously
:param fields: list of keys to be included in the response
:param timeout: `float` or :any:`aiohttp.ClientTimeout`, 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 PathNotFoundError: resource was not found on Disk
:raises PathExistsError: destination path already exists
:raises ForbiddenError: application doesn't have enough rights for this request
:raises ResourceIsLockedError: resource is locked by another request
:returns: :any:`ResourceLinkObject` or :any:`OperationLinkObject`
"""
if len(args) == 0:
relative_path = dst_path = None
elif len(args) == 1:
relative_path, dst_path = None, args[0]
elif len(args) == 2:
relative_path, dst_path = args
else:
raise TypeError("restore() takes up to 2 positional arguments")
if self._yadisk is None:
raise ValueError("This object is not bound to a YaDisk instance")
if self.path is None:
raise ValueError("TrashResourceObject doesn't have a path")
path = PurePosixPath(self.path) / (relative_path or "")
return await self._yadisk.restore_trash(str(path), dst_path, **kwargs)
async def get_public_meta(self, *args, **kwargs):
""""""
raise NotImplementedError
async def public_listdir(self, *args, **kwargs):
""""""
raise NotImplementedError
async def get_download_link(self, *args, **kwargs):
""""""
raise NotImplementedError
async def download(self, *args, **kwargs):
""""""
raise NotImplementedError
async def get_upload_link(self, *args, **kwargs):
""""""
raise NotImplementedError
async def upload(self, *args, **kwargs):
""""""
raise NotImplementedError
async def copy(self, *args, **kwargs):
""""""
raise NotImplementedError
async def move(self, *args, **kwargs):
""""""
raise NotImplementedError
async def rename(self, *args, **kwargs):
""""""
raise NotImplementedError
async def upload_url(self, *args, **kwargs):
""""""
raise NotImplementedError
[документация]
async def patch(self, *args, **kwargs):
raise NotImplementedError
async def publish(self, *args, **kwargs):
""""""
raise NotImplementedError
async def unpublish(self, *args, **kwargs):
""""""
raise NotImplementedError
async def mkdir(self, *args, **kwargs):
""""""
raise NotImplementedError
[документация]
class TrashResourceListObject(ResourceListObject):
"""
List of trash resources.
:param trash_resource_list: `dict` or `None`
:param yadisk: :any:`YaDisk` or `None`, `YaDisk` object
:ivar sort: `str`, sort type
:ivar items: `list`, list of resources (:any:`TrashResourceObject`)
:ivar limit: `int`, maximum number of elements in the list
:ivar offset: `int`, offset from the beginning of the list
:ivar path: `str`, path to the directory that contains the elements of the list
:ivar total: `int`, number of elements in the list
"""
items: Optional[List[TrashResourceObject]]
def __init__(self,
trash_resource_list: Optional[dict] = None,
yadisk: Optional["YaDisk"] = None):
ResourceListObject.__init__(self, None, yadisk)
self.set_field_type("items", typed_list(partial(TrashResourceObject, yadisk=yadisk)))
self.import_fields(trash_resource_list)