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. In most user code, prefer the client helper method NFPClient.fetch_file() because it handles streaming, caching, and local file management.

Inputs

Parameter	Required	Description
url	Yes	File URL to fetch
chunk_size	No	Number of bytes to return from the given offset when invoking the task directly
offset	No	Byte offset for direct task invocation
destination	No	Local destination path when using the CLI/helper
read	No	Return file content as text instead of only downloading

Output

The client helper returns a dictionary whose content value is either the local file path or the file text when read=True. Direct task invocation returns a chunk-oriented response intended for the helper protocol.

Examples

CLIPython

Download a file:

nf#file copy url nf://filesharing/test_file_1.txt destination ./test_file_1.txt

Print file content:

nf#file copy url nf://filesharing/test_file_1.txt read

Context manager - helper method:

from norfab.core.nfapi import NorFab

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

    result = client.fetch_file(url="nf://filesharing/test_file_1.txt")
    local_path = result["content"]
    print(local_path)

Context manager - read content as text:

from norfab.core.nfapi import NorFab

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

    result = client.fetch_file(
        url="nf://filesharing/test_file_1.txt",
        read=True,
    )
    print(result["content"])

Direct lifecycle - direct task invocation:

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
try:
    nf.start()
    client = nf.make_client()

    result = client.run_job(
        service="filesharing",
        task="fetch_file",
        workers="any",
        kwargs={
            "url": "nf://filesharing/test_file_1.txt",
            "chunk_size": 256000,
            "offset": 0,
        },
    )
    print(result)
finally:
    nf.destroy()

Filesharing Fetch File Command Shell Reference

NorFab shell supports these command options for Filesharing fetch_file task:

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#

Python 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