Source code for supervisely.api.file_api

# coding: utf-8
"""download/upload/manipulate files from/to Supervisely team files"""

from __future__ import annotations

import mimetypes
import os
import re
import shutil
import tarfile
import urllib
from pathlib import Path
from typing import Callable, Dict, List, NamedTuple, Optional, Union

from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
from tqdm import tqdm
from typing_extensions import Literal

import supervisely.io.env as env
import supervisely.io.fs as sly_fs
from supervisely._utils import batched, is_development, rand_str
from supervisely.api.module_api import ApiField, ModuleApiBase
from supervisely.imaging.image import get_hash, write_bytes
from supervisely.io.fs import (
    ensure_base_path,
    get_file_ext,
    get_file_hash,
    get_file_name,
    get_file_name_with_ext,
    get_file_size,
    list_files_recursively,
    silent_remove,
)
from supervisely.io.fs_cache import FileCache
from supervisely.sly_logger import logger
from supervisely.task.progress import Progress, handle_original_tqdm, tqdm_sly


class FileInfo(NamedTuple):
    """ """

    team_id: int
    id: int
    user_id: int
    name: str
    hash: str
    path: str
    storage_path: str
    mime: str
    ext: str
    sizeb: int
    created_at: str
    updated_at: str
    full_storage_url: str
    is_dir: bool


[docs]class FileApi(ModuleApiBase):
    """
    API for working with Files. :class:`FileApi<FileApi>` object is immutable.

    :param api: API connection to the server.
    :type api: Api
    :Usage example:

     .. code-block:: python

        import os
        from dotenv import load_dotenv

        import supervisely as sly

        # Load secrets and create API object from .env file (recommended)
        # Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
        if sly.is_development():
            load_dotenv(os.path.expanduser("~/supervisely.env"))
        api = sly.Api.from_env()

        # Pass values into the API constructor (optional, not recommended)
        # api = sly.Api(server_address="https://app.supervise.ly", token="4r47N...xaTatb")

        team_id = 8
        file_path = "/999_App_Test/"
        files = api.file.list(team_id, file_path)
    """

[docs]    @staticmethod
    def info_sequence():
        """
        NamedTuple FileInfo information about File.

        :Example:

         .. code-block:: python

            FileInfo(team_id=8,
                     id=7660,
                     user_id=7,
                     name='00135.json',
                     hash='z7Hv9a7WIC5HIJrfX/69KVrvtDaLqucSprWHoCxyq0M=',
                     path='/999_App_Test/ds1/00135.json',
                     storage_path='/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/y/P/rn/...json',
                     mime='application/json',
                     ext='json',
                     sizeb=261,
                     created_at='2021-01-11T09:04:17.959Z',
                     updated_at='2021-01-11T09:04:17.959Z',
                     full_storage_url='http://supervise.ly/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/y/P/rn/...json')
        """
        return [
            ApiField.TEAM_ID,
            ApiField.ID,
            ApiField.USER_ID,
            ApiField.NAME,
            ApiField.HASH,
            ApiField.PATH,
            ApiField.STORAGE_PATH,
            ApiField.MIME2,
            ApiField.EXT2,
            ApiField.SIZEB2,
            ApiField.CREATED_AT,
            ApiField.UPDATED_AT,
            ApiField.FULL_STORAGE_URL,
            ApiField.IS_DIR,
        ]

[docs]    @staticmethod
    def info_tuple_name():
        """
        NamedTuple name - **FileInfo**.
        """
        return "FileInfo"

    def list_on_agent(
        self,
        team_id: int,
        path: str,
        recursive: bool = True,
        return_type: Literal["dict", "fileinfo"] = "dict",
    ) -> List[Union[Dict, FileInfo]]:
        if self.is_on_agent(path) is False:
            raise ValueError(f"Data is not on agent: {path}")

        agent_id, path_in_agent_folder = self.parse_agent_id_and_path(path)
        dirs_queue: List[str] = [path_in_agent_folder]

        results = []
        while len(dirs_queue) > 0:
            cur_dir = dirs_queue.pop(0)
            if cur_dir.endswith("/") is False:
                cur_dir += "/"
            response = self._api.post(
                "agents.storage.list",
                {ApiField.ID: agent_id, ApiField.TEAM_ID: team_id, ApiField.PATH: cur_dir},
            )
            items = response.json()
            for item in items:
                if item["type"] == "file":
                    results.append(item)
                elif item["type"] == "directory" and recursive is True:
                    dirs_queue.append(os.path.join(cur_dir, item["name"]))

        if return_type == "dict":
            return results
        elif return_type == "fileinfo":
            return [self._convert_json_info(info_json) for info_json in results]
        else:
            raise ValueError(
                "The specified value for the 'return_type' parameter should be either 'dict' or 'fileinfo'."
            )

[docs]    def list(
        self,
        team_id: int,
        path: str,
        recursive: bool = True,
        return_type: Literal["dict", "fileinfo"] = "dict",
    ) -> List[Union[Dict, FileInfo]]:
        """
        List of files in the Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param path: Path to File or Directory.
        :type path: str
        :param recursive: If True return all files recursively.
        :type recursive: bool
        :param return_type: The specified value between 'dict' or 'fileinfo'. By default: 'dict'.
        :type return_type: str
        :return: List of all Files with information. See classes info_sequence and FileInfo
        :rtype: class List[Union[Dict, FileInfo]]
        :Usage example:

         .. code-block:: python

            import os
            from dotenv import load_dotenv

            import supervisely as sly

            # Load secrets and create API object from .env file (recommended)
            # Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
            if sly.is_development():
                load_dotenv(os.path.expanduser("~/supervisely.env"))
            api = sly.Api.from_env()

            # Pass values into the API constructor (optional, not recommended)
            # api = sly.Api(server_address="https://app.supervise.ly", token="4r47N...xaTatb")

            team_id = 8
            file_path = "/999_App_Test/"

            # Get information about file in dict way..
            files = api.file.list(team_id, file_path)
            file = files[0]
            print(file['id'])
            # Output: 7660

            print(files)
            # Output: [
            #     {
            #         "id":7660,
            #         "userId":7,
            #         "path":"/999_App_Test/ds1/00135.json",
            #         "storagePath":"/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/y/P/rn/...json",
            #         "meta":{
            #             "ext":"json",
            #             "mime":"application/json",
            #             "size":261
            #         },
            #         "createdAt":"2021-01-11T09:04:17.959Z",
            #         "updatedAt":"2021-01-11T09:04:17.959Z",
            #         "hash":"z7Wv1a7WIC5HIJrfX/69XXrqtDaLxucSprWHoCxyq0M=",
            #         "fullStorageUrl":"http://supervise.ly/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/y/P/rn/...json",
            #         "teamId":8,
            #         "name":"00135.json"
            #     },
            #     {
            #         "id":7661,
            #         "userId":7,
            #         "path":"/999_App_Test/ds1/01587.json",
            #         "storagePath":"/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/9/k/Hs/...json",
            #         "meta":{
            #             "ext":"json",
            #             "mime":"application/json",
            #             "size":252
            #         },
            #         "createdAt":"2021-01-11T09:04:18.099Z",
            #         "updatedAt":"2021-01-11T09:04:18.099Z",
            #         "hash":"La9+XtF2+cTlAqUE/I72e/xS12LqyH1+z<3T+SgD4CTU=",
            #         "fullStorageUrl":"http://supervise.ly/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/9/k/Hs/...json",
            #         "teamId":8,
            #         "name":"01587.json"
            #     }
            # ]

            # ..or as FileInfo with attributes:
            files = api.file.list(team_id, file_path, return_type='fileinfo')
            file = files[0]
            print(file.id)
            # Output: 7660

            print(files)
            # Output: [
            # FileInfo(team_id=8, id=7660, user_id=7, name='00135.json', hash='z7Wv1a7WI...
            # FileInfo(team_id=8, id=7661, user_id=7, name='01587.json', hash='La9+XtF2+...
            # ]
        """

        if not path.endswith("/") and recursive is False:
            path += "/"
        if self.is_on_agent(path) is True:
            return self.list_on_agent(team_id, path, recursive, return_type)

        response = self._api.post(
            "file-storage.list",
            {ApiField.TEAM_ID: team_id, ApiField.PATH: path, ApiField.RECURSIVE: recursive},
        )

        if return_type == "dict":
            return response.json()
        elif return_type == "fileinfo":
            return [self._convert_json_info(info_json) for info_json in response.json()]
        else:
            raise ValueError(
                "The specified value for the 'return_type' parameter should be either 'dict' or 'fileinfo'."
            )

[docs]    def list2(self, team_id: int, path: str, recursive: bool = True) -> List[FileInfo]:
        """
        Disclaimer: Method is not recommended. Use api.file.list instead

        List of files in the Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param path: Path to File or Directory.
        :type path: str
        :param recursive: If True return all FileInfos recursively.
        :type recursive: bool
        :return: List of all Files with information. See class info_sequence
        :rtype: class List[FileInfo]
        :Usage example:

         .. code-block:: python

            import os
            from dotenv import load_dotenv

            import supervisely as sly

            # Load secrets and create API object from .env file (recommended)
            # Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
            if sly.is_development():
                load_dotenv(os.path.expanduser("~/supervisely.env"))
            api = sly.Api.from_env()

            # Pass values into the API constructor (optional, not recommended)
            # api = sly.Api(server_address="https://app.supervise.ly", token="4r47N...xaTatb")

            team_id = 9
            file_path = "/My_App_Test/"
            files = api.file.list2(team_id, file_path)

            print(files)
            # Output: [
            # FileInfo(team_id=9, id=18421, user_id=8, name='5071_3734_mot_video_002.tar.gz', hash='+0nrNoDjBxxJA...
            # FileInfo(team_id=9, id=18456, user_id=8, name='5164_4218_mot_video_bitmap.tar.gz', hash='fwtVI+iptY...
            # FileInfo(team_id=9, id=18453, user_id=8, name='all_vars.tar', hash='TVkUE+K1bnEb9QrdEm9akmHm/QEWPJK...
            # ]
        """
        return self.list(team_id=team_id, path=path, recursive=recursive, return_type="fileinfo")

[docs]    def listdir(self, team_id: int, path: str, recursive: bool = False) -> List[str]:
        """
        List dirs and files in the directiry with given path.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param path: Path to directory.
        :type path: str
        :param recursive: If True return all paths recursively.
        :type recursive: bool
        :return: List of paths
        :rtype: :class:`List[str]`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            team_id = 8
            path = "/999_App_Test/"
            files = api.file.listdir(team_id, path)

            print(files)
            # Output: ["/999_App_Test/ds1", "/999_App_Test/image.png"]
        """
        files = self.list(team_id, path, recursive)
        files_paths = [file["path"] for file in files]
        return files_paths

[docs]    def get_directory_size(self, team_id: int, path: str) -> int:
        """
        Get directory size in the Team Files.
        If directory is on local agent, then optimized method will be used (without api requests)

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param path: Path to Directory.
        :type path: str
        :return: Directory size in the Team Files
        :rtype: :class:`int`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            team_id = 9
            path = "/My_App_Test/"
            size = api.file.get_directory_size(team_id, path)

            print(size)
            # Output: 3478687
        """
        if self.is_on_agent(path) is True:
            agent_id, path_in_agent_folder = self.parse_agent_id_and_path(path)
            if (
                agent_id == env.agent_id(raise_not_found=False)
                and env.agent_storage(raise_not_found=False) is not None
            ):
                path_on_agent = os.path.normpath(env.agent_storage() + path_in_agent_folder)
                logger.info(f"Optimized getting directory size from agent: {path_on_agent}")
                dir_size = sly_fs.get_directory_size(path_on_agent)
                return dir_size

        dir_size = 0
        file_infos = self.list(team_id=team_id, path=path, recursive=True, return_type="fileinfo")
        for file_info in file_infos:
            dir_size += file_info.sizeb
        return dir_size

    def _download(
        self, team_id, remote_path, local_save_path, progress_cb=None
    ):  # TODO: progress bar
        response = self._api.post(
            "file-storage.download",
            {ApiField.TEAM_ID: team_id, ApiField.PATH: remote_path},
            stream=True,
        )
        # print(response.headers)
        # print(response.headers['Content-Length'])
        ensure_base_path(local_save_path)
        with open(local_save_path, "wb") as fd:
            for chunk in response.iter_content(chunk_size=1024 * 1024):
                fd.write(chunk)
                if progress_cb is not None:
                    progress_cb(len(chunk))

[docs]    def download(
        self,
        team_id: int,
        remote_path: str,
        local_save_path: str,
        cache: Optional[FileCache] = None,
        progress_cb: Optional[Union[tqdm, Callable]] = None,
    ) -> None:
        """
        Download File from Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param remote_path: Path to File in Team Files.
        :type remote_path: str
        :param local_save_path: Local save path.
        :type local_save_path: str
        :param cache: optional
        :type cache: FileCache, optional
        :param progress_cb: Function for tracking download progress.
        :type progress_cb: tqdm or callable, optional
        :return: None
        :rtype: :class:`NoneType`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            path_to_file = "/999_App_Test/ds1/01587.json"
            local_save_path = "/home/admin/Downloads/01587.json"

            api.file.download(8, path_to_file, local_save_path)
        """

        if self.is_on_agent(remote_path):
            self.download_from_agent(remote_path, local_save_path, progress_cb)
            return

        if cache is None:
            self._download(team_id, remote_path, local_save_path, progress_cb)
        else:
            file_info = self.get_info_by_path(team_id, remote_path)
            if file_info.hash is None:
                self._download(team_id, remote_path, local_save_path, progress_cb)
            else:
                cache_path = cache.check_storage_object(file_info.hash, get_file_ext(remote_path))
                if cache_path is None:
                    # file not in cache
                    self._download(team_id, remote_path, local_save_path, progress_cb)
                    if file_info.hash != get_file_hash(local_save_path):
                        raise KeyError(
                            f"Remote and local hashes are different (team id: {team_id}, file: {remote_path})"
                        )
                    cache.write_object(local_save_path, file_info.hash)
                else:
                    cache.read_object(file_info.hash, local_save_path)
                    if progress_cb is not None:
                        progress_cb(get_file_size(local_save_path))

    def is_on_agent(self, remote_path: str):
        return sly_fs.is_on_agent(remote_path)

    def parse_agent_id_and_path(self, remote_path: str) -> int:
        return sly_fs.parse_agent_id_and_path(remote_path)

    def download_from_agent(
        self,
        remote_path: str,
        local_save_path: str,
        progress_cb: Optional[Union[tqdm, Callable]] = None,
    ) -> None:
        agent_id, path_in_agent_folder = self.parse_agent_id_and_path(remote_path)
        if (
            agent_id == env.agent_id(raise_not_found=False)
            and env.agent_storage(raise_not_found=False) is not None
        ):
            path_on_agent = os.path.normpath(env.agent_storage() + path_in_agent_folder)
            logger.info(f"Optimized download from agent: {path_on_agent}")
            sly_fs.copy_file(path_on_agent, local_save_path)
            if progress_cb is not None:
                progress_cb(sly_fs.get_file_size(path_on_agent))
            return

        response = self._api.post(
            "agents.storage.download",
            {ApiField.ID: agent_id, ApiField.PATH: path_in_agent_folder},
            stream=True,
        )
        ensure_base_path(local_save_path)
        with open(local_save_path, "wb") as fd:
            for chunk in response.iter_content(chunk_size=1024 * 1024):
                fd.write(chunk)
                if progress_cb is not None:
                    progress_cb(len(chunk))

[docs]    def download_directory(
        self,
        team_id: int,
        remote_path: str,
        local_save_path: str,
        progress_cb: Optional[Union[tqdm, Callable]] = None,
    ) -> None:
        """
        Download Directory from Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param remote_path: Path to Directory in Team Files.
        :type remote_path: str
        :param local_save_path: Local save path.
        :type local_save_path: str
        :param progress_cb: Function for tracking download progress.
        :type progress_cb: tqdm or callable, optional
        :return: None
        :rtype: :class:`NoneType`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            path_to_dir = "/My_App_Test/ds1"
            local_save_path = "/home/admin/Downloads/My_local_test"

            api.file.download_directory(9, path_to_dir, local_save_path)
        """
        if not remote_path.endswith("/"):
            remote_path += "/"

        if self.is_on_agent(remote_path) is True:
            agent_id, path_in_agent_folder = self.parse_agent_id_and_path(remote_path)
            if (
                agent_id == env.agent_id(raise_not_found=False)
                and env.agent_storage(raise_not_found=False) is not None
            ):
                dir_on_agent = os.path.normpath(env.agent_storage() + path_in_agent_folder)
                logger.info(f"Optimized download from agent: {dir_on_agent}")
                sly_fs.copy_dir_recursively(dir_on_agent, local_save_path)
                if progress_cb is not None:
                    progress_cb(sly_fs.get_directory_size(dir_on_agent))
                return

        local_temp_archive = os.path.join(local_save_path, "temp.tar")
        self.download(team_id, remote_path, local_temp_archive, cache=None, progress_cb=progress_cb)
        tr = tarfile.open(local_temp_archive)
        tr.extractall(local_save_path)
        silent_remove(local_temp_archive)
        temp_dir = os.path.join(local_save_path, rand_str(10))
        to_move_dir = os.path.join(local_save_path, os.path.basename(os.path.normpath(remote_path)))
        os.rename(to_move_dir, temp_dir)
        file_names = os.listdir(temp_dir)
        for file_name in file_names:
            shutil.move(os.path.join(temp_dir, file_name), local_save_path)
        shutil.rmtree(temp_dir)

[docs]    def download_input(
        self,
        save_path: str,
        unpack_if_archive: Optional[bool] = True,
        remove_archive: Optional[bool] = True,
        force: Optional[bool] = False,
        log_progress: Optional[bool] = False,
    ) -> None:
        """Downloads data for application from input using environment variables.
        Automatically detects is data is a file or a directory and saves it to the specified directory.
        If data is an archive, it will be unpacked to the specified directory if unpack_if_archive is True.

        :param save_path: path to a directory where data will be saved
        :type save_path: str
        :param unpack_if_archive: if True, archive will be unpacked to the specified directory
        :type unpack_if_archive: Optional[bool]
        :param remove_archive: if True, archive will be removed after unpacking
        :type remove_archive: Optional[bool]
        :param force: if True, data will be downloaded even if it already exists in the specified directory
        :type force: Optional[bool]
        :param log_progress: if True, progress bar will be displayed
        :type log_progress: Optional[bool]
        :raises RuntimeError: if both file and folder paths not found in environment variables
        :raises RuntimeError: if both file and folder paths found in environment variables (debug)
        :raises RuntimeError: if team id not found in environment variables
        :Usage example:

         .. code-block:: python

            import os
            from dotenv import load_dotenv

            import supervisely as sly

            # Load secrets and create API object from .env file (recommended)
            # Learn more here: https://developer.supervisely.com/getting-started/basics-of-authentication
            load_dotenv(os.path.expanduser("~/supervisely.env"))
            api = sly.Api.from_env()

            # Application is started...
            save_path = "/my_app_data"
            api.file.download_input(save_path)

            # The data is downloaded to the specified directory.
        """
        remote_file_path = env.file(raise_not_found=False)
        remote_folder_path = env.folder(raise_not_found=False)
        team_id = env.team_id()

        sly_fs.mkdir(save_path)

        if remote_file_path is None and remote_folder_path is None:
            raise RuntimeError(
                "Both file and folder paths not found in environment variables. "
                "Please, specify one of them."
            )
        elif remote_file_path is not None and remote_folder_path is not None:
            raise RuntimeError(
                "Both file and folder paths found in environment variables. "
                "Please, specify only one of them."
            )
        if team_id is None:
            raise RuntimeError("Team id not found in environment variables.")

        if remote_file_path is not None:
            file_name = sly_fs.get_file_name_with_ext(remote_file_path)
            local_file_path = os.path.join(save_path, file_name)

            if os.path.isfile(local_file_path) and not force:
                logger.info(
                    f"The file {local_file_path} already exists. "
                    "Download is skipped, if you want to download it again, "
                    "use force=True."
                )
                return

            sly_fs.silent_remove(local_file_path)

            progress_cb = None
            file_info = self.get_info_by_path(team_id, remote_file_path)
            if log_progress is True and file_info is not None:
                progress = Progress(
                    f"Downloading {remote_file_path}", file_info.sizeb, is_size=True
                )
                progress_cb = progress.iters_done_report
            if self.is_on_agent(remote_file_path):
                self.download_from_agent(remote_file_path, local_file_path, progress_cb=progress_cb)
            else:
                self.download(team_id, remote_file_path, local_file_path, progress_cb=progress_cb)
            if unpack_if_archive and sly_fs.is_archive(local_file_path):
                sly_fs.unpack_archive(local_file_path, save_path)
                if remove_archive:
                    sly_fs.silent_remove(local_file_path)
                else:
                    logger.info(
                        f"Achive {local_file_path} was unpacked, but not removed. "
                        "If you want to remove it, use remove_archive=True."
                    )
        elif remote_folder_path is not None:
            folder_name = os.path.basename(os.path.normpath(remote_folder_path))
            local_folder_path = os.path.join(save_path, folder_name)
            if os.path.isdir(local_folder_path) and not force:
                logger.info(
                    f"The folder {folder_name} already exists. "
                    "Download is skipped, if you want to download it again, "
                    "use force=True."
                )
                return

            sly_fs.remove_dir(local_folder_path)

            progress_cb = None
            if log_progress is True:
                sizeb = self.get_directory_size(team_id, remote_folder_path)
                progress = Progress(f"Downloading: {remote_folder_path}", sizeb, is_size=True)
                progress_cb = progress.iters_done_report
            self.download_directory(
                team_id, remote_folder_path, local_folder_path, progress_cb=progress_cb
            )

    def _upload_legacy(self, team_id, src, dst):
        """ """

        def path_to_bytes_stream(path):
            return open(path, "rb")

        item = get_file_name_with_ext(dst)
        content_dict = {}
        content_dict[ApiField.NAME] = item

        dst_dir = os.path.dirname(dst)
        if not dst_dir.endswith("/"):
            dst_dir += "/"
        content_dict[ApiField.PATH] = dst_dir  # os.path.basedir ...
        content_dict["file"] = (
            item,
            path_to_bytes_stream(src),
            mimetypes.MimeTypes().guess_type(src)[0],
        )
        encoder = MultipartEncoder(fields=content_dict)
        resp = self._api.post("file-storage.upload?teamId={}".format(team_id), encoder)
        return resp.json()

[docs]    def upload(
        self, team_id: int, src: str, dst: str, progress_cb: Optional[Union[tqdm, Callable]] = None
    ) -> FileInfo:
        """
        Upload File to Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param src: Local source file path.
        :type src: str
        :param dst: Path to File in Team Files.
        :type dst: str
        :param progress_cb: Function for tracking download progress.
        :type progress_cb: tqdm or callable, optional
        :return: Information about File. See :class:`info_sequence<info_sequence>`
        :rtype: :class:`FileInfo`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            src_path = "/home/admin/Downloads/01587.json"
            dst_remote_path = "/999_App_Test/ds1/01587.json"

            api.file.upload(8, src_path, dst_remote_path)
        """
        return self.upload_bulk(team_id, [src], [dst], progress_cb)[0]

[docs]    def upload_bulk(
        self,
        team_id: int,
        src_paths: List[str],
        dst_paths: List[str],
        progress_cb: Optional[Union[tqdm, Callable]] = None,
    ) -> List[FileInfo]:
        """
        Upload Files to Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param src: Local source file paths.
        :type src: List[str]
        :param dst: Destination paths for Files to Team Files.
        :type dst: List[str]
        :param progress_cb: Function for tracking download progress.
        :type progress_cb: tqdm or callable, optional
        :return: Information about Files. See :class:`info_sequence<info_sequence>`
        :rtype: :class:`List[FileInfo]`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            src_paths = ["/home/admin/Downloads/01587.json", "/home/admin/Downloads/01588.json","/home/admin/Downloads/01589.json"]
            dst_remote_paths = ["/999_App_Test/ds1/01587.json", "/999_App_Test/ds1/01588.json", "/999_App_Test/ds1/01589.json"]

            api.file.upload_bulk(8, src_paths, dst_remote_paths)
        """

        def path_to_bytes_stream(path):
            return open(path, "rb")

        content_dict = []
        for idx, (src, dst) in enumerate(zip(src_paths, dst_paths)):
            name = get_file_name_with_ext(dst)
            content_dict.append((ApiField.NAME, name))
            dst_dir = os.path.dirname(dst)
            if not dst_dir.endswith("/"):
                dst_dir += "/"
            content_dict.append((ApiField.PATH, dst_dir))
            content_dict.append(
                (
                    "file",
                    (
                        name,
                        path_to_bytes_stream(src),
                        mimetypes.MimeTypes().guess_type(src)[0],
                    ),
                )
            )
        encoder = MultipartEncoder(fields=content_dict)

        # progress = None
        # if progress_logger is not None:
        #     progress = Progress("Uploading", encoder.len, progress_logger, is_size=True)
        #
        # def _print_progress(monitor):
        #     if progress is not None:
        #         progress.set_current_value(monitor.bytes_read)
        #         print(monitor.bytes_read, monitor.len)
        # last_percent = 0
        # def _progress_callback(monitor):
        #     cur_percent = int(monitor.bytes_read * 100.0 / monitor.len)
        #     if cur_percent - last_percent > 10 or cur_percent == 100:
        #         api.task.set_fields(task_id, [{"field": "data.previewProgress", "payload": cur_percent}])
        #     last_percent = cur_percent

        if progress_cb is None:
            data = encoder
        else:
            try:
                data = MultipartEncoderMonitor(encoder, progress_cb.get_partial())
            except AttributeError:
                data = MultipartEncoderMonitor(encoder, progress_cb)
        resp = self._api.post("file-storage.bulk.upload?teamId={}".format(team_id), data)
        results = [self._convert_json_info(info_json) for info_json in resp.json()]

        return results

[docs]    def rename(self, old_name: str, new_name: str) -> None:
        """
        Renames file in Team Files

        :param old_name: Old File name.
        :type old_name: str
        :param new_name: New File name.
        :type new_name: str
        :return: None
        :rtype: :class:`NoneType`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            # NotImplementedError('Method is not supported')
        """
        pass

    def remove_from_agent(self, team_id: int, path: str) -> None:
        raise NotImplementedError()
        agent_id, path_in_agent_folder = self.parse_agent_id_and_path(path)
        if (
            agent_id == env.agent_id(raise_not_found=False)
            and env.agent_storage(raise_not_found=False) is not None
        ):
            # path_on_agent = os.path.normpath(env.agent_storage() + path_in_agent_folder)
            # logger.info(f"Optimized download from agent: {path_on_agent}")
            # sly_fs.copy_file(path_on_agent, local_save_path)
            return

[docs]    def remove(self, team_id: int, path: str) -> None:
        """
        Removes a file from the Team Files. If the specified path is a directory,
        the entire directory (including all recursively included files) will be removed.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param path: Path in Team Files.
        :type path: str
        :return: None
        :rtype: :class:`NoneType`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            api.file.remove(8, "/999_App_Test/ds1/01587.json") # remove file
            api.file.remove(8, "/999_App_Test/ds1/") # remove folder
        """

        if self.is_on_agent(path) is True:
            # self.remove_from_agent(team_id, path)
            logger.warn(
                f"Data '{path}' is on agent. Method does not support agent storage. Remove your data manually on the computer with agent."
            )
            return

        resp = self._api.post(
            "file-storage.remove", {ApiField.TEAM_ID: team_id, ApiField.PATH: path}
        )

[docs]    def remove_file(self, team_id: int, path: str) -> None:
        """
        Removes file from Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param path: Path to File in Team Files.
        :type path: str
        :return: None
        :rtype: :class:`NoneType`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            api.file.remove_file(8, "/999_App_Test/ds1/01587.json")
        """

        file_info = self.get_info_by_path(team_id, path)

        if file_info is None:
            raise ValueError(
                f"File not found in Team files. Maybe you entered directory? (Path: '{path}')"
            )

        self.remove(team_id, path)

[docs]    def remove_dir(self, team_id: int, path: str, silent: bool = False) -> None:
        """
        Removes folder from Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param path: Path to folder in Team Files.
        :type path: str
        :param silent: Ignore if directory not exists.
        :type silent: bool

        :return: None
        :rtype: :class:`NoneType`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            api.file.remove_dir(8, "/999_App_Test/ds1/")
        """

        if not path.endswith("/"):
            raise ValueError("Please add a slash in the end to recognize path as a directory.")

        if silent is False:
            if not self.dir_exists(team_id, path):
                raise ValueError(f"Folder not found in Team files. (Path: '{path}')")

        self.remove(team_id, path)

[docs]    def remove_batch(
        self,
        team_id: int,
        paths: List[str],
        progress_cb: Optional[Union[tqdm, Callable]] = None,
    ) -> None:
        """
        Removes list of files from Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param paths: List of paths to Files in Team Files.
        :type paths: List[str]
        :return: None
        :rtype: :class:`NoneType`
        :Usage example:

        .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            paths_to_del = [
                "/999_App_Test/ds1/01587.json",
                "/999_App_Test/ds1/01588.json",
                "/999_App_Test/ds1/01587.json"
            ]
            api.file.remove(8, paths_to_del)
        """

        for paths_batch in batched(paths, batch_size=100):
            for path in paths_batch:
                if self.is_on_agent(path) is True:
                    logger.warn(
                        f"Data '{path}' is on agent. File skipped. Method does not support agent storage. Remove your data manually on the computer with agent."
                    )
                    paths_batch.remove(path)

            self._api.post(
                "file-storage.bulk.remove", {ApiField.TEAM_ID: team_id, ApiField.PATHS: paths_batch}
            )
            if progress_cb is not None:
                progress_cb(len(paths_batch))

[docs]    def exists(self, team_id: int, remote_path: str) -> bool:
        """
        Checks if file exists in Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param remote_path: Remote path to File in Team Files.
        :type remote_path: str
        :return: True if file exists, otherwise False
        :rtype: :class:`bool`
        :Usage example:

         .. code-block:: python

           import supervisely as sly

           os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
           os.environ['API_TOKEN'] = 'Your Supervisely API Token'
           api = sly.Api.from_env()

           file = api.file.exists(8, "/999_App_Test/ds1/02163.json") # True
           file = api.file.exists(8, "/999_App_Test/ds1/01587.json") # False
        """
        path_infos = self.list(team_id, remote_path)
        for info in path_infos:
            if info["path"] == remote_path:
                return True
        return False

[docs]    def dir_exists(self, team_id: int, remote_directory: str) -> bool:
        """
        Checks if directory exists in Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param remote_path: Remote path to directory in Team Files.
        :type remote_path: str
        :return: True if directory exists, otherwise False
        :rtype: :class:`bool`
        :Usage example:

         .. code-block:: python

           import supervisely as sly

           os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
           os.environ['API_TOKEN'] = 'Your Supervisely API Token'
           api = sly.Api.from_env()

           file = api.file.dir_exists(8, "/999_App_Test/")   # True
           file = api.file.dir_exists(8, "/10000_App_Test/") # False
        """
        files_infos = self.list(team_id, remote_directory)
        if len(files_infos) > 0:
            return True
        return False

[docs]    def get_free_name(self, team_id: int, path: str) -> str:
        """
        Adds suffix to the end of the file name.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param path: Remote path to file in Team Files.
        :type path: str
        :return: New File name with suffix at the end
        :rtype: :class:`str`
        :Usage example:

         .. code-block:: python

           import supervisely as sly

           os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
           os.environ['API_TOKEN'] = 'Your Supervisely API Token'
           api = sly.Api.from_env()

           file = api.file.get_free_name(8, "/999_App_Test/ds1/02163.json")
           print(file)
           # Output: /999_App_Test/ds1/02163_000.json
        """
        directory = Path(path).parent
        name = get_file_name(path)
        ext = get_file_ext(path)
        res_name = name
        suffix = 0

        def _combine(suffix: int = None):
            res = "{}/{}".format(directory, res_name)
            if suffix is not None:
                res += "_{:03d}".format(suffix)
            if ext:
                res += "{}".format(ext)
            return res

        res_path = _combine()
        while self.exists(team_id, res_path):
            res_path = _combine(suffix)
            suffix += 1
        return res_path

[docs]    def get_url(self, file_id: int) -> str:
        """
        Gets URL for the File by ID.

        :param file_id: File ID in Supervisely.
        :type file_id: int
        :return: File URL
        :rtype: :class:`str`
        :Usage example:

         .. code-block:: python

           import supervisely as sly

           os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
           os.environ['API_TOKEN'] = 'Your Supervisely API Token'
           api = sly.Api.from_env()

           file_id = 7660
           file_url = sly.api.file.get_url(file_id)
           print(file_url)
           # Output: http://supervise.ly/files/7660
        """
        return f"/files/{file_id}"

[docs]    def get_info_by_path(self, team_id: int, remote_path: str) -> FileInfo:
        """
        Gets File information by path in Team Files.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param remote_path: Remote path to file in Team Files.
        :type remote_path: str
        :return: Information about File. See :class:`info_sequence<info_sequence>`
        :rtype: :class:`FileInfo`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            file_path = "/999_App_Test/ds1/00135.json"
            file_info = api.file.get_info_by_id(8, file_path)
            print(file_info)
            # Output: FileInfo(team_id=8,
            #                  id=7660,
            #                  user_id=7,
            #                  name='00135.json',
            #                  hash='z7Hv9a7WIC5HIJrfX/69KVrvtDaLqucSprWHoCxyq0M=',
            #                  path='/999_App_Test/ds1/00135.json',
            #                  storage_path='/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/y/P/rn/...json',
            #                  mime='application/json',
            #                  ext='json',
            #                  sizeb=261,
            #                  created_at='2021-01-11T09:04:17.959Z',
            #                  updated_at='2021-01-11T09:04:17.959Z',
            #                  full_storage_url='http://supervise.ly/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/y/P/rn/...json')
        """
        if self.is_on_agent(remote_path) is True:
            path_infos = self.list_on_agent(team_id, os.path.dirname(remote_path), recursive=False)
            for info in path_infos:
                if info["path"] == remote_path:
                    return self._convert_json_info(info)
        else:
            path_infos = self.list(team_id, remote_path)
            for info in path_infos:
                if info["path"] == remote_path:
                    return self._convert_json_info(info)
        return None

    def _convert_json_info(self, info: dict, skip_missing=True) -> FileInfo:
        """ """
        res = super()._convert_json_info(info, skip_missing=skip_missing)
        # if res.storage_path is not None:
        #    res = res._replace(full_storage_url=urllib.parse.urljoin(self._api.server_address, res.storage_path))
        return FileInfo(**res._asdict())

[docs]    def get_info_by_id(self, id: int) -> FileInfo:
        """
        Gets information about File by ID.

        :param id: File ID in Supervisely.
        :type id: int
        :return: Information about File. See :class:`info_sequence<info_sequence>`
        :rtype: :class:`FileInfo`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            file_id = 7660
            file_info = api.file.get_info_by_id(file_id)
            print(file_info)
            # Output: FileInfo(team_id=8,
            #                  id=7660,
            #                  user_id=7,
            #                  name='00135.json',
            #                  hash='z7Hv9a7WIC5HIJrfX/69KVrvtDaLqucSprWHoCxyq0M=',
            #                  path='/999_App_Test/ds1/00135.json',
            #                  storage_path='/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/y/P/rn/...json',
            #                  mime='application/json',
            #                  ext='json',
            #                  sizeb=261,
            #                  created_at='2021-01-11T09:04:17.959Z',
            #                  updated_at='2021-01-11T09:04:17.959Z',
            #                  full_storage_url='http://supervise.ly/h5un6l2bnaz1vj8a9qgms4-public/teams_storage/8/y/P/rn/...json')
        """
        resp = self._api.post("file-storage.info", {ApiField.ID: id})
        return self._convert_json_info(resp.json())

[docs]    def get_free_dir_name(self, team_id: int, dir_path: str) -> str:
        """
        Adds suffix to the end of the Directory name.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param dir_path: Path to Directory in Team Files.
        :type dir_path: str
        :return: New Directory name with suffix at the end
        :rtype: :class:`str`
        :Usage example:

         .. code-block:: python

           import supervisely as sly

           os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
           os.environ['API_TOKEN'] = 'Your Supervisely API Token'
           api = sly.Api.from_env()

           new_dir_name = api.file.get_free_dir_name(9, "/My_App_Test")
           print(new_dir_name)
           # Output: /My_App_Test_001
        """
        res_dir = dir_path.rstrip("/")
        suffix = 1
        while self.dir_exists(team_id, res_dir):
            res_dir = dir_path.rstrip("/") + f"_{suffix:03d}"
            suffix += 1
        return res_dir

[docs]    def upload_directory(
        self,
        team_id: int,
        local_dir: str,
        remote_dir: str,
        change_name_if_conflict: Optional[bool] = True,
        progress_size_cb: Optional[Union[tqdm, Callable]] = None,
        replace_if_conflict: Optional[bool] = False,
    ) -> str:
        """
        Upload Directory to Team Files from local path.

        :param team_id: Team ID in Supervisely.
        :type team_id: int
        :param local_dir: Path to local Directory.
        :type local_dir: str
        :param remote_dir: Path to Directory in Team Files.
        :type remote_dir: str
        :param change_name_if_conflict: Checks if given name already exists and adds suffix to the end of the name.
        :type change_name_if_conflict: bool, optional
        :param progress_size_cb: Function for tracking download progress.
        :type progress_size_cb: Progress, optional
        :return: Path to Directory in Team Files
        :rtype: :class:`str`
        :Usage example:

         .. code-block:: python

            import supervisely as sly

            os.environ['SERVER_ADDRESS'] = 'https://app.supervise.ly'
            os.environ['API_TOKEN'] = 'Your Supervisely API Token'
            api = sly.Api.from_env()

            path_to_dir = "/My_App_Test/ds1"
            local_path = "/home/admin/Downloads/My_local_test"

            api.file.upload_directory(9, local_path, path_to_dir)
        """
        if self.dir_exists(team_id, remote_dir):
            if change_name_if_conflict is True:
                res_remote_dir = self.get_free_dir_name(team_id, remote_dir)
            elif replace_if_conflict is True:
                res_remote_dir = remote_dir
            else:
                raise FileExistsError(
                    f"Directory {remote_dir} already exists in your team (id={team_id})"
                )
        else:
            res_remote_dir = remote_dir

        local_files = list_files_recursively(local_dir)
        remote_files = []
        dir_parts = local_dir.strip("/").split("/")
        for file in local_files:
            path_parts = file.strip("/").split("/")
            path_parts = path_parts[len(dir_parts) :]
            remote_parts = [res_remote_dir.rstrip("/")] + path_parts
            remote_file = "/".join(remote_parts)
            remote_files.append(remote_file)
            

        for local_paths_batch, remote_files_batch in zip(
            batched(local_files, batch_size=50), batched(remote_files, batch_size=50)
        ):
            self.upload_bulk(team_id, local_paths_batch, remote_files_batch, progress_size_cb)
        return res_remote_dir