YaDisk

https://yadisk.readthedocs.io/en/latest/
Admin
YaDisk is a Python library providing synchronous and asynchronous API access to the Yandex.Disk REST...
Tokens:127,974
Snippets:620
Trust Score:8.3
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install optional HTTP client libraries

Source: https://yadisk.readthedocs.io/en/latest/_sources/intro.rst.txt

Manual installation of specific HTTP client libraries for custom configurations.

```bash
# For use with pycurl
pip install yadisk[pycurl]

# For use with aiohttp, will also install aiofiles
pip install yadisk[async-files,aiohttp]
```

--------------------------------

### Install asynchronous API dependencies

Source: https://yadisk.readthedocs.io/en/latest/_sources/intro.rst.txt

Installs the default aiohttp/httpx-based asynchronous client.

```bash
pip install yadisk[async-defaults]
```

--------------------------------

### Install synchronous API dependencies

Source: https://yadisk.readthedocs.io/en/latest/_sources/intro.rst.txt

Installs the default requests-based synchronous client.

```bash
pip install yadisk[sync-defaults]
```

--------------------------------

### AsyncHTTPXSession Usage Example

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/sessions/async_httpx_session.html

Demonstrates how to initialize an AsyncClient with the httpx session and pass custom httpx-specific arguments.

```python
import yadisk

async def main():
    async with yadisk.AsyncClient(..., session="httpx") as client:
        await client.get_meta(
            "/my_file.txt",
            n_retries=5,
            httpx_args={
                "proxy": "http://localhost:11234",
                "verify": False,
                "max_redirects": 10
            }
         )
```

--------------------------------

### AIOHTTPSession Usage Example

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/sessions/aiohttp_session.html

Demonstrates how to initialize an AsyncClient with the aiohttp session and pass specific aiohttp arguments.

```python
import yadisk

async def main():
    async with yadisk.AsyncClient(..., session="aiohttp") as client:
        await client.get_meta(
            "/my_file.txt",
            n_retries=5,
            aiohttp_args={
                "proxies": {"https": "http://example.com:1234"},
                "verify": False
            }
         )
```

--------------------------------

### HTTPXSession Usage Example

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/sessions/httpx_session.html

Demonstrates how to initialize a yadisk Client with the httpx session and pass custom httpx-specific arguments.

```python
import yadisk

with yadisk.Client(..., session="httpx") as client:
    client.get_meta(
        "/my_file.txt",
        n_retries=5,
        httpx_args={
            "proxy": "http://localhost:11234",
            "verify": False,
            "max_redirects": 10
        }
     )
```

--------------------------------

### Using PycURLSession with custom options

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/sessions/pycurl_session.html

Example of initializing a Client with the pycurl session and passing specific curl options.

```python
import yadisk
import pycurl

with yadisk.Client(..., session="pycurl") as client:
    client.get_meta(
        "/my_file.txt",
        n_retries=5,
        curl_options={
            pycurl.MAX_SEND_SPEED_LARGE: 5 * 1024**2,
            pycurl.MAX_RECV_SPEED_LARGE: 5 * 1024**2,
            pycurl.PROXY: "http://localhost:12345",
            pycurl.MAXREDIRS: 15
        }
     )
```

--------------------------------

### RequestsSession Usage Example

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/sessions/requests_session.html

Demonstrates how to use RequestsSession with custom requests arguments via the yadisk Client.

```python
import yadisk

with yadisk.Client(..., session="requests") as client:
    client.get_meta(
        "/my_file.txt",
        n_retries=5,
        requests_args={
            "proxies": {"https": "http://example.com:1234"},
            "verify": False
        }
     )
```

--------------------------------

### Get Resource Type

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Determines if the resource at the path is a file or a directory.

```python
    def get_type(self, path: str, /, **kwargs) -> str:
        """
            Get resource type.

            :param path: path to the resource
            :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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param requests_args: `dict`, additional parameters for :any:`RequestsSession`
            :param httpx_args: `dict`, additional parameters for :any:`HTTPXSession`
            :param curl_options: `dict`, additional options for :any:`PycURLSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :raises PathNotFoundError: resource was not found on Disk
            :raises ForbiddenError: application doesn't have enough rights for this request

            :returns: "file" or "dir"
        """

        return _get_type(self.get_meta, path, **kwargs)
```

--------------------------------

### Client.get_trash_meta

Source: https://yadisk.readthedocs.io/en/latest/api_reference/sync_api.html

Get meta information about a trash resource.

```APIDOC
## Client.get_trash_meta(path, **kwargs)

### Description
Get meta information about a trash resource.

### Parameters
- **path** (str) - Required - path to the trash resource
- **limit** (int) - Optional - number of children resources to be included in the response
- **offset** (int) - Optional - number of children resources to be skipped in the response
- **preview_size** (str) - Optional - size of the file preview
- **preview_crop** (bool) - Optional - cut the preview to the size specified in the preview_size
- **sort** (str) - Optional - field to be used as a key to sort children resources
- **fields** (list) - Optional - list of keys to be included in the response

### Returns
- **SyncTrashResourceObject** - The metadata object for the trash resource.
```

--------------------------------

### get_public_meta(public_key, **kwargs)

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Get meta-information about a public resource.

```APIDOC
## get_public_meta(public_key, **kwargs)

### Description
Get meta-information about a public resource.

### Parameters
- **public_key** (str) - Required - Public key or public URL of the public resource

### Returns
- **SyncPublicResourceObject**
```

--------------------------------

### Get public resource meta-information

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Retrieves meta-information about a public resource.

```python
    def get_public_meta(
        self,
        public_key: str,
        /,
        **kwargs
    ) -> "SyncPublicResourceObject":
        """
            Get meta-information about a public resource.
```

--------------------------------

### Client.__init__

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Initializes a new instance of the Client class to interact with the Yandex.Disk API.

```APIDOC
## Client(id, secret, token, default_args, session, open_file, session_factory)

### Description
Initializes the Yandex.Disk client. This client handles authentication and provides a session for making API requests.

### Parameters
- **id** (str) - Optional - Application ID.
- **secret** (str) - Optional - Application secret password.
- **token** (str) - Optional - Application token.
- **default_args** (dict) - Optional - Default arguments for methods (e.g., timeout, headers).
- **session** (Session|str) - Optional - HTTP session instance or name ('httpx', 'pycurl', 'requests').
- **open_file** (callable) - Optional - Function used to open files.
- **session_factory** (callable) - Optional - Factory function returning a session instance.
```

--------------------------------

### Initialize Client Class

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Constructor for the Client class, used to configure application credentials and session settings.

```python
def __init__(self,
             id:     str = "",
             secret: str = "",
             token:  str = "",
             *,
             default_args:    Optional[Dict[str, Any]] = None,
             session:         Optional[Union[Session, SessionName]] = None,
             open_file:       Optional[OpenFileCallback] = None,
             session_factory: Optional[SessionFactory] = None) -> None:
    self.id = id
    self.secret = secret
    self.token = ""
```

--------------------------------

### Get operation status

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_operations.html

Retrieves the current status of an asynchronous operation.

```python
[docs]
    async def get_status(self, **kwargs) -> OperationStatus:
        """
            Get operation status.

            :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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param aiohttp_args: `dict`, additional parameters for :any:`AIOHTTPSession`
            :param httpx_args: `dict`, additional parameters for :any:`AsyncHTTPXSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :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)
```

--------------------------------

### AsyncClient.mkdir

Source: https://yadisk.readthedocs.io/en/latest/api_reference/async_api.html

Creates a new directory at the specified path.

```APIDOC
## AsyncClient.mkdir(path: str, **kwargs)

### Description
Create a new directory at the specified path.

### Parameters
- **path** (str) - Required - Path to the directory to be created
- **fields** (list) - Optional - List of keys to be included in the response
- **timeout** (float/tuple/None) - Optional - Request timeout
- **headers** (dict/None) - Optional - Additional request headers
- **n_retries** (int) - Optional - Maximum number of retries
- **retry_interval** (int) - Optional - Delay between retries in seconds
- **retry_on** (tuple) - Optional - Additional exception classes to retry on
- **aiohttp_args** (dict) - Optional - Additional parameters for AIOHTTPSession
- **httpx_args** (dict) - Optional - Additional parameters for AsyncHTTPXSession

### Returns
- **AsyncResourceLinkObject** - The created resource link object.

### Raises
- **ParentNotFoundError** - Parent directory doesn't exist
- **DirectoryExistsError** - Destination path already exists
- **InsufficientStorageError** - Lack of storage space
- **ForbiddenError** - Insufficient rights
- **ResourceIsLockedError** - Resource is locked
```

--------------------------------

### Get Disk Information

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Retrieves metadata and capacity information about the Yandex.Disk account.

```python
    def get_disk_info(self, **kwargs) -> "DiskInfoObject":
        """
            Get disk information.

            :param extra_fields: list of additional keys to be included in the response
            :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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param requests_args: `dict`, additional parameters for :any:`RequestsSession`
            :param httpx_args: `dict`, additional parameters for :any:`HTTPXSession`
            :param curl_options: `dict`, additional options for :any:`PycURLSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :raises ForbiddenError: application doesn't have enough rights for this request

            More info about this request:

            * `Official docs <https://yandex.com/dev/disk-api/doc/en/reference/capacity>`__
            * `Polygon <https://yandex.com/dev/disk/poligon#!/v147disk/GetDisk>`__

            :returns: :any:`DiskInfoObject`
        """
```

--------------------------------

### Create directory

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Creates a new directory at the specified path.

```python
    def mkdir(self, path: str, /, **kwargs) -> SyncResourceLinkObject:
        """
            Create a new directory.
```

--------------------------------

### Get Upload Link

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Generates a URL for uploading a file to a specific path on Yandex.Disk.

```python
        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 self._yadisk.get_upload_link(str(path), **kwargs)
```

--------------------------------

### AsyncClient.makedirs

Source: https://yadisk.readthedocs.io/en/latest/api_reference/async_api.html

Creates a new directory at the specified path, creating parent directories recursively if they do not exist.

```APIDOC
## AsyncClient.makedirs(path: str, **kwargs)

### Description
Create a new directory at path. If its parent directory doesn't exist it will also be created recursively.

### Parameters
- **path** (str) - Required - Path to the directory to be created
- **fields** (list) - Optional - List of keys to be included in the response
- **timeout** (float/tuple/None) - Optional - Request timeout
- **headers** (dict/None) - Optional - Additional request headers
- **n_retries** (int) - Optional - Maximum number of retries
- **retry_interval** (int) - Optional - Delay between retries in seconds
- **retry_on** (tuple) - Optional - Additional exception classes to retry on

### Returns
- **AsyncResourceLinkObject** - The created resource link object.

### Raises
- **DirectoryExistsError** - Destination path already exists
- **InsufficientStorageError** - Lack of storage space
- **ForbiddenError** - Insufficient rights
- **ResourceIsLockedError** - Resource is locked
```

--------------------------------

### Get Operation Status

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Retrieves the current status of a Yandex.Disk operation using its ID.

```python
    def get_operation_status(self, operation_id: str, /, **kwargs) -> OperationStatus:
        """
            Get operation status.

            :param operation_id: ID of the operation or a link
            :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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param requests_args: `dict`, additional parameters for :any:`RequestsSession`
            :param httpx_args: `dict`, additional parameters for :any:`HTTPXSession`
            :param curl_options: `dict`, additional options for :any:`PycURLSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :raises OperationNotFoundError: requested operation was not found

            More info about this request:

            * `Official docs <https://yandex.com/dev/disk-api/doc/en/reference/operations>`__
            * `Polygon <https://yandex.com/dev/disk/poligon#!/v147disk47operations47123operation95id125/GetOperationStatus/>`__

            :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
        """

        _apply_default_args(kwargs, self.default_args)
        _add_authorization_header(kwargs, self.token)

        return GetOperationStatusRequest(
            self.session, operation_id, fields=["status"], **kwargs
        ).send(yadisk=self).status
```

--------------------------------

### Get Disk Information Asynchronously

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_async_client.html

Retrieves metadata and capacity information for the Yandex.Disk account.

```python
    async def get_disk_info(self, **kwargs) -> "DiskInfoObject":
        """
            Get disk information.

            :param extra_fields: list of additional keys to be included in the response
            :param fields: list of keys to be included in the response
            :param timeout: `float`, `tuple` or `None`, 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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param aiohttp_args: `dict`, additional parameters for :any:`AIOHTTPSession`
            :param httpx_args: `dict`, additional parameters for :any:`AsyncHTTPXSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :raises ForbiddenError: application doesn't have enough rights for this request

            More info about this request:

            * `Official docs <https://yandex.com/dev/disk-api/doc/en/reference/capacity>`__
            * `Polygon <https://yandex.com/dev/disk/poligon#!/v147disk/GetDisk>`__

            :returns: :any:`DiskInfoObject`
        """

        _apply_default_args(kwargs, self.default_args)
        _add_authorization_header(kwargs, self.token)

        return await DiskInfoRequest(self.session, **kwargs).asend(yadisk=self)
```

--------------------------------

### Create directories recursively with makedirs

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Creates a directory and all necessary parent directories recursively.

```python
while True:
    try:
        return self.mkdir(path, **kwargs)
    except ParentNotFoundError as e:
        # We first have to remove the scheme, otherwise posixpath.split()
        # may treat it as part of the path
        scheme, path_without_scheme = remove_path_scheme(path)

        # Extract the parent directory
        head, _tail = posixpath.split(path_without_scheme)
        head = head.strip("/")

        if head == "":
            # We should never find ourselves in this situation
            raise e from None

        # Restore the scheme
        if scheme:
            head = f"{scheme}:/{head}"

        self.makedirs(head, **kwargs)
```

--------------------------------

### SystemFoldersObject Initialization

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_disk.html

Initializes the SystemFoldersObject to map system folder paths from a dictionary.

```python
def __init__(
    self,
    system_folders: Optional[Dict] = None,
    yadisk: Optional[Any] = None
) -> None:
    YaDiskObject.__init__(
        self,
        {"odnoklassniki": str_or_error,
         "google":        str_or_error,
         "instagram":     str_or_error,
         "vkontakte":     str_or_error,
         "attach":        str_or_error,
         "mailru":        str_or_error,
         "downloads":     str_or_error,
         "applications":  str_or_error,
         "facebook":      str_or_error,
         "social":        str_or_error,
         "messenger":     str_or_error,
         "calendar":      str_or_error,
         "photostream":   str_or_error,
         "screenshots":   str_or_error,
         "scans":         str_or_error},
        yadisk)

    self.import_fields(system_folders)
```

--------------------------------

### Get Last Uploaded Files

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Retrieves a list of the most recently uploaded files sorted by date.

```python
_apply_default_args(kwargs, self.default_args)
        _add_authorization_header(kwargs, self.token)

        response: "SyncLastUploadedResourceListObject" = LastUploadedRequest(
            self.session, **kwargs
        ).send(yadisk=self)

        if response.items is None:
            raise InvalidResponseError("Response did not contain key field")

        return response.items
```

--------------------------------

### Get trash resource type

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Returns the type of a trash resource as either 'file' or 'dir'.

```python
return _get_type(self.get_trash_meta, path, **kwargs)
```

--------------------------------

### Client.mkdir

Source: https://yadisk.readthedocs.io/en/latest/api_reference/sync_api.html

Creates a new directory at the specified path.

```APIDOC
## Client.mkdir

### Description
Create a new directory.

### Parameters
- **path** (str) - Required - path to the directory to be created
- **fields** (list) - Optional - list of keys to be included in the response
- **timeout** (float or tuple) - Optional - request timeout
- **headers** (dict) - Optional - additional request headers
- **n_retries** (int) - Optional - maximum number of retries
- **retry_interval** (int) - Optional - delay between retries in seconds
- **retry_on** (tuple) - Optional - additional exception classes to retry on
- **requests_args** (dict) - Optional - additional parameters for RequestsSession
- **httpx_args** (dict) - Optional - additional parameters for HTTPXSession
- **curl_options** (dict) - Optional - additional options for PycURLSession
- **kwargs** (dict) - Optional - any other parameters accepted by Session.send_request()

### Returns
- **SyncResourceLinkObject** - The created resource link object.
```

--------------------------------

### Get download link

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Retrieves a direct download link for a specified file or directory path.

```python
    def get_download_link(self, path: str, /, **kwargs) -> str:
        """
            Get a download link for a file (or a directory).

            :param path: path to the resource
            :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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param requests_args: `dict`, additional parameters for :any:`RequestsSession`
            :param httpx_args: `dict`, additional parameters for :any:`HTTPXSession`
            :param curl_options: `dict`, additional options for :any:`PycURLSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :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

            More info about this request:

```

--------------------------------

### Download Resource Method Implementation

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Handles downloading a resource or a file via a link, supporting optional relative paths.

```python
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 AsyncResourceLinkObject.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)
```

--------------------------------

### Get file upload link

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Retrieves a URL for uploading a file via a PUT request.

```python
[docs]
    def get_upload_link(
        self,
        path: str,
        /,
        spoof_user_agent: bool = True,
        **kwargs
    ) -> str:
        """
            Get a link to upload the file using the PUT request.
```

--------------------------------

### Get Upload Link Object

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Retrieves a ResourceUploadLinkObject containing an asynchronous operation ID for file uploads.

```python
    def get_upload_link_object(
        self: ResourceProtocol,
        relative_path: Optional[str] = None,
        /,
        **kwargs
    ) -> "ResourceUploadLinkObject":
        """
            Get a link to upload the file using the PUT request.
            This is similar to :any:`Client.get_upload_link()`, except it returns
            an instance of :any:`ResourceUploadLinkObject` which also contains
            an asynchronous operation ID.

            :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 spoof_user_agent: `bool`, if `True` (default), the `User-Agent` header
                will be set to a special value, which should allow bypassing of
                Yandex.Disk's upload speed limit
            :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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param requests_args: `dict`, additional parameters for :any:`RequestsSession`
            :param httpx_args: `dict`, additional parameters for :any:`HTTPXSession`
            :param curl_options: `dict`, additional options for :any:`PycURLSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :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:`ResourceUploadLinkObject`
        """

        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 self._yadisk.get_upload_link_object(str(path), **kwargs)
```

--------------------------------

### DiskObject Initialization

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_disk.html

Initializes the DiskObject with field mappings for disk information and system folders.

```python
def __init__(self, disk_info: Optional[Dict] = None, yadisk: Optional[Any] = None) -> None:
    YaDiskObject.__init__(
        self,
        {
            "deletion_restriction_days":            int_or_error,
            "free_photounlim_end_date":             int_or_error,
            "hide_screenshots_in_photoslice":       bool_or_error,
            "is_idm_managed_folder_address_access": bool_or_error,
            "is_idm_managed_public_access":         bool_or_error,
            "is_legal_entity":                      bool_or_error,
            "is_paid":                              bool_or_error,
            "max_file_size":                        int_or_error,
            "paid_max_file_size":                   int_or_error,
            "payment_flow":                         bool_or_error,
            "photounlim_size":                      int_or_error,
            "reg_time":                             yandex_date,
            "revision":                             int_or_error,
            "system_folders":                       partial(SystemFoldersObject, yadisk=yadisk),
            "total_space":                          int_or_error,
            "trash_size":                           int_or_error,
            "unlimited_autoupload_enabled":         bool_or_error,
            "used_space":                           int_or_error,
            "user":                                 partial(UserObject, yadisk=yadisk),
            "will_be_overdrawn":                    bool_or_error
        },
        yadisk
    )

    self.import_fields(disk_info)
```

--------------------------------

### Get public resource type

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Retrieves the type of a public resource, returning either 'file' or 'dir'.

```python
    def get_public_type(self, public_key: str, /, **kwargs) -> str:
        """
            Get public resource type.

            :param public_key: public key or public URL of the public resource
            :param path: relative path to the resource within the public folder
            :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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param requests_args: `dict`, additional parameters for :any:`RequestsSession`
            :param httpx_args: `dict`, additional parameters for :any:`HTTPXSession`
            :param curl_options: `dict`, additional options for :any:`PycURLSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :raises PathNotFoundError: resource was not found on Disk
            :raises ForbiddenError: application doesn't have enough rights for this request

            :returns: "file" or "dir"
        """

        return _get_type(self.get_public_meta, public_key, **kwargs)
```

--------------------------------

### Recursively create directories asynchronously

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Creates a directory and any missing parent directories recursively.

```python
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.makedirs(str(path), **kwargs)
```

--------------------------------

### Initialize AsyncClient

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_async_client.html

Constructor for the AsyncClient class, allowing configuration of credentials, session, and file handling.

```python
def __init__(
        self,
        id:     str = "",
        secret: str = "",
        token:  str = "",
        *,
        default_args:    Optional[Dict[str, Any]] = None,
        session:         Optional[Union[AsyncSession, AsyncSessionName]] = None,
        open_file:       Optional[AsyncOpenFileCallback] = None,
        session_factory: Optional[AsyncSessionFactory] = None
    ) -> None:
        self.id = id
        self.secret = secret

        self.token = ""

        self.default_args = {} if default_args is None else default_args

        if session is None:
            if session_factory is not None:
                session = session_factory()
            else:
                try:
                    session = import_async_session("httpx")()
                except ModuleNotFoundError as e:
                    if e.name == "httpx":
                        raise ModuleNotFoundError(
                            "httpx is not installed. Either install httpx or provide a custom session",
                            name=e.name,
                            path=e.path) from e
                    else:
                        raise
        elif isinstance(session, str):
            session = import_async_session(session)()

        self.session = session

        if open_file is None:
            open_file = _default_open_file

        self.open_file = open_file

        self.token = token
```

--------------------------------

### Get Authentication URL

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Constructs the authentication URL for user authorization without performing any HTTP requests.

```python
    def get_auth_url(
        self,
        type:                  Union[Literal["code"], Literal["token"]],
        device_id:             Optional[str] = None,
        device_name:           Optional[str] = None,
        redirect_uri:          Optional[str] = None,
        login_hint:            Optional[str] = None,
        scope:                 Optional[str] = None,
        optional_scope:        Optional[str] = None,
        force_confirm:         bool = True,
        state:                 Optional[str] = None,
        code_challenge:        Optional[str] = None,
        code_challenge_method: Optional[Union[Literal["plain"], Literal["S256"]]] = None,
        display:               None = None
    ) -> str:
        """
            Get authentication URL for the user to go to.
            This method doesn't send any HTTP requests and merely constructs the URL.
```

--------------------------------

### Download Method Overloads and Implementation

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Provides overloaded signatures and the implementation for downloading files.

```python
    @overload
    def download(self: ResourceProtocol,
                 dst_path_or_file: FileOrPathDestination, /, **kwargs) -> "SyncResourceLinkObject":
        pass

    @overload
    def download(self: ResourceProtocol,
                 relative_path: Optional[str],
                 dst_path_or_file: FileOrPathDestination, /, **kwargs) -> "SyncResourceLinkObject":
        pass

    def download(self: ResourceProtocol, /, *args, **kwargs) -> "SyncResourceLinkObject":
        """
            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 `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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param requests_args: `dict`, additional parameters for :any:`RequestsSession`
            :param httpx_args: `dict`, additional parameters for :any:`HTTPXSession`
            :param curl_options: `dict`, additional options for :any:`PycURLSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :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:`SyncResourceLinkObject`, 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:
            self._yadisk.download_by_link(self.file, dst_path_or_file, **kwargs)

            return SyncResourceLinkObject.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 self._yadisk.download(str(src_path), dst_path_or_file, **kwargs)
```

--------------------------------

### Client.makedirs

Source: https://yadisk.readthedocs.io/en/latest/api_reference/sync_api.html

Creates a new directory at the specified path, creating parent directories recursively if they do not exist.

```APIDOC
## Client.makedirs

### Description
Create a new directory at path. If its parent directory doesn’t exist it will also be created recursively.

### Parameters
- **path** (str) - Required - path to the directory to be created
- **fields** (list) - Optional - list of keys to be included in the response
- **timeout** (float or tuple) - Optional - request timeout
- **headers** (dict) - Optional - additional request headers
- **n_retries** (int) - Optional - maximum number of retries
- **retry_interval** (int) - Optional - delay between retries in seconds
- **retry_on** (tuple) - Optional - additional exception classes to retry on
- **requests_args** (dict) - Optional - additional parameters for RequestsSession
- **httpx_args** (dict) - Optional - additional parameters for HTTPXSession
- **curl_options** (dict) - Optional - additional options for PycURLSession
- **kwargs** (dict) - Optional - any other parameters accepted by Session.send_request()

### Returns
- **SyncResourceLinkObject** - The created resource link object.
```

--------------------------------

### Public Resource Link Initialization and Factory

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Constructor and factory method for public resource links, handling public keys or URLs.

```python
    def __init__(self, link: Optional[Dict] = None, yadisk: Optional[Any] = None):
        LinkObject.__init__(self, None, 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)

        self.import_fields(link)

        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://", "https://")):
                self.public_url = public_key_or_url
            else:
                self.public_key = public_key_or_url

    @classmethod
    def from_public_key(cls, public_key: Optional[str], yadisk: Optional[Any] = None):
        if public_key is None:
            return cls(yadisk=yadisk)

        return cls(
            {"method": "GET",
             "href": f"{settings.BASE_API_URL}/v1/disk/public/resources?{urlencode({'public_key': public_key})}",
             "templated": False},
            yadisk=yadisk)
```

--------------------------------

### Get public settings of a resource

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Retrieves public settings for a resource, requiring the object to be bound to a YaDisk instance.

```python
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 self._yadisk.get_public_settings(str(path), **kwargs)
```

--------------------------------

### Initialize Resource Object

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Initializes a resource object with optional public defaults and a yadisk instance, mapping fields to their respective types.

```python
    def __init__(
        self,
        public_default: Optional[Dict] = None,
        yadisk: Optional[Any] = None
    ) -> None:
        YaDiskObject.__init__(
            self,
            {
                "macros": typed_list(str_or_error),
                "org_id": int_or_error,
                "rights": typed_list(str_or_error)
            },
            yadisk
        )
        self.import_fields(public_default)
```

--------------------------------

### Get resource type asynchronously

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_async_client.html

The get_type method returns the type of the resource as a string, either 'file' or 'dir'.

```python
async def get_type(self, path: str, /, **kwargs) -> str:
        """
            Get resource type.

            :param path: path to the resource
            :param timeout: `float`, `tuple` or `None`, 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
            :param retry_on: `tuple`, additional exception classes to retry on
            :param aiohttp_args: `dict`, additional parameters for :any:`AIOHTTPSession`
            :param httpx_args: `dict`, additional parameters for :any:`AsyncHTTPXSession`
            :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

            :raises PathNotFoundError: resource was not found on Disk
            :raises ForbiddenError: application doesn't have enough rights for this request

            :returns: "file" or "dir"
        """

        return await _get_type(self.get_meta, path, **kwargs)
```

--------------------------------

### AsyncClient Constructor

Source: https://yadisk.readthedocs.io/en/latest/api_reference/async_api.html

Initializes a new instance of the AsyncClient to interact with the Yandex.Disk REST API.

```APIDOC
## class yadisk.AsyncClient

### Description
Initializes the asynchronous client for Yandex.Disk. It is recommended to use the async with statement to ensure connections are closed properly.

### Parameters
- **id** (str) - Optional - Application ID.
- **secret** (str) - Optional - Application secret password.
- **token** (str) - Optional - Application token.
- **default_args** (dict[str, Any]) - Optional - Default arguments for methods (e.g., timeout, headers).
- **session** (AsyncSession | Literal['aiohttp'] | Literal['httpx']) - Optional - HTTP client implementation to use.
- **open_file** (Callable) - Optional - Async function that opens a file for reading or writing.
- **session_factory** (Callable) - Optional - Callable that returns an instance of AsyncSession.
```

--------------------------------

### Initialize PublicSettingsObject

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Configures the object with field types for public settings and imports data from a dictionary.

```python
def __init__(
    self,
    public_settings: Optional[Dict] = None,
    yadisk: Optional[Any] = None
) -> None:
    YaDiskObject.__init__(
        self,
        {
            "available_until": int_or_error,
             "read_only": bool_or_error,
             "available_until_verbose": AvailableUntilVerboseObject,
             "password": str_or_error,
             "password_verbose": PasswordVerboseObject,
             "external_organization_id": str_or_error,
             "external_organization_id_verbose": ExternalOrganizationIdVerboseObject,
             "accesses": typed_list(PublicAccessObject)
        },
        yadisk
    )
    self.import_fields(public_settings)
```

--------------------------------

### AIOHTTPSession implementation details

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/sessions/aiohttp_session.html

Internal implementation of the session initialization, request handling, and closure logic using aiohttp.

```python
        headers.update(kwargs.get("headers") or {})

        kwargs["headers"] = headers

        self._session = aiohttp.ClientSession(*args, **kwargs)

    @property
    def aiohttp_session(self) -> aiohttp.ClientSession:
        return self._session

    async def send_request(
        self,
        method: HTTPMethod,
        url: str,
        *,
        params: Optional[Dict[str, Any]] = None,
        data: Optional[AsyncPayload] = None,
        headers: Optional[Headers] = None,
        **kwargs
    ) -> AsyncResponse:
        converted_kwargs: Dict[str, Any] = {
            "params": params,
            "data": data,
            "headers": headers
        }

        if "timeout" in kwargs:
            converted_kwargs["timeout"] = convert_timeout(kwargs["timeout"])

        if "aiohttp_args" in kwargs:
            converted_kwargs.update(kwargs["aiohttp_args"] or {})

        try:
            return AIOHTTPResponse(await self._session.request(method, url, **converted_kwargs))
        except aiohttp.ClientError as e:
            raise convert_aiohttp_exception(e) from e

    async def close(self) -> None:
        await self._session.close()
```

--------------------------------

### Get resource type

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/objects/_resources.html

Retrieves the type of a resource as either 'file' or 'dir'. Requires the object to be bound to a YaDisk instance.

```python
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 `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
        :param retry_on: `tuple`, additional exception classes to retry on
        :param requests_args: `dict`, additional parameters for :any:`RequestsSession`
        :param httpx_args: `dict`, additional parameters for :any:`HTTPXSession`
        :param curl_options: `dict`, additional options for :any:`PycURLSession`
        :param kwargs: any other parameters, accepted by :any:`Session.send_request()`

        :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 self._yadisk.get_type(str(path), **kwargs)
```

--------------------------------

### Get Confirmation Code URL

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Method to generate the URL for obtaining a confirmation code, which can subsequently be exchanged for an access token.

```python
    def get_code_url(
        self,
        device_id:             Optional[str] = None,
        device_name:           Optional[str] = None,
        redirect_uri:          Optional[str] = None,
        login_hint:            Optional[str] = None,
        scope:                 Optional[str] = None,
        optional_scope:        Optional[str] = None,
        force_confirm:         bool = True,
        state:                 Optional[str] = None,
        code_challenge:        Optional[str] = None,
        code_challenge_method: Optional[Union[Literal["plain"], Literal["S256"]]] = None,
        display:               None = None
    ) -> str:
        """
            Get the URL for the user to get the confirmation code.
            The confirmation code can later be used to get the token.
            This method doesn't send any HTTP requests and merely constructs the URL.
```

--------------------------------

### publish(path: str, /, **kwargs)

Source: https://yadisk.readthedocs.io/en/latest/_modules/yadisk/_client.html

Makes a resource public. Returns a SyncResourceLinkObject containing the link to the resource.

```APIDOC
## publish(path: str, /, **kwargs)

### Description
Make a resource public.

### Parameters
- **path** (str) - Required - Path to the resource to be published
- **allow_address_access** (bool) - Optional - Specifies the request format (with or without personal access settings)
- **public_settings** (PublicSettings) - Optional - Public access settings for the resource
- **fields** (list) - Optional - Keys to be included in the response

### Returns
- **SyncResourceLinkObject** - Link to the resource
```
Tokens:
Raw