FileService

FileService(tm1_rest)

Bases: ObjectService

Parameters:

Name	Type	Description	Default
tm1_rest	RestService		required

Source code in TM1py/Services/FileService.py

def __init__(self, tm1_rest: RestService):
    """

    :param tm1_rest:
    """
    super().__init__(tm1_rest)
    self._rest = tm1_rest
    if verify_version(required_version="12", version=self.version):
        self.version_content_path = "Files"
    else:
        self.version_content_path = "Blobs"

MPU_REQUIRED_VERSION = '12' class-attribute instance-attribute

SUBFOLDER_REQUIRED_VERSION = '12' class-attribute instance-attribute

version_content_path = 'Files' instance-attribute

create(file_name, file_content, multi_part_upload=None, max_mb_per_part=200, max_workers=1, **kwargs)

Create file

Folders in file_name (e.g. folderA/folderB/file.csv) will be created implicitly

Parameters:

Name	Type	Description	Default
file_name	Union[str, Path]	file name in root or path to file	required
file_content	Union[bytes, BytesIO]	file_content as bytes or BytesIO	required
multi_part_upload	bool	boolean use multipart upload or not (only available from TM1 12 onwards) By default, multi_part_upload is used for TM1 v12 and not used for TM1 v11	None
max_mb_per_part	float	max megabyte per part in multipart upload (only available from TM1 12 onwards)	200
max_workers	int	max parallel workers for multipart upload (only available from TM1 12 onwards)	1

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def create(
    self,
    file_name: Union[str, Path],
    file_content: Union[bytes, BytesIO],
    multi_part_upload: bool = None,
    max_mb_per_part: float = 200,
    max_workers: int = 1,
    **kwargs,
):
    """Create file

    Folders in file_name (e.g. folderA/folderB/file.csv) will be created implicitly

    :param file_name: file name in root or path to file
    :param file_content: file_content as bytes or BytesIO
    :param multi_part_upload: boolean use multipart upload or not (only available from TM1 12 onwards)
    By default, multi_part_upload is used for TM1 v12 and not used for TM1 v11
    :param max_mb_per_part: max megabyte per part in multipart upload (only available from TM1 12 onwards)
    :param max_workers: max parallel workers for multipart upload (only available from TM1 12 onwards)
    """
    path = Path(file_name)
    self._check_subfolder_support(path=path, function="FileService.create")
    if multi_part_upload:
        self._check_mpu_support(function="FileService.create")

    # Create folder structure iteratively
    if path.parents:
        folder_path = Path()
        for parent_folder in path.parts[:-1]:
            folder_path = folder_path.joinpath(parent_folder)
            self._create_folder(folder_name=folder_path)

    url = self._construct_content_url(path, exclude_path_end=True, extension="Contents")
    body = {"@odata.type": "#ibm.tm1.api.v1.Document", "ID": path.name, "Name": path.name}
    self._rest.POST(url, json.dumps(body), **kwargs)

    return self._upload_file_content(path, file_content, multi_part_upload, max_mb_per_part, max_workers, **kwargs)

delete(file_name, **kwargs)

Delete file

Parameters:

Name	Type	Description	Default
file_name	Union[str, Path]	file name in root or path to file	required

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def delete(self, file_name: Union[str, Path], **kwargs):
    """Delete file

    :param file_name: file name in root or path to file
    """
    path = Path(file_name)
    self._check_subfolder_support(path=path, function="FileService.delete")

    url = self._construct_content_url(path=path, exclude_path_end=False, extension="")

    return self._rest.DELETE(url, **kwargs)

exists(file_name, **kwargs)

Check if file exists

Parameters:

Name	Type	Description	Default
file_name	Union[str, Path]	file name in root or path to file	required

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def exists(self, file_name: Union[str, Path], **kwargs):
    """Check if file exists

    :param file_name: file name in root or path to file
    """
    url = self._construct_content_url(path=Path(file_name), exclude_path_end=False, extension="")

    return self._exists(url, **kwargs)

get(file_name, **kwargs)

Get file

Parameters:

Name	Type	Description	Default
file_name	str	file name in root or path to file	required

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def get(self, file_name: str, **kwargs) -> bytes:
    """Get file

    :param file_name: file name in root or path to file
    """
    path = Path(file_name)
    self._check_subfolder_support(path=path, function="FileService.get")

    url = self._construct_content_url(path=path, exclude_path_end=False, extension="Content")

    return self._rest.GET(url, **kwargs).content

get_all_names(path='', **kwargs)

return list of blob file names

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	path to folder. When empty searches in root	''

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def get_all_names(self, path: Union[str, Path] = "", **kwargs) -> List[str]:
    """return list of blob file names

    :param path: path to folder. When empty searches in root
    """
    path = Path(path)
    url = self._construct_content_url(path, exclude_path_end=False, extension="Contents")

    response = self._rest.GET(url, **kwargs).content
    return [file["Name"] for file in json.loads(response)["value"]]

get_names(**kwargs)

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def get_names(self, **kwargs) -> bytes:
    warnings.warn(
        "Function get_names will be deprecated. Use get_all_names instead", DeprecationWarning, stacklevel=2
    )

    url = format_url(
        "/Contents('{version_content_path}')/Contents?$select=Name", version_content_path=self.version_content_path
    )

    return self._rest.GET(url, **kwargs).content

search_string_in_name(name_startswith=None, name_contains=None, name_contains_operator='and', path='', **kwargs)

Return list of blob files that match search critera

Parameters:

Name	Type	Description	Default
name_startswith	str	str, file name begins with (case insensitive)	None
name_contains	Iterable	iterable, found anywhere in name (case insensitive)	None
name_contains_operator	str	'and' or 'or'	'and'
path	Union[Path, str]	search in given path or root	''

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def search_string_in_name(
    self,
    name_startswith: str = None,
    name_contains: Iterable = None,
    name_contains_operator: str = "and",
    path: Union[Path, str] = "",
    **kwargs,
) -> List[str]:
    """Return list of blob files that match search critera

    :param name_startswith: str, file name begins with (case insensitive)
    :param name_contains: iterable, found anywhere in name (case insensitive)
    :param name_contains_operator: 'and' or 'or'
    :param path: search in given path or root
    """

    name_contains_operator = name_contains_operator.strip().lower()
    if name_contains_operator not in ("and", "or"):
        raise ValueError("'name_contains_operator' must be either 'AND' or 'OR'")

    name_filters = []

    if name_startswith:
        name_filters.append(format_url("startswith(tolower(Name),tolower('{}'))", name_startswith))

    if name_contains:
        if isinstance(name_contains, str):
            name_filters.append(format_url("contains(tolower(Name),tolower('{}'))", name_contains))

        elif isinstance(name_contains, Iterable):
            name_contains_filters = [
                format_url("contains(tolower(Name),tolower('{}'))", wildcard) for wildcard in name_contains
            ]
            name_filters.append("({})".format(f" {name_contains_operator} ".join(name_contains_filters)))

        else:
            raise ValueError("'name_contains' must be str or iterable")

    path = Path(path)
    self._check_subfolder_support(path=path, function="FileService.search_string_in_name")

    url = self._construct_content_url(
        path=Path(path),
        exclude_path_end=False,
        extension="Contents?$select=Name&$filter={}".format(" and ".join(name_filters)),
    )
    response = self._rest.GET(url, **kwargs).content

    return list(file["Name"] for file in json.loads(response)["value"])

update(file_name, file_content, multi_part_upload=None, max_mb_per_part=200, max_workers=1, **kwargs)

Update existing file

Parameters:

Name	Type	Description	Default
file_name	Union[str, Path]	file name in root or path to file	required
file_content	Union[bytes, BytesIO]	file_content as bytes or BytesIO	required
multi_part_upload	bool	boolean use multipart upload or not (only available from TM1 12 onwards) By default, multi_part_upload is used for TM1 v12 and not used for TM1 v11	None
max_mb_per_part	float	max megabyte per part in multipart upload (only available from TM1 12 onwards)	200
max_workers	int	max parallel workers for multipart upload (only available from TM1 12 onwards)	1

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def update(
    self,
    file_name: Union[str, Path],
    file_content: Union[bytes, BytesIO],
    multi_part_upload: bool = None,
    max_mb_per_part: float = 200,
    max_workers: int = 1,
    **kwargs,
):
    """Update existing file

    :param file_name: file name in root or path to file
    :param file_content: file_content as bytes or BytesIO
    :param multi_part_upload: boolean use multipart upload or not (only available from TM1 12 onwards)
    By default, multi_part_upload is used for TM1 v12 and not used for TM1 v11
    :param max_mb_per_part: max megabyte per part in multipart upload (only available from TM1 12 onwards)
    :param max_workers: max parallel workers for multipart upload (only available from TM1 12 onwards)
    """
    path = Path(file_name)
    self._check_subfolder_support(path=path, function="FileService.update")
    if multi_part_upload:
        self._check_mpu_support(function="FileService.create")

    return self._upload_file_content(path, file_content, multi_part_upload, max_mb_per_part, max_workers, **kwargs)

update_or_create(file_name, file_content, multi_part_upload=None, max_mb_per_part=200, max_workers=1, **kwargs)

Create file or update file if it already exists

Parameters:

Name	Type	Description	Default
file_name	Union[str, Path]	file name in root or path to file	required
file_content	bytes	file_content as bytes or BytesIO	required
multi_part_upload	bool	boolean use multipart upload or not (only available from TM1 12 onwards). By default, multi_part_upload is used for TM1 v12 and not used for TM1 v11	None
max_mb_per_part	float	max megabyte per part in multipart upload (only available from TM1 12 onwards)	200
max_workers	int	max parallel workers for multipart upload (only available from TM1 12 onwards)	1

Source code in TM1py/Services/FileService.py

@require_version(version="11.4")
def update_or_create(
    self,
    file_name: Union[str, Path],
    file_content: bytes,
    multi_part_upload: bool = None,
    max_mb_per_part: float = 200,
    max_workers: int = 1,
    **kwargs,
):
    """Create file or update file if it already exists

    :param file_name: file name in root or path to file
    :param file_content: file_content as bytes or BytesIO
    :param multi_part_upload: boolean use multipart upload or not (only available from TM1 12 onwards).
    By default, multi_part_upload is used for TM1 v12 and not used for TM1 v11
    :param max_mb_per_part: max megabyte per part in multipart upload (only available from TM1 12 onwards)
    :param max_workers: max parallel workers for multipart upload (only available from TM1 12 onwards)
    """
    if self.exists(file_name, **kwargs):
        return self.update(file_name, file_content, multi_part_upload, max_mb_per_part, max_workers, **kwargs)

    return self.create(file_name, file_content, multi_part_upload, max_mb_per_part, max_workers, **kwargs)