Repository: huggingface/huggingface_hub
Pull request #1494 (2023-06-02 16:53:10)

`CommitScheduler`: upload folder every 5 minutes
Previously discussed in https://github.com/huggingface/huggingface_hub/pull/1479. See also [slack thread](https://huggingface.slack.com/archives/C03V11RNS7P/p1685702404165379) (internal).

In the end we are going for a simple implementation based on a scheduler that uploads a folder every X minutes. The main use case we are focusing on is when the folder is **append-only**. We consider that existing content is never overwritten. Examples: tensorboard logger, append data to a CSV/JSONL, save generated images to a folder,... It should be possible in the future to support other cases if the demand is there.

If needed, `with scheduler.lock: ...` is available to avoid writing to a file while committing. In 99% cases it'll be useless as the scheduler needs the lock only when it scans the folder (not when uploading it - thanks to the append-only mode).

Here is a small example on how to use it:
```py
file_path = Path("watched_folder/data.csv")

# Fake example : commit every 5s. In practice, should be 5-10 minutes.
CommitScheduler(repo_id="test_scheduler", folder_path=file_path.parent, every=5 / 60)

# or maybe keep the same existing `upload_folder` and add a `every` parameter? :
# upload_folder(repo_id="test_scheduler", folder_path=file_path.parent, every=5 / 60)

time.sleep(2)
with file_path.open("a") as f:
    f.write("first line\n")

time.sleep(6)  # 1 commit in between
with file_path.open("a") as f:
    f.write("second line\n")

time.sleep(12)  # 2 or 3 commits in between but only 1 is pushed (no changes otherwise)
with file_path.open("a") as f:
    f.write("third line\n")

# One last commit triggered before script ends
```

I end up with this repo ([initial commit + 3 commits](https://huggingface.co/Wauplin/test_scheduler/commits/main)). The scheduler keeps track of the files previously uploaded to avoid pushing empty commits.

**TODOs:**
- [x] implement the "partial file upload"
- [ ] ~add to `HfApi.upload_folder` ?~
- [x] add unit tests
- [x] document ([guide](https://moon-ci-docs.huggingface.co/docs/huggingface_hub/pr_1494/en/guides/upload#scheduled-uploads))
- [ ] _(discussed but I'm not a big fan)_ add warnings / error if a file is modified in non-append mode (requires to monkeypatch `.open(...)`)

HuggingFaceDocBuilderDev: The docs for this PR live [here](https://moon-ci-docs.huggingface.co/docs/huggingface_hub/pr_1494). All of your documentation changes will be reflected on that endpoint.
julien-c: love the simple API so far!
Wauplin: Thanks for the review :pray:  Let's merge this! :tada: 

Patch:
diff --git a/docs/source/guides/upload.mdx b/docs/source/guides/upload.mdx
index 62622720..79cdf248 100644
--- a/docs/source/guides/upload.mdx
+++ b/docs/source/guides/upload.mdx
@@ -195,6 +195,74 @@ notice.
 
 </Tip>
 
+### Scheduled uploads
+
+The Hugging Face Hub makes it easy to save and version data. However, there are some limitations when updating the same file thousands of times. For instance, you might want to save logs of a training process or user
+feedback on a deployed Space. In these cases, uploading the data as a dataset on the Hub makes sense, but it can be hard to do properly. The main reason is that you don't want to version every update of your data because it'll make the git repository unusable. The [`CommitScheduler`] class offers a solution to this problem.
+
+The idea is to run a background job that regularly pushes a local folder to the Hub. Let's assume you have a
+Gradio Space that takes as input some text and generates two translations of it. Then, the user can select their preferred translation. For each run, you want to save the input, output, and user preference to analyze the results. This is a
+perfect use case for [`CommitScheduler`]; you want to save data to the Hub (potentially millions of user feedback), but
+you don't _need_ to save in real-time each user's input. Instead, you can save the data locally in a JSON file and
+upload it every 10 minutes. For example:
+
+```py
+>>> import json
+>>> import uuid
+>>> from pathlib import Path
+>>> import gradio as gr
+>>> from huggingface_hub import CommitScheduler
+
+# Define the file where to save the data. Use UUID to make sure not to overwrite existing data from a previous run.
+>>> feedback_file = Path("user_feedback/") / f"data_{uuid.uuid4()}.json"
+>>> feedback_folder = feedback_file.parent
+
+# Schedule regular uploads. Remote repo and local folder are created if they don't already exist.
+>>> scheduler = CommitScheduler(
+...     repo_id="report-translation-feedback",
+...     repo_type="dataset",
+...     folder_path=feedback_folder,
+...     path_in_repo="data",
+...     every=10,
+... )
+
+# Define the function that will be called when the user submits its feedback (to be called in Gradio)
+>>> def save_feedback(input_text:str, output_1: str, output_2:str, user_choice: int) -> None:
+...     """
+...     Append input/outputs and user feedback to a JSON Lines file using a thread lock to avoid concurrent writes from different users.
+...     """
+...     with scheduler.lock:
+...         with feedback_file.open("a") as f:
+...             f.write(json.dumps({"input": input_text, "output_1": output_1, "output_2": output_2, "user_choice": user_choice}))
+...             f.write("\n")
+
+# Start Gradio
+>>> with gr.Blocks() as demo:
+>>>     ... # define Gradio demo + use `save_feedback`
+>>> demo.launch()
+```
+
+And that's it! User input/outputs and feedback will be available as a dataset on the Hub. By using a unique JSON file name, you are guaranteed you won't overwrite data from a previous run or data from another
+Spaces/replicas pushing concurrently to the same repository.
+
+For more details about the [`CommitScheduler`], here is what you need to know:
+- **append-only:**
+    It is assumed that you will only add content to the folder. You must only append data to existing files or create
+    new files. Deleting or overwriting a file might corrupt your repository.
+- **git history**:
+    The scheduler will commit the folder every `every` minutes. To avoid polluting the git repository too much, it is
+    recommended to set a minimal value of 5 minutes. Besides, the scheduler is designed to avoid empty commits. If no
+    new content is detected in the folder, the scheduled commit is dropped.
+- **errors:**
+    The scheduler run as background thread. It is started when you instantiate the class and never stops. In particular,
+    if an error occurs during the upload (example: connection issue), the scheduler will silently ignore it and retry
+    at the next scheduled commit.
+- **thread-safety:**
+    In most cases it is safe to assume that you can write to a file without having to worry about a lock file. The
+    scheduler will not crash or be corrupted if you write content to the folder while it's uploading. In practice,
+    _it is possible_ that concurrency issues happen for heavy-loaded apps. In this case, we advice to use the
+    `scheduler.lock` lock to ensure thread-safety. The lock is blocked only when the scheduler scans the folder for
+    changes, not when it uploads data. You can safely assume that it will not affect the user experience on your Space.
 
 ### create_commit
 
@@ -257,7 +325,7 @@ huggingface-cli lfs-enable-largefiles
 
 You should install this for each repository that has a very large file. Once installed, you'll be able to push files larger than 5GB.
 
-## commit context manager
+### commit context manager
 
 The `commit` context manager handles four of the most common Git commands: pull, add, commit, and push. `git-lfs` automatically tracks any file larger than 10MB. In the following example, the `commit` context manager:
 
@@ -316,7 +384,7 @@ When `blocking=False`, commands are tracked, and your script will only exit when
 >>> last_command.failed
 ```
 
-## push_to_hub
+### push_to_hub
 
 The [`Repository`] class has a [`~Repository.push_to_hub`] function to add files, make a commit, and push them to a repository. Unlike the `commit` context manager, you'll need to pull from a repository first before calling [`~Repository.push_to_hub`].
 
diff --git a/docs/source/package_reference/hf_api.mdx b/docs/source/package_reference/hf_api.mdx
index 098c6abf..3f1f4c10 100644
--- a/docs/source/package_reference/hf_api.mdx
+++ b/docs/source/package_reference/hf_api.mdx
@@ -81,6 +81,10 @@ Below are the supported values for [`CommitOperation`]:
 
 [[autodoc]] CommitOperationCopy
 
+## CommitScheduler
+
+[[autodoc]] CommitScheduler
+
 ## Token helper
 
 `huggingface_hub` stores the authentication information locally so that it may be re-used in subsequent
diff --git a/src/huggingface_hub/__init__.py b/src/huggingface_hub/__init__.py
index 00b09410..5a33a5e4 100644
--- a/src/huggingface_hub/__init__.py
+++ b/src/huggingface_hub/__init__.py
@@ -52,6 +52,9 @@ __version__ = "0.16.0.dev0"
 # WARNING: any comment added in this dictionary definition will be lost when
 # re-generating the file !
 _SUBMOD_ATTRS = {
+    "_commit_scheduler": [
+        "CommitScheduler",
+    ],
     "_inference": [
         "InferenceClient",
         "InferenceTimeoutError",
@@ -347,6 +350,7 @@ __getattr__, __dir__, __all__ = _attach(__name__, submodules=[], submod_attrs=_S
 # make style
 # ```
 if TYPE_CHECKING:  # pragma: no cover
+    from ._commit_scheduler import CommitScheduler  # noqa: F401
     from ._inference import (
         InferenceClient,  # noqa: F401
         InferenceTimeoutError,  # noqa: F401
diff --git a/src/huggingface_hub/_commit_scheduler.py b/src/huggingface_hub/_commit_scheduler.py
new file mode 100644
index 00000000..58655b70
--- /dev/null
+++ b/src/huggingface_hub/_commit_scheduler.py
@@ -0,0 +1,288 @@
+import atexit
+import logging
+import os
+import time
+from dataclasses import dataclass
+from io import SEEK_END, SEEK_SET, BytesIO
+from pathlib import Path
+from threading import Lock, Thread
+from typing import Dict, List, Optional, Union
+
+from .hf_api import IGNORE_GIT_FOLDER_PATTERNS, CommitInfo, CommitOperationAdd, HfApi
+from .utils import filter_repo_objects
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class _FileToUpload:
+    """Temporary dataclass to store info about files to upload. Not meant to be used directly."""
+
+    local_path: Path
+    path_in_repo: str
+    size_limit: int
+    last_modified: float
+
+
+class CommitScheduler:
+    """
+    Scheduler to upload a local folder to the Hub at regular intervals (e.g. push to hub every 5 minutes).
+
+    The scheduler is started when instantiated and run indefinitely. At the end of your script, a last commit is
+    triggered. Checkout the [upload guide](https://huggingface.co/docs/huggingface_hub/guides/upload#scheduled-uploads)
+    to learn more about how to use it.
+
+    Args:
+        repo_id (`str`):
+            The id of the repo to commit to.
+        folder_path (`str` or `Path`):
+            Path to the local folder to upload regularly.
+        every (`int` or `float`, *optional*):
+            The number of minutes between each commit. Defaults to 5 minutes.
+        path_in_repo (`str`, *optional*):
+            Relative path of the directory in the repo, for example: `"checkpoints/"`. Defaults to the root folder
+            of the repository.
+        repo_type (`str`, *optional*):
+            The type of the repo to commit to. Defaults to `model`.
+        revision (`str`, *optional*):
+            The revision of the repo to commit to. Defaults to `main`.
+        private (`bool`, *optional*):
+            Whether to make the repo private. Defaults to `False`. This value is ignored if the repo already exist.
+        token (`str`, *optional*):
+            The token to use to commit to the repo. Defaults to the token saved on the machine.
+        allow_patterns (`List[str]` or `str`, *optional*):
+            If provided, only files matching at least one pattern are uploaded.
+        ignore_patterns (`List[str]` or `str`, *optional*):
+            If provided, files matching any of the patterns are not uploaded.
+        hf_api (`HfApi`, *optional*):
+            The [`HfApi`] client to use to commit to the Hub. Can be set with custom settings (user agent, token,...).
+
+    Example:
+    ```py
+    >>> from pathlib import Path
+    >>> from huggingface_hub import CommitScheduler
+
+    # Scheduler uploads every 10 minutes
+    >>> csv_path = Path("watched_folder/data.csv")
+    >>> CommitScheduler(repo_id="test_scheduler", repo_type="dataset", folder_path=csv_path.parent, every=10)
+
+    >>> with csv_path.open("a") as f:
+    ...     f.write("first line")
+
+    # Some time later (...)
+    >>> with csv_path.open("a") as f:
+    ...     f.write("second line")
+    ```
+    """
+
+    def __init__(
+        self,
+        *,
+        repo_id: str,
+        folder_path: Union[str, Path],
+        every: Union[int, float] = 5,
+        path_in_repo: Optional[str] = None,
+        repo_type: Optional[str] = None,
+        revision: Optional[str] = None,
+        private: bool = False,
+        token: Optional[str] = None,
+        allow_patterns: Optional[Union[List[str], str]] = None,
+        ignore_patterns: Optional[Union[List[str], str]] = None,
+        hf_api: Optional["HfApi"] = None,
+    ) -> None:
+        self.api = hf_api or HfApi()
+
+        # Folder
+        self.folder_path = Path(folder_path).expanduser().resolve()
+        self.path_in_repo = path_in_repo or ""
+        self.allow_patterns = allow_patterns
+
+        if ignore_patterns is None:
+            ignore_patterns = []
+        elif isinstance(ignore_patterns, str):
+            ignore_patterns = [ignore_patterns]
+        self.ignore_patterns = ignore_patterns + IGNORE_GIT_FOLDER_PATTERNS
+
+        if self.folder_path.is_file():
+            raise ValueError(f"'folder_path' must be a directory, not a file: '{self.folder_path}'.")
+        self.folder_path.mkdir(parents=True, exist_ok=True)
+
+        # Repository
+        repo_url = self.api.create_repo(
+            repo_id=repo_id, token=token, private=private, repo_type=repo_type, exist_ok=True
+        )
+        self.repo_id = repo_url.repo_id
+        self.repo_type = repo_type
+        self.revision = revision
+        self.token = token
+
+        # Keep track of already uploaded files
+        self.last_uploaded: Dict[Path, float] = {}  # key is local path, value is timestamp
+
+        # Scheduler
+        if not every > 0:
+            raise ValueError(f"'every' must be a positive integer, not '{every}'.")
+        self.lock = Lock()
+        self.every = every
+
+        logger.info(f"Scheduled job to push '{self.folder_path}' to '{self.repo_id}' every {self.every} minutes.")
+        self._scheduler_thread = Thread(target=self._run_scheduler, daemon=True)
+        self._scheduler_thread.start()
+        atexit.register(self._push_to_hub)
+
+        self.__stopped = False
+
+    def stop(self) -> None:
+        """Stop the scheduler.
+
+        A stopped scheduler cannot be restarted. Mostly for tests purposes.
+        """
+        self.__stopped = True
+
+    def _run_scheduler(self) -> None:
+        """Dumb thread waiting between each scheduled push to Hub."""
+        while True:
+            self.last_future = self.api.run_as_future(self._push_to_hub)
+            time.sleep(self.every * 60)
+            if self.__stopped:
+                break
+
+    def _push_to_hub(self) -> Optional[CommitInfo]:
+        if self.__stopped:  # If stopped, already scheduled commits are ignored
+            return None
+
+        logger.info("(Background) scheduled commit triggered.")
+
+        # Check files to upload (with lock)
+        with self.lock:
+            logger.debug("Listing files to upload for scheduled commit.")
+
+            # List files from folder (taken from `_prepare_upload_folder_additions`)
+            relpath_to_abspath = {
+                path.relative_to(self.folder_path).as_posix(): path
+                for path in sorted(self.folder_path.glob("**/*"))  # sorted to be deterministic
+                if path.is_file()
+            }
+            prefix = f"{self.path_in_repo.strip('/')}/" if self.path_in_repo else ""
+
+            # Filter with pattern + filter out unchanged files + retrieve current file size
+            files_to_upload: List[_FileToUpload] = []
+            for relpath in filter_repo_objects(
+                relpath_to_abspath.keys(), allow_patterns=self.allow_patterns, ignore_patterns=self.ignore_patterns
+            ):
+                local_path = relpath_to_abspath[relpath]
+                stat = local_path.stat()
+                if self.last_uploaded.get(local_path) is None or self.last_uploaded[local_path] != stat.st_mtime:
+                    files_to_upload.append(
+                        _FileToUpload(
+                            local_path=local_path,
+                            path_in_repo=prefix + relpath,
+                            size_limit=stat.st_size,
+                            last_modified=stat.st_mtime,
+                        )
+                    )
+
+        # Return if nothing to upload
+        if len(files_to_upload) == 0:
+            logger.debug("Dropping schedule commit: no changed file to upload.")
+            return None
+
+        # Convert `_FileToUpload` as `CommitOperationAdd` (=> compute file shas + limit to file size)
+        logger.debug("Removing unchanged files since previous scheduled commit.")
+        add_operations = [
+            CommitOperationAdd(
+                # Cap the file to its current size, even if the user append data to it while a scheduled commit is happening
+                path_or_fileobj=PartialFileIO(file_to_upload.local_path, size_limit=file_to_upload.size_limit),
+                path_in_repo=file_to_upload.path_in_repo,
+            )
+            for file_to_upload in files_to_upload
+        ]
+
+        # Upload files (append mode expected - no need for lock)
+        logger.debug("Uploading files for scheduled commit.")
+        commit_info = self.api.create_commit(
+            repo_id=self.repo_id,
+            repo_type=self.repo_type,
+            operations=add_operations,
+            commit_message="Scheduled Commit",
+            revision=self.revision,
+        )
+
+        # Successful commit: keep track of the latest "last_modified" for each file
+        for file in files_to_upload:
+            self.last_uploaded[file.local_path] = file.last_modified
+        return commit_info
+
+
+class PartialFileIO(BytesIO):
+    """A file-like object that reads only the first part of a file.
+
+    Useful to upload a file to the Hub when the user might still be appending data to it. Only the first part of the
+    file is uploaded (i.e. the part that was available when the filesystem was first scanned).
+
+    In practice, only used internally by the CommitScheduler to regularly push a folder to the Hub with minimal
+    disturbance for the user. The object is passed to `CommitOperationAdd`.
+
+    Only supports `read`, `tell` and `seek` methods.
+
+    Args:
+        file_path (`str` or `Path`):
+            Path to the file to read.
+        size_limit (`int`):
+            The maximum number of bytes to read from the file. If the file is larger than this, only the first part
+            will be read (and uploaded).
+    """
+
+    def __init__(self, file_path: Union[str, Path], size_limit: int) -> None:
+        self._file_path = Path(file_path)
+        self._file = self._file_path.open("rb")
+        self._size_limit = min(size_limit, os.fstat(self._file.fileno()).st_size)
+
+    def __del__(self) -> None:
+        self._file.close()
+        return super().__del__()
+
+    def __repr__(self) -> str:
+        return f"<PartialFileIO file_path={self._file_path} size_limit={self._size_limit}>"
+
+    def __len__(self) -> int:
+        return self._size_limit
+
+    def __getattribute__(self, name: str):
+        if name.startswith("_") or name in ("read", "tell", "seek"):  # only 3 public methods supported
+            return super().__getattribute__(name)
+        raise NotImplementedError(f"PartialFileIO does not support '{name}'.")
+
+    def tell(self) -> int:
+        """Return the current file position."""
+        return self._file.tell()
+
+    def seek(self, __offset: int, __whence: int = SEEK_SET) -> int:
+        """Change the stream position to the given offset.
+
+        Behavior is the same as a regular file, except that the position is capped to the size limit.
+        """
+        if __whence == SEEK_END:
+            # SEEK_END => set from the truncated end
+            __offset = len(self) + __offset
+            __whence = SEEK_SET
+
+        pos = self._file.seek(__offset, __whence)
+        if pos > self._size_limit:
+            return self._file.seek(self._size_limit)
+        return pos
+
+    def read(self, __size: Optional[int] = -1) -> bytes:
+        """Read at most `__size` bytes from the file.
+
+        Behavior is the same as a regular file, except that it is capped to the size limit.
+        """
+        current = self._file.tell()
+        if __size is None or __size < 0:
+            # Read until file limit
+            truncated_size = self._size_limit - current
+        else:
+            # Read until file limit or __size
+            truncated_size = min(__size, self._size_limit - current)
+        return self._file.read(truncated_size)


Test patch:
diff --git a/tests/test_commit_scheduler.py b/tests/test_commit_scheduler.py
new file mode 100644
index 00000000..477563e4
--- /dev/null
+++ b/tests/test_commit_scheduler.py
@@ -0,0 +1,243 @@
+import time
+import unittest
+from io import SEEK_END
+from pathlib import Path
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+
+from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download
+from huggingface_hub._commit_scheduler import CommitScheduler, PartialFileIO
+
+from .testing_constants import ENDPOINT_STAGING, TOKEN
+from .testing_utils import repo_name
+
+
+@pytest.mark.usefixtures("fx_cache_dir")
+class TestCommitScheduler(unittest.TestCase):
+    cache_dir: Path
+
+    def setUp(self) -> None:
+        self.api = HfApi(token=TOKEN, endpoint=ENDPOINT_STAGING)
+        self.repo_name = repo_name()
+
+    def tearDown(self) -> None:
+        try:  # try stopping scheduler (if exists)
+            self.scheduler.stop()
+        except AttributeError:
+            pass
+
+        try:  # try delete temporary repo
+            self.api.delete_repo(self.repo_name)
+        except Exception:
+            pass
+
+    @patch("huggingface_hub._commit_scheduler.CommitScheduler._push_to_hub")
+    def test_mocked_push_to_hub(self, push_to_hub_mock: MagicMock) -> None:
+        self.scheduler = CommitScheduler(
+            folder_path=self.cache_dir,
+            repo_id=self.repo_name,
+            every=1 / 60 / 10,  # every 0.1s
+            hf_api=self.api,
+        )
+        time.sleep(0.25)
+
+        # Triggered 3 times (at 0.0s, 0.1s and 0.2s) with empty args
+        push_to_hub_mock.assert_has_calls([call(), call(), call()])
+
+        # Can get the last upload result
+        self.assertEqual(self.scheduler.last_future.result(), push_to_hub_mock.return_value)
+
+    def test_invalid_folder_path_is_a_file(self) -> None:
+        """Test cannot scheduler upload of a single file."""
+        file_path = self.cache_dir / "file.txt"
+        file_path.write_text("something")
+
+        with self.assertRaises(ValueError):
+            CommitScheduler(folder_path=file_path, repo_id=self.repo_name, hf_api=self.api)
+
+    def test_missing_folder_is_created(self) -> None:
+        folder_path = self.cache_dir / "folder" / "subfolder"
+        self.scheduler = CommitScheduler(folder_path=folder_path, repo_id=self.repo_name, hf_api=self.api)
+        self.assertTrue(folder_path.is_dir())
+
+    def test_sync_local_folder(self) -> None:
+        """Test sync local folder to remote repo."""
+        watched_folder = self.cache_dir / "watched_folder"
+        hub_cache = self.cache_dir / "hub"  # to download hub files
+
+        file_path = watched_folder / "file.txt"
+        lfs_path = watched_folder / "lfs.bin"
+
+        self.scheduler = CommitScheduler(
+            folder_path=watched_folder,
+            repo_id=self.repo_name,
+            every=1 / 60,  # every 1s
+            hf_api=self.api,
+        )
+
+        # 1 push to hub triggered (empty commit not pushed)
+        time.sleep(0.5)
+
+        # write content to files
+        with file_path.open("a") as f:
+            f.write("first line\n")
+        with lfs_path.open("a") as f:
+            f.write("binary content")
+
+        # 2 push to hub triggered (1 commit + 1 ignored)
+        time.sleep(2)
+        self.scheduler.last_future.result()
+
+        # new content in file
+        with file_path.open("a") as f:
+            f.write("second line\n")
+
+        # 1 push to hub triggered (1 commit)
+        time.sleep(1)
+        self.scheduler.last_future.result()
+
+        with lfs_path.open("a") as f:
+            f.write(" updated")
+
+        # 5 push to hub triggered (1 commit)
+        time.sleep(5)  # wait for every threads/uploads to complete
+        self.scheduler.stop()
+        self.scheduler.last_future.result()
+
+        # 4 commits expected (initial commit + 3 push to hub)
+        repo_id = self.scheduler.repo_id
+        commits = self.api.list_repo_commits(repo_id)
+        self.assertEqual(len(commits), 4)
+        push_1 = commits[2].commit_id  # sorted by last first
+        push_2 = commits[1].commit_id
+        push_3 = commits[0].commit_id
+
+        def _download(filename: str, revision: str) -> Path:
+            return Path(hf_hub_download(repo_id=repo_id, filename=filename, cache_dir=hub_cache, revision=revision))
+
+        # Check file.txt consistency
+        file_push1 = _download(filename="file.txt", revision=push_1)
+        file_push2 = _download(filename="file.txt", revision=push_2)
+        file_push3 = _download(filename="file.txt", revision=push_3)
+
+        self.assertEqual(file_push1.read_text(), "first line\n")
+        self.assertEqual(file_push2.read_text(), "first line\nsecond line\n")
+        self.assertEqual(file_push3.read_text(), "first line\nsecond line\n")
+
+        # Check lfs.bin consistency
+        lfs_push1 = _download(filename="lfs.bin", revision=push_1)
+        lfs_push2 = _download(filename="lfs.bin", revision=push_2)
+        lfs_push3 = _download(filename="lfs.bin", revision=push_3)
+
+        self.assertEqual(lfs_push1.read_text(), "binary content")
+        self.assertEqual(lfs_push2.read_text(), "binary content")
+        self.assertEqual(lfs_push3.read_text(), "binary content updated")
+
+
+@pytest.mark.usefixtures("fx_cache_dir")
+class TestPartialFileIO(unittest.TestCase):
+    """Test PartialFileIO object."""
+
+    cache_dir: Path
+
+    def setUp(self) -> None:
+        """Set up a test file."""
+        self.file_path = self.cache_dir / "file.txt"
+        self.file_path.write_text("123456789")  # file size: 9 bytes
+
+    def test_read_partial_file_twice(self) -> None:
+        file = PartialFileIO(self.file_path, size_limit=5)
+        self.assertEqual(file.read(), b"12345")
+        self.assertEqual(file.read(), b"")  # End of file
+
+    def test_read_partial_file_by_chunks(self) -> None:
+        file = PartialFileIO(self.file_path, size_limit=5)
+        self.assertEqual(file.read(2), b"12")
+        self.assertEqual(file.read(2), b"34")
+        self.assertEqual(file.read(2), b"5")
+        self.assertEqual(file.read(2), b"")
+
+    def test_read_partial_file_too_much(self) -> None:
+        file = PartialFileIO(self.file_path, size_limit=5)
+        self.assertEqual(file.read(20), b"12345")
+
+    def test_partial_file_len(self) -> None:
+        """Useful for `requests` internally."""
+        file = PartialFileIO(self.file_path, size_limit=5)
+        self.assertEqual(len(file), 5)
+
+        file = PartialFileIO(self.file_path, size_limit=50)
+        self.assertEqual(len(file), 9)
+
+    def test_partial_file_seek_and_tell(self) -> None:
+        file = PartialFileIO(self.file_path, size_limit=5)
+
+        self.assertEqual(file.tell(), 0)
+
+        file.read(2)
+        self.assertEqual(file.tell(), 2)
+
+        file.seek(0)
+        self.assertEqual(file.tell(), 0)
+
+        file.seek(2)
+        self.assertEqual(file.tell(), 2)
+
+        file.seek(50)
+        self.assertEqual(file.tell(), 5)
+
+        file.seek(-3, SEEK_END)
+        self.assertEqual(file.tell(), 2)  # 5-3
+
+    def test_methods_not_implemented(self) -> None:
+        """Test `PartialFileIO` only implements a subset of the `io` interface. This is on-purpose to avoid misuse."""
+        file = PartialFileIO(self.file_path, size_limit=5)
+
+        with self.assertRaises(NotImplementedError):
+            file.readline()
+
+        with self.assertRaises(NotImplementedError):
+            file.write(b"123")
+
+    def test_append_to_file_then_read(self) -> None:
+        file = PartialFileIO(self.file_path, size_limit=9)
+
+        with self.file_path.open("ab") as f:
+            f.write(b"abcdef")
+
+        # Output is truncated even if new content appended to the wrapped file
+        self.assertEqual(file.read(), b"123456789")
+
+    def test_high_size_limit(self) -> None:
+        file = PartialFileIO(self.file_path, size_limit=20)
+        with self.file_path.open("ab") as f:
+            f.write(b"abcdef")
+
+        # File size limit is truncated to the actual file size at instance creation (not on the fly)
+        self.assertEqual(len(file), 9)
+        self.assertEqual(file._size_limit, 9)
+
+    def test_with_commit_operation_add(self) -> None:
+        # Truncated file
+        op_truncated = CommitOperationAdd(
+            path_or_fileobj=PartialFileIO(self.file_path, size_limit=5), path_in_repo="file.txt"
+        )
+        self.assertEqual(op_truncated.upload_info.size, 5)
+        self.assertEqual(op_truncated.upload_info.sample, b"12345")
+
+        with op_truncated.as_file() as f:
+            self.assertEqual(f.read(), b"12345")
+
+        # Full file
+        op_full = CommitOperationAdd(
+            path_or_fileobj=PartialFileIO(self.file_path, size_limit=9), path_in_repo="file.txt"
+        )
+        self.assertEqual(op_full.upload_info.size, 9)
+        self.assertEqual(op_full.upload_info.sample, b"123456789")
+
+        with op_full.as_file() as f:
+            self.assertEqual(f.read(), b"123456789")
+
+        # Truncated file has a different hash than the full file
+        self.assertNotEqual(op_truncated.upload_info.sha256, op_full.upload_info.sha256)