from typing import Optional, List
from ._private._cluster_apis import ClusterAPIS
from .document import Document
from .kw_module_enum import KWModuleEnum
from ._private._well_dto import FileType
from ._private.dto_converters._document_dto_converter import DocumentDtoConverter
[docs]
class File:
"""
KA file object.
Presents a KA file.
.. note:: Should not be instantiated directly.
"""
[docs]
def __init__(self,
field_id: str,
well_group_id: Optional[str],
well_id: Optional[str],
custom_workflow_id: Optional[str],
file_id: str,
name: str,
file_type: FileType,
folder_id: Optional[str],
folder_name: Optional[str],
labels: List[str],
cluster_apis: ClusterAPIS,
document_dto_converter: DocumentDtoConverter):
self.__field_id: str = field_id
self.__well_group_id: Optional[str] = well_group_id
self.__well_id: Optional[str] = well_id
self.__custom_workflow_id: Optional[str] = custom_workflow_id
self.__file_id: str = file_id
self.__name: str = name
self.__type: str = str(file_type.value)
self.__folder_id: Optional[str] = folder_id
self.__folder_name: Optional[str] = folder_name
self.__labels: List[str] = labels
self.__cluster_apis: ClusterAPIS = cluster_apis
self.__dto_converter: DocumentDtoConverter = document_dto_converter
@property
def file_id(self) -> str:
""" Gets the id of the KA :class:`File`.
"""
return self.__file_id
@property
def name(self) -> str:
""" Gets the name of the KA :class:`File`.
"""
return self.__name
@property
def well_group_id(self) -> Optional[str]:
"""Gets the well_group_id of the KA :class:`File`.
"""
return self.__well_group_id
@property
def type(self) -> str:
""" Gets the type of the KA :class:`File`.
"""
return self.__type
@property
def folder_id(self) -> Optional[str]:
""" Gets the id of the folder where the KA :class:`File` is located.
"""
return self.__folder_id
@property
def folder_name(self) -> Optional[str]:
""" Gets the name of the folder where the KA :class:`File` is located.
"""
return self.__folder_name
@property
def labels(self) -> List[str]:
""" Gets the list of labels of the KA :class:`File`.
"""
return self.__labels
[docs]
def as_kw_document(self, parent_ipta_name: Optional[str] = None) -> Document:
""" Gets this :class:`File` as a KW :class:`Document` entity.
Parameters
----------
parent_ipta_name:
The name of the parent ipta document if there is one
"""
try:
return Document(self.__field_id, self.__well_group_id, self.__well_id, self.__custom_workflow_id, self.__file_id, self.name, KWModuleEnum(self.type),
parent_ipta_name, self.__labels, self.__cluster_apis, self.__dto_converter)
except ValueError:
raise ValueError(f"File {self.name} is not a valid KW document")
[docs]
def download_content(self, file_location: str) -> None:
""" Downloads the content of the document to the specified location.
Parameters
----------
New file_location:
location where to download the file
"""
self.__cluster_apis.field_api.download_content(self.__file_id, file_location)
[docs]
def rename(self, name: str) -> None:
""" Rename this file
Parameters
----------
New name:
New name of the file
"""
dto = {'name': f'{name}'}
self.__name = self.__cluster_apis.field_api.rename_file(self.__field_id, self.__well_id, self.__file_id, dto)
[docs]
def copy(self, new_name: Optional[str] = None, field_id: Optional[str] = None, well_id: Optional[str] = None, well_group_id: Optional[str] = None, user_task_id: Optional[str] = None) -> str:
""" Copy this file"""
field_id = self.__field_id if field_id is None else field_id
well_id = self.__well_id if well_id is None else well_id
if new_name is None:
new_file_id = self.__cluster_apis.field_api.copy_file(field_id, well_id, well_group_id, user_task_id, self.__file_id).id
else:
new_file_id = self.__cluster_apis.field_api.copy_and_rename_file(field_id, well_id, well_group_id, user_task_id, self.file_id, new_name).id
return new_file_id
[docs]
def add_labels(self, labels: List[str]) -> None:
"""Add labels to this file"""
dto = {"userDefinedLabels": labels}
self.__cluster_apis.field_api.add_labels(self.__field_id, self.__well_id, self.__file_id, "file", dto)
self.__labels.extend(labels)
[docs]
def delete(self) -> None:
"""Delete this file"""
self.__cluster_apis.field_api.delete_file(self.__field_id, self.__well_id, self.file_id)
