Skip to content

LargeFile#

from great_ai.large_file import *

For more details about using LargeFiles, check out the how to guide.

LargeFileLocal #

Bases: LargeFileBase

LargeFile implementation using local filesystem as a backend.

Store large files remotely using the familiar API of open(). With built-in versioning, pruning and local cache.

IMPORTANT: If LargeFileLocal.max_cache_size is too small, it won't be enough to store all your files and they can end up deleted.

See parent for more details.

Examples:

>>> LargeFileLocal.cache_path = Path(".cache")

>>> LargeFileLocal.max_cache_size = "30GB"

>>> with LargeFileLocal("my_test.txt", "w", keep_last_n=2) as f:
...     f.write('test')
4

>>> with LargeFileLocal("my_test.txt") as f:
...     print(f.read())
test
Source code in great_ai/large_file/large_file/large_file_local.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class LargeFileLocal(LargeFileBase):
    """LargeFile implementation using local filesystem as a backend.

    Store large files remotely using the familiar API of `open()`. With built-in
    versioning, pruning and local cache.

    IMPORTANT: If LargeFileLocal.max_cache_size is too small, it won't be enough to
    store all your files and they can end up deleted.

    See parent for more details.

    Examples:
        >>> LargeFileLocal.cache_path = Path(".cache")

        >>> LargeFileLocal.max_cache_size = "30GB"

        >>> with LargeFileLocal("my_test.txt", "w", keep_last_n=2) as f:
        ...     f.write('test')
        4

        >>> with LargeFileLocal("my_test.txt") as f:
        ...     print(f.read())
        test
    """

    def __init__(
        self,
        name: str,
        mode: str = "r",
        *,
        buffering: int = -1,
        encoding: Optional[str] = None,
        errors: Optional[str] = None,
        newline: Optional[str] = None,
        version: Optional[int] = None,
        keep_last_n: Optional[int] = None,
    ):
        super().__init__(
            name,
            mode,
            buffering=buffering,
            encoding=encoding,
            errors=errors,
            newline=newline,
            version=version,
            keep_last_n=keep_last_n,
            cache_only_mode=True,
        )
        super().configure_credentials()

    def _find_remote_instances(self) -> List[DataInstance]:
        return []

    def _get_cache_namespace(self) -> str:
        # Preserve existing local model stores; remote caches are namespaced.
        return ""

    def _download(
        self, remote_path: Any, local_path: Path, hide_progress: bool
    ) -> None:
        # This will never be called because the file must be in the cache
        raise NotImplementedError()

    def _upload(self, local_path: Path, hide_progress: bool) -> None:
        pass  # the "upload" is already done py the parent's caching mechanism

    def _delete_old_remote_versions(self) -> None:
        if self._keep_last_n is not None:
            for i in (
                self._instances[: -self._keep_last_n]
                if self._keep_last_n > 0
                else self._instances
            ):
                logger.info(
                    f"Removing old version (keep_last_n={self._keep_last_n}): {i.remote_path}"
                )
                if i.remote_path.is_dir():
                    shutil.rmtree(i.remote_path)
                else:
                    i.remote_path.unlink()

LargeFileS3 #

Bases: LargeFileBase

LargeFile implementation using S3-compatible storage as a backend.

Store large files remotely using the familiar API of open(). With built-in versioning, pruning and local cache.

See parent for more details.

Source code in great_ai/large_file/large_file/large_file_s3.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
class LargeFileS3(LargeFileBase):
    """LargeFile implementation using S3-compatible storage as a backend.

    Store large files remotely using the familiar API of `open()`. With built-in
    versioning, pruning and local cache.

    See parent for more details.
    """

    region_name = None
    access_key_id = None
    secret_access_key = None
    bucket_name = None
    endpoint_url = None
    _configuration_attributes = (
        "region_name",
        "access_key_id",
        "secret_access_key",
        "bucket_name",
        "endpoint_url",
    )

    def _get_cache_namespace(self) -> str:
        identity = ("s3", self.endpoint_url, self.region_name, self.bucket_name)
        return sha256(repr(identity).encode()).hexdigest()

    @classmethod
    def configure_credentials(  # type: ignore
        cls,
        *,
        aws_region_name: str,
        aws_access_key_id: str,
        aws_secret_access_key: str,
        large_files_bucket_name: str,
        aws_endpoint_url: Optional[str] = None,
        **_: Any,
    ) -> None:
        cls.region_name = aws_region_name
        cls.access_key_id = aws_access_key_id
        cls.secret_access_key = aws_secret_access_key
        cls.bucket_name = large_files_bucket_name
        cls.endpoint_url = aws_endpoint_url
        super().configure_credentials()

    @cached_property
    def _client(self) -> boto3.client:
        if (
            self.region_name is None
            or self.access_key_id is None
            or self.secret_access_key is None
            or self.bucket_name is None
        ):
            raise ValueError(
                "Please configure the S3 access options by calling LargeFileS3.configure_credentials or set cache_only_mode=True in the constructor."
            )

        return boto3.client(
            "s3",
            aws_access_key_id=self.access_key_id,
            aws_secret_access_key=self.secret_access_key,
            region_name=self.region_name,
            endpoint_url=self.endpoint_url,
        )

    def _find_remote_instances(self) -> List[DataInstance]:
        logger.debug(f"Fetching S3 versions of {self._name}")

        prefix = f"{self._name}{S3_NAME_VERSION_SEPARATOR}"
        parameters: Dict[str, Any] = {"Bucket": self.bucket_name, "Prefix": prefix}
        instances = []
        while True:
            page = self._client.list_objects_v2(**parameters)
            for item in page.get("Contents", []):
                key = item["Key"]
                suffix = key[len(prefix) :] if key.startswith(prefix) else ""
                if suffix.isdigit():
                    instances.append(
                        DataInstance(
                            name=self._name,
                            version=int(suffix),
                            remote_path=key,
                            cache_identity=(
                                repr((item["ETag"], item.get("LastModified")))
                                if item.get("ETag") is not None
                                else None
                            ),
                        )
                    )
            if not page.get("IsTruncated"):
                return instances
            parameters["ContinuationToken"] = page["NextContinuationToken"]

    def _download(
        self, remote_path: Any, local_path: Path, hide_progress: bool
    ) -> None:
        logger.info(f"Downloading {remote_path} from S3")

        size = self._client.head_object(Bucket=self.bucket_name, Key=remote_path)[
            "ContentLength"
        ]

        self._client.download_file(
            Bucket=self.bucket_name,
            Key=remote_path,
            Filename=str(local_path),
            Callback=(
                None
                if hide_progress
                else DownloadProgressBar(
                    name=str(remote_path), size=size, logger=logger
                )
            ),
        )

    def _upload(self, local_path: Path, hide_progress: bool) -> None:
        key = f"{self._name}/{self.version}"
        logger.info(f"Uploading {self._local_name} to S3 as {key}")

        def prevent_overwrite(params: Dict[str, Any], **_: Any) -> None:
            if params.get("Bucket") == self.bucket_name and params.get("Key") == key:
                params["IfNoneMatch"] = "*"

        # S3Transfer does not accept IfNoneMatch in ExtraArgs. These botocore
        # hooks protect both single-part and multipart commits while retaining
        # the managed transfer's streaming, retries, and progress reporting.
        events = self._client.meta.events
        operations = ("PutObject", "CompleteMultipartUpload")
        for operation in operations:
            events.register(f"before-parameter-build.s3.{operation}", prevent_overwrite)
        try:
            self._client.upload_file(
                Filename=str(local_path),
                Bucket=self.bucket_name,
                Key=key,
                Callback=(
                    None
                    if hide_progress
                    else UploadProgressBar(path=local_path, logger=logger)
                ),
                Config=TransferConfig(preferred_transfer_client="classic"),
            )
        except (ClientError, S3UploadFailedError) as error:
            cause: Optional[BaseException] = error
            while cause is not None:
                if isinstance(cause, ClientError) and cause.response["Error"][
                    "Code"
                ] in {"PreconditionFailed", "ConditionalRequestConflict", "412", "409"}:
                    raise FileExistsError(
                        f"Version {key} was created by another writer; retry the upload"
                    ) from error
                cause = cause.__cause__ or cause.__context__
            raise
        finally:
            for operation in operations:
                events.unregister(
                    f"before-parameter-build.s3.{operation}", prevent_overwrite
                )

    def _delete_old_remote_versions(self) -> None:
        if self._keep_last_n is not None:
            for i in (
                self._instances[: -self._keep_last_n]
                if self._keep_last_n > 0
                else self._instances
            ):
                logger.info(
                    f"Removing old version from S3 (keep_last_n={self._keep_last_n}): {i.remote_path}"
                )
                self._client.delete_object(Bucket=self.bucket_name, Key=i.remote_path)

LargeFileMongo #

Bases: LargeFileBase

LargeFile implementation using GridFS (MongoDB) as a backend.

Store large files remotely using the familiar API of open(). With built-in versioning, pruning and local cache.

See parent for more details.

Source code in great_ai/large_file/large_file/large_file_mongo.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class LargeFileMongo(LargeFileBase):
    """LargeFile implementation using GridFS (MongoDB) as a backend.

    Store large files remotely using the familiar API of `open()`. With built-in
    versioning, pruning and local cache.

    See parent for more details.
    """

    mongo_connection_string = None
    mongo_database = None
    _configuration_attributes = ("mongo_connection_string", "mongo_database")

    def _get_cache_namespace(self) -> str:
        return sha256(
            repr(("gridfs", self.mongo_connection_string, self.mongo_database)).encode()
        ).hexdigest()

    @classmethod
    def configure_credentials(  # type: ignore
        cls,
        *,
        mongo_connection_string: str,
        mongo_database: str,
        **_: Any,
    ) -> None:
        cls.mongo_connection_string = mongo_connection_string
        cls.mongo_database = mongo_database
        super().configure_credentials()

    @cached_property
    def _database(self) -> Database:
        if self.mongo_connection_string is None or self.mongo_database is None:
            raise ValueError(
                "Please configure the MongoDB access options by calling LargeFileMongo.configure_credentials or set cache_only_mode=True in the constructor."
            )

        return MongoClient(self.mongo_connection_string)[self.mongo_database]

    @cached_property
    def _client(self) -> GridFSBucket:
        return GridFSBucket(self._database)

    def _find_remote_instances(self) -> List[DataInstance]:
        logger.debug(f"Fetching Mongo (GridFS) versions of {self._name}")

        return [
            DataInstance(
                name=MONGO_NAME_VERSION_SEPARATOR.join(
                    f.name.split(MONGO_NAME_VERSION_SEPARATOR)[:-1]
                ),
                version=int(f.name.split(MONGO_NAME_VERSION_SEPARATOR)[-1]),
                remote_path=(f._id, f.length),
                cache_identity=str(f._id),
            )
            for f in self._client.find(
                {
                    "filename": re.compile(
                        "^"
                        + re.escape(self._name + MONGO_NAME_VERSION_SEPARATOR)
                        + r"[0-9]+$"
                    )
                }
            )
        ]

    def _download(
        self, remote_path: Any, local_path: Path, hide_progress: bool
    ) -> None:
        logger.info(f"Downloading {remote_path[0]} from Mongo (GridFS)")

        progress = (
            DownloadProgressBar(
                name=str(remote_path[0]), size=remote_path[1], logger=logger
            )
            if not hide_progress
            else None
        )
        with self._client.open_download_stream(remote_path[0]) as stream:
            with open(local_path, "wb") as f:
                while True:
                    content = stream.read(DEFAULT_CHUNK_SIZE)
                    f.write(content)

                    if progress:
                        progress(len(content))
                    if len(content) < DEFAULT_CHUNK_SIZE:
                        break

    def _upload(self, local_path: Path, hide_progress: bool) -> None:
        logger.info(f"Uploading {local_path} to Mongo (GridFS)")

        progress = (
            UploadProgressBar(path=local_path, logger=logger)
            if not hide_progress
            else None
        )
        # Only GreatAI's versioned uploads participate in this unique index;
        # unrelated GridFS users may still store duplicate filenames.
        self._database.fs.files.create_index(
            "filename",
            name="great_ai_unique_version",
            unique=True,
            partialFilterExpression={"metadata.great_ai_version": True},
        )
        filename = f"{self._name}{MONGO_NAME_VERSION_SEPARATOR}{self.version}"
        stream = self._client.open_upload_stream(
            filename, metadata={"great_ai_version": True}
        )
        try:
            with stream:
                with open(local_path, "rb") as f:
                    while True:
                        content = f.read(DEFAULT_CHUNK_SIZE)
                        stream.write(content)
                        if progress:
                            progress(len(content))
                        if len(content) < DEFAULT_CHUNK_SIZE:
                            break
        except DuplicateKeyError as error:
            stream.abort()
            raise FileExistsError(
                f"Version {filename} was created by another writer; retry the upload"
            ) from error

    def _delete_old_remote_versions(self) -> None:
        if self._keep_last_n is not None:
            for i in (
                self._instances[: -self._keep_last_n]
                if self._keep_last_n > 0
                else self._instances
            ):
                logger.info(
                    f"Removing old version from MongoDB (GridFS) (keep_last_n={self._keep_last_n}): {i.name}{MONGO_NAME_VERSION_SEPARATOR}{i.version}"
                )
                self._client.delete(i.remote_path[0])

LargeFileBase #

Bases: ABC

Base for LargeFile implementations with different backends.

Store large files remotely using the familiar API of open(). With built-in versioning, pruning and local cache.

By default, files are stored in the ".cache" folder and the least recently used is deleted after the overall size reaches 30 GBs.

Examples:

>>> LargeFileBase.cache_path = Path(".cache")

>>> LargeFileBase.max_cache_size = "30GB"

Attributes:

Name Type Description
initialized

Tell whether configure_credentials or configure_credentials_from_file has been already called.
cache_path

Storage location for cached files.
max_cache_size Optional[str]

Delete files until the folder at cache_path is smaller than this value. Examples: "5 GB", "10MB", "0.3 TB". Set to None for no automatic cache-pruning.
Source code in great_ai/large_file/large_file/large_file_base.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
class LargeFileBase(ABC):
    """Base for LargeFile implementations with different backends.

    Store large files remotely using the familiar API of `open()`. With built-in
    versioning, pruning and local cache.

    By default, files are stored in the ".cache" folder and the least recently used is
    deleted after the overall size reaches 30 GBs.

    Examples:
        >>> LargeFileBase.cache_path = Path(".cache")

        >>> LargeFileBase.max_cache_size = "30GB"

    Attributes:
        initialized: Tell whether `configure_credentials` or
            `configure_credentials_from_file` has been already called.
        cache_path: Storage location for cached files.
        max_cache_size: Delete files until the folder at `cache_path` is smaller than
            this value. Examples: "5 GB", "10MB", "0.3 TB". Set to `None` for no
            automatic cache-pruning.
    """

    initialized = False
    cache_path = Path(".cache")
    max_cache_size: Optional[str] = "30GB"
    _configuration_attributes: Tuple[str, ...] = ()

    def __init__(
        self,
        name: str,
        mode: str = "r",
        *,
        buffering: int = -1,
        encoding: Optional[str] = None,
        errors: Optional[str] = None,
        newline: Optional[str] = None,
        version: Optional[int] = None,
        keep_last_n: Optional[int] = None,
        cache_only_mode: bool = False,
    ):
        if keep_last_n is not None and keep_last_n < 0:
            raise ValueError("keep_last_n must be non-negative")
        clean_name = re.sub(r"[^!a-zA-Z0-9._-]+", "", name)
        if clean_name != name:
            raise ValueError(
                f"Given name contains illegal characters, consider changing it to: `{clean_name}`"
            )

        self._name = name
        self._version = version
        self._mode = mode
        self._keep_last_n = keep_last_n
        self._cache_only_mode = cache_only_mode

        self._buffering = buffering
        self._encoding = encoding

        if errors is not None and sys.version_info[1] < 8:
            raise RuntimeError(
                "The `errors` kwarg is only supported in 3.8 <= Python versions."
            )
        self._errors = errors

        self._newline = newline

        # Existing instances keep the storage configuration they were opened with.
        for attribute in self._configuration_attributes:
            setattr(self, attribute, getattr(self, attribute))
        self._cache_namespace = self._get_cache_namespace()
        self.cache_path.mkdir(parents=True, exist_ok=True)

        self._find_instances()
        self._check_mode_and_set_version()

    @classmethod
    def configure_credentials_from_file(
        cls,
        secrets: Union[Path, str, ConfigFile],
    ) -> None:
        """Load file and feed its content to `configure_credentials`.

        Extra keys are ignored.
        """

        if not isinstance(secrets, ConfigFile):
            secrets = ConfigFile(secrets)
        cls.configure_credentials(**{k.lower(): v for k, v in secrets.items()})

    @classmethod
    def configure_credentials(cls, **kwargs: str) -> None:
        """Configure required credentials for the LargeFile backend."""

        cls.initialized = True

    def __enter__(self) -> IO:
        params = dict(
            mode=self._mode,
            buffering=self._buffering,
            encoding=self._encoding,
            newline=self._newline,
            delete=False,
            prefix="large_file-",
        )

        if sys.version_info[1] >= 8:
            params["errors"] = self._errors

        self._file: IO[Any] = (
            tempfile.NamedTemporaryFile(**params)  # type: ignore
            if "w" in self._mode
            else open(
                self.get(),
                mode=self._mode,
                buffering=self._buffering,
                encoding=self._encoding,
                newline=self._newline,
                errors=self._errors,
            )
        )

        return self._file

    def __exit__(
        self,
        type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> Literal[False]:
        self._file.close()

        if "w" in self._mode:
            try:
                if type is None:
                    self.push(Path(self._file.name))
            finally:
                os.unlink(self._file.name)
        elif type is not None:
            logger.exception("Could not finish operation.")

        return False

    @property
    def version(self) -> int:
        """Numeric version of the file proxied by this LargeFile instance."""

        return cast(int, self._version)

    def get(self, hide_progress: bool = False) -> Path:
        """Return path to the proxy of a file (or directory).

        If not available in the local cache, an attempt is made to download it.

        Args:
            hide_progress: Do not show a progress update after each 10% of progress.
        """
        with self._model_lock():
            return self._get_cached_file(hide_progress)

    def _get_cached_file(self, hide_progress: bool) -> Path:
        if not self._cache_only_mode:
            self._find_instances()
        instance = next(
            (i for i in self._instances if i.version == self._version), None
        )
        if instance is None:
            raise FileNotFoundError(
                f"File {self._name} version {self.version} not found"
            )

        destination = self.cache_path / self._cache_name
        if destination.exists() and not self._cache_only_mode:
            identity = instance.cache_identity
            if (
                identity is None
                or not self._cache_identity_path.exists()
                or self._cache_identity_path.read_text() != identity
            ):
                self._remove_cached_path(destination)
        if not destination.exists():
            logger.info(f"File {self._local_name} does not exist locally")

            with tempfile.TemporaryDirectory() as tmp:
                local_root_path = Path(tmp)
                tmp_file_archive = (
                    local_root_path / f"{self._local_name}{ARCHIVE_EXTENSION}"
                )
                self._download(
                    instance.remote_path, tmp_file_archive, hide_progress=hide_progress
                )

                logger.info(f"Decompressing {self._local_name}")
                shutil.unpack_archive(
                    str(tmp_file_archive), tmp, COMPRESSION_ALGORITHM, filter="data"
                )
                shutil.move(str(local_root_path / self._local_name), str(destination))
                self._write_cache_identity()
        else:
            logger.info(f"File {self._local_name} found in cache")

        os.utime(destination, None)
        self._prune_cache(protected=destination)
        return destination

    def push(self, path: Union[Path, str], hide_progress: bool = False) -> None:
        """Upload a file (or directory) as a new version of `key`.

        The file/directory is compressed before upload.

        Args:
            hide_progress: Do not show a progress update after each 10% of progress.
        """

        if isinstance(path, str):
            path = Path(path)

        staging_directory = self.cache_path / ".staging"
        staging_directory.mkdir(parents=True, exist_ok=True)
        with self._model_lock():
            # Allocate at commit time, after earlier local writers have finished.
            self._find_instances()
            self._version = self._instances[-1].version + 1 if self._instances else 0
            with tempfile.TemporaryDirectory(dir=staging_directory) as tmp:
                root = Path(tmp)
                staged = root / self._local_name
                if path.is_dir():
                    shutil.copytree(path, staged)
                else:
                    shutil.copy2(path, staged)

                with tempfile.TemporaryDirectory() as archive_directory:
                    archive = shutil.make_archive(
                        str(Path(archive_directory) / self._local_name),
                        COMPRESSION_ALGORITHM,
                        root,
                    )
                    # Remote backends must reject version conflicts atomically.
                    self._upload(Path(archive), hide_progress=hide_progress)

                # Stage the source before publishing: it may itself be cached.
                destination = self.cache_path / self._cache_name
                if destination.exists():
                    if self._cache_only_mode:
                        raise FileExistsError(f"Version already exists: {destination}")
                    # The remote conditional commit succeeded. Any existing local
                    # entry belongs to a deleted generation of this version.
                    self._remove_cached_path(destination)
                shutil.move(str(staged), str(destination))

            self._find_instances()
            self._write_cache_identity()
            self.clean_up()

    def delete(self) -> None:
        """Delete all versions of the files under this `key`."""

        self._keep_last_n = 0
        self._delete_old_remote_versions()

    @property
    def versions_pretty(self) -> str:
        """Formatted string of all available versions."""
        return ", ".join((str(i.version) for i in self._instances))

    def clean_up(self) -> None:
        """Delete local and remote versions according to currently set cache and retention policy."""

        self._delete_old_remote_versions()
        self._prune_cache()

    @property
    def _local_name(self) -> str:
        return f"{self._name}{CACHE_NAME_VERSION_SEPARATOR}{self.version}"

    @property
    def _cache_name(self) -> str:
        # '@' is forbidden in model names, so even local names that resemble
        # namespace hashes cannot collide with remote cache entries.
        prefix = f"@{self._cache_namespace}-" if self._cache_namespace else ""
        return f"{prefix}{self._local_name}"

    def _get_cache_namespace(self) -> str:
        return sha256(
            f"{type(self).__module__}.{type(self).__qualname__}".encode()
        ).hexdigest()

    def _model_lock(self) -> FileLock:
        directory = self.cache_path / ".locks"
        directory.mkdir(parents=True, exist_ok=True)
        key = sha256(f"{self._cache_namespace}:{self._name}".encode()).hexdigest()
        return FileLock(directory / f"{key}.lock")

    @property
    def _cache_identity_path(self) -> Path:
        key = sha256(self._cache_name.encode()).hexdigest()
        return self.cache_path / ".metadata" / key

    def _write_cache_identity(self) -> None:
        if self._cache_only_mode:
            return
        identity = next(
            (i.cache_identity for i in self._instances if i.version == self.version),
            None,
        )
        if identity is not None:
            self._cache_identity_path.parent.mkdir(parents=True, exist_ok=True)
            self._cache_identity_path.write_text(identity)

    @staticmethod
    def _remove_cached_path(path: Path) -> None:
        if path.is_dir():
            shutil.rmtree(path)
        else:
            path.unlink()

    def _find_instances(self) -> None:
        if self._cache_only_mode:
            self._instances = self._find_instances_from_cache()
        else:
            self._instances = self._find_remote_instances()

        self._instances = sorted(self._instances, key=lambda i: i.version)

    def _find_instances_from_cache(self) -> List[DataInstance]:
        logger.info(f"Fetching cached versions of {self._name}")

        namespace = f"@{self._cache_namespace}-" if self._cache_namespace else ""
        prefix = f"{namespace}{self._name}{CACHE_NAME_VERSION_SEPARATOR}"
        return [
            DataInstance(
                name=self._name, version=int(f.name[len(prefix) :]), remote_path=f
            )
            for f in self.cache_path.glob(f"{prefix}*")
            if f.name[len(prefix) :].isdigit()
        ]

    def _check_mode_and_set_version(self) -> None:
        if "+" in self._mode:
            raise ValueError(
                f"File mode `{self._mode}` is not allowed3, remove the `+`."
            )

        if "w" in self._mode:
            if self._version is not None:
                raise ValueError("Providing a version is not allowed in write mode.")

            self._version = self._instances[-1].version + 1 if self._instances else 0

        elif "r" in self._mode:
            if not self._instances:
                raise FileNotFoundError(
                    f"File {self._name} not found. No versions are available."
                )

            if self._version is None:
                self._version = self._instances[-1].version
                logger.info(
                    f"Latest version of {self._name} is {self._version} "
                    + f"(from versions: {self.versions_pretty})"
                )
            elif self._version not in [i.version for i in self._instances]:
                raise FileNotFoundError(
                    f"File {self._name} not found with version {self._version}. "
                    + f"(from versions: {self.versions_pretty})"
                )
        else:
            raise ValueError("Unsupported file mode.")

    def _prune_cache(self, protected: Optional[Path] = None) -> None:
        self.cache_path.mkdir(parents=True, exist_ok=True)

        if self.max_cache_size is None:
            return

        allowed_size = human_readable_to_byte(self.max_cache_size)
        assert allowed_size >= 0

        least_recently_read = sorted(
            [
                f
                for f in self.cache_path.glob("*")
                if f.name not in {".locks", ".staging", ".metadata"}
            ],
            key=lambda f: f.stat().st_atime,
        )

        def size(path: Path) -> int:
            if path.is_dir():
                return sum(f.stat().st_size for f in path.rglob("*") if f.is_file())
            return path.stat().st_size

        if protected is not None and size(protected) > allowed_size:
            self._remove_cached_path(protected)
            self._cache_identity_path.unlink(missing_ok=True)
            raise ValueError(
                f"File {self._name} exceeds max_cache_size={self.max_cache_size}; "
                "increase the cache quota to read this file"
            )

        while sum(size(f) for f in least_recently_read) > allowed_size:
            file = least_recently_read.pop(0)
            if file == protected:
                least_recently_read.append(file)
                continue
            logger.info(
                f"Deleting file from cache to meet quota (max_cache_size={self.max_cache_size}): {file}"
            )
            if file.is_dir():
                shutil.rmtree(file)
            else:
                file.unlink()
            metadata = (
                self.cache_path / ".metadata" / sha256(file.name.encode()).hexdigest()
            )
            metadata.unlink(missing_ok=True)

    @abstractmethod
    def _find_remote_instances(self) -> List[DataInstance]:
        pass

    @abstractmethod
    def _download(
        self, remote_path: Any, local_path: Path, hide_progress: bool
    ) -> None:
        pass

    @abstractmethod
    def _upload(self, local_path: Path, hide_progress: bool) -> None:
        pass

    @abstractmethod
    def _delete_old_remote_versions(self) -> None:
        pass

version property #

Numeric version of the file proxied by this LargeFile instance.

versions_pretty property #

Formatted string of all available versions.

clean_up() #

Delete local and remote versions according to currently set cache and retention policy.

Source code in great_ai/large_file/large_file/large_file_base.py 
291
292
293
294
295
def clean_up(self) -> None:
    """Delete local and remote versions according to currently set cache and retention policy."""

    self._delete_old_remote_versions()
    self._prune_cache()

configure_credentials(**kwargs) classmethod #

Configure required credentials for the LargeFile backend.

Source code in great_ai/large_file/large_file/large_file_base.py 
116
117
118
119
120
@classmethod
def configure_credentials(cls, **kwargs: str) -> None:
    """Configure required credentials for the LargeFile backend."""

    cls.initialized = True

configure_credentials_from_file(secrets) classmethod #

Load file and feed its content to configure_credentials.

Extra keys are ignored.

Source code in great_ai/large_file/large_file/large_file_base.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
@classmethod
def configure_credentials_from_file(
    cls,
    secrets: Union[Path, str, ConfigFile],
) -> None:
    """Load file and feed its content to `configure_credentials`.

    Extra keys are ignored.
    """

    if not isinstance(secrets, ConfigFile):
        secrets = ConfigFile(secrets)
    cls.configure_credentials(**{k.lower(): v for k, v in secrets.items()})

delete() #

Delete all versions of the files under this key.

Source code in great_ai/large_file/large_file/large_file_base.py 
280
281
282
283
284
def delete(self) -> None:
    """Delete all versions of the files under this `key`."""

    self._keep_last_n = 0
    self._delete_old_remote_versions()

get(hide_progress=False) #

Return path to the proxy of a file (or directory).

If not available in the local cache, an attempt is made to download it.

Parameters:

Name Type Description Default
hide_progress bool

Do not show a progress update after each 10% of progress.

 False
Source code in great_ai/large_file/large_file/large_file_base.py 
175
176
177
178
179
180
181
182
183
184
def get(self, hide_progress: bool = False) -> Path:
    """Return path to the proxy of a file (or directory).

    If not available in the local cache, an attempt is made to download it.

    Args:
        hide_progress: Do not show a progress update after each 10% of progress.
    """
    with self._model_lock():
        return self._get_cached_file(hide_progress)

push(path, hide_progress=False) #

Upload a file (or directory) as a new version of key.

The file/directory is compressed before upload.

Parameters:

Name Type Description Default
hide_progress bool

Do not show a progress update after each 10% of progress.

 False
Source code in great_ai/large_file/large_file/large_file_base.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def push(self, path: Union[Path, str], hide_progress: bool = False) -> None:
    """Upload a file (or directory) as a new version of `key`.

    The file/directory is compressed before upload.

    Args:
        hide_progress: Do not show a progress update after each 10% of progress.
    """

    if isinstance(path, str):
        path = Path(path)

    staging_directory = self.cache_path / ".staging"
    staging_directory.mkdir(parents=True, exist_ok=True)
    with self._model_lock():
        # Allocate at commit time, after earlier local writers have finished.
        self._find_instances()
        self._version = self._instances[-1].version + 1 if self._instances else 0
        with tempfile.TemporaryDirectory(dir=staging_directory) as tmp:
            root = Path(tmp)
            staged = root / self._local_name
            if path.is_dir():
                shutil.copytree(path, staged)
            else:
                shutil.copy2(path, staged)

            with tempfile.TemporaryDirectory() as archive_directory:
                archive = shutil.make_archive(
                    str(Path(archive_directory) / self._local_name),
                    COMPRESSION_ALGORITHM,
                    root,
                )
                # Remote backends must reject version conflicts atomically.
                self._upload(Path(archive), hide_progress=hide_progress)

            # Stage the source before publishing: it may itself be cached.
            destination = self.cache_path / self._cache_name
            if destination.exists():
                if self._cache_only_mode:
                    raise FileExistsError(f"Version already exists: {destination}")
                # The remote conditional commit succeeded. Any existing local
                # entry belongs to a deleted generation of this version.
                self._remove_cached_path(destination)
            shutil.move(str(staged), str(destination))

        self._find_instances()
        self._write_cache_identity()
        self.clean_up()