cirro.models.file

  1from dataclasses import dataclass
  2from pathlib import PurePath
  3from typing import TypedDict, Dict, Optional
  4
  5from cirro_api_client.v1.models import FileAccessRequest, AccessType, FileEntry
  6
  7from cirro.models.s3_path import S3Path
  8
  9
 10class DirectoryStatistics(TypedDict):
 11    size: float
 12    sizeFriendly: str
 13    numberOfFiles: int
 14
 15
 16class FileAccessContext:
 17    """
 18    Context holder for accessing various files in Cirro and abstracting out their location.
 19    Prefer to use the class methods to instantiate.
 20    """
 21    def __init__(self,
 22                 file_access_request: FileAccessRequest,
 23                 project_id: str,
 24                 base_url: str):
 25        self.file_access_request = file_access_request
 26        self.base_url = base_url
 27        self.project_id = project_id
 28        self._s3_path = S3Path(base_url)
 29
 30    @classmethod
 31    def download(cls, project_id: str, base_url: str, token_lifetime_override: int = None):
 32        return cls(
 33            file_access_request=FileAccessRequest(
 34                access_type=AccessType.PROJECT_DOWNLOAD,
 35                token_lifetime_hours=token_lifetime_override
 36            ),
 37            base_url=base_url,
 38            project_id=project_id
 39        )
 40
 41    @classmethod
 42    def upload_dataset(cls, project_id: str, dataset_id: str, base_url: str, token_lifetime_override: int = None):
 43        return cls(
 44            file_access_request=FileAccessRequest(
 45                access_type=AccessType.DATASET_UPLOAD,
 46                dataset_id=dataset_id,
 47                token_lifetime_hours=token_lifetime_override
 48            ),
 49            base_url=f'{base_url}/data',
 50            project_id=project_id
 51        )
 52
 53    @classmethod
 54    def upload_reference(cls, project_id: str, base_url: str):
 55        return cls(
 56            file_access_request=FileAccessRequest(
 57                access_type=AccessType.REFERENCE_UPLOAD
 58            ),
 59            base_url=base_url,
 60            project_id=project_id
 61        )
 62
 63    @classmethod
 64    def upload_sample_sheet(cls, project_id: str, dataset_id: str, base_url: str):
 65        return cls(
 66            file_access_request=FileAccessRequest(
 67                access_type=AccessType.SAMPLESHEET_UPLOAD,
 68                dataset_id=dataset_id
 69            ),
 70            base_url=base_url,
 71            project_id=project_id
 72        )
 73
 74    @property
 75    def bucket(self) -> str:
 76        """ S3 Bucket """
 77        return self._s3_path.bucket
 78
 79    @property
 80    def prefix(self) -> str:
 81        """ S3 Prefix """
 82        return self._s3_path.key
 83
 84
 85@dataclass(frozen=True)
 86class File:
 87    relative_path: str
 88    size: int
 89    access_context: FileAccessContext
 90    metadata: Optional[Dict] = None
 91
 92    @classmethod
 93    def from_file_entry(cls, file: FileEntry, project_id: str, domain: str = None):
 94        # Path is absolute rather than relative
 95        if 's3://' in file.path:
 96            parts = S3Path(file.path)
 97            domain = parts.base
 98            path = parts.key
 99        else:
100            path = file.path
101
102        return cls(
103            relative_path=path,
104            metadata=file.metadata.additional_properties,
105            size=file.size,
106            access_context=FileAccessContext.download(project_id=project_id, base_url=domain)
107        )
108
109    @property
110    def absolute_path(self):
111        return f'{self.access_context.base_url}/{self.relative_path.strip("/")}'
112
113    @property
114    def name(self):
115        return PurePath(self.absolute_path).name
116
117    def __repr__(self):
118        return f'{self.__class__.__name__}(path={self.relative_path})'