Filesharing Service Fetch File Task

task api name: fetch_file

The fetch_file task streams a file from the File Sharing worker to the client in chunks with offset support. This enables efficient downloading of large files with resume capability. The task is typically called through the client helper method NFPClient.fetch_file() which handles the streaming protocol, caching, and local file management automatically.

Using it from Python

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()

    # Download file (recommended way - uses client helper)
    ret = client.fetch_file(url="nf://filesharing/test_file_1.txt")
    local_path = ret["content"]
    print(local_path)

    # Download and read content as text
    ret = client.fetch_file(url="nf://filesharing/test_file_1.txt", read=True)
    print(ret["content"])  # text content

    # Direct task invocation (not recommended - use client helper instead)
    reply = client.run_job(
        service="filesharing",
        task="fetch_file",
        workers="any",
        kwargs={
            "url": "nf://filesharing/test_file_1.txt",
            "chunk_size": 256000,
            "offset": 0,
        },
    )
    print(reply)

Using it from nfcli

NORFAB CLI exposes File Sharing under the file command group.

nf#man tree file.copy
root
└── file:    File sharing service
    └── copy:    Copy files
        ├── url:    File location, default 'nf://'
        ├── destination:    File location to save downloaded content
        └── read:    Print file content, default 'False'

nf#

API Reference

Fetch a file in chunks with offset support.

Parameters:

Name	Type	Description	Default
url	str	URL path starting with 'nf://' to fetch file from	required
chunk_size	int	Size of chunk to read in bytes (default: 256KB)	256000
offset	int	Number of bytes to offset (default: 0)	0

Returns:

Type	Description
Result	Result containing file chunk bytes or error message

Source code in norfab\workers\filesharing_worker\filesharing_worker.py

@Task(
    input=FetchFileInput,
    output=FetchFileResult,
    fastapi={"methods": ["GET"]},
    agent={"enabled": False},
    mcp={
        "annotations": {
            "title": "Fetch File",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": False,
        }
    },
)
def fetch_file(
    self,
    job: Job,
    url: str,
    chunk_size: int = 256000,
    offset: int = 0,
    chunk_timeout: int = 5,
) -> Result:
    """
    Fetch a file in chunks with offset support.

    Args:
        url: URL path starting with 'nf://' to fetch file from
        chunk_size: Size of chunk to read in bytes (default: 256KB)
        offset: Number of bytes to offset (default: 0)

    Returns:
        Result containing file chunk bytes or error message
    """
    ret = Result(result=None)
    try:
        full_path = self._safe_path(url)
    except ValueError as e:
        ret.failed = True
        ret.errors = [str(e)]
        return ret

    if os.path.exists(full_path):
        size = os.path.getsize(full_path)
        with open(full_path, "rb") as f:
            while True:
                f.seek(offset, os.SEEK_SET)
                chunk = f.read(chunk_size)
                if chunk:
                    job.stream(chunk)
                if f.tell() >= size:
                    break
                client_response = job.wait_client_input(timeout=chunk_timeout)
                if not client_response:
                    raise RuntimeError(
                        f"{self.name}:fetch_file - {chunk_timeout}s chunk timeout reached before received next chunk request from client"
                    )
                offset = client_response["offset"]

        ret.result = True
    else:
        ret.failed = True
        ret.errors = [f"'{url}' file not found"]

    return ret