Filesharing Resolve Git URL Task

task api name: resolve_git_url

The resolve_git_url task validates a git:// file URL, synchronizes its configured remote, and returns the file's published nf:// URL. The task does not stream file content. NFPClient.fetch_file() calls it automatically before using the existing File Sharing transfer flow.

Inputs

Parameter	Required	Description
url	Yes	File URL in git://<remote-name>/<path> format

Output

Successful resolution returns an nf:// URL using the remote's configured mount:

"nf://repositories/network-assets/templates/base.j2"

Invalid URLs, unknown remotes, paths outside the selected mount, missing files, and Git synchronization failures return a failed result.

Examples

CLIPython

Use the client-level file command, which resolves the URL automatically:

nf#file copy url git://network-assets/templates/base.j2 read

Direct task invocation:

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()
    result = client.run_job(
        service="filesharing",
        task="resolve_git_url",
        workers="filesharing-worker-1",
        kwargs={"url": "git://network-assets/templates/base.j2"},
    )
    print(result)

Normal file retrieval:

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()
    result = client.fetch_file(
        "git://network-assets/templates/base.j2",
        read=True,
    )
    print(result["content"])

Notes

• The remote must already be configured or registered with create_remote_git.
• Resolution synchronizes the complete configured branch before returning.
• The returned URL uses the configured mount, which may differ from the remote name.
• Git access and credentials remain inside the File Sharing worker.

Python API Reference

Synchronize a Git remote and resolve its file URL to an nf:// URL.

Parameters:

Name	Type	Description	Default
job	Job	Active worker job used by the Git synchronization task.	required
url	str	File URL in git://<remote-name>/<path> format.	required

Returns:

Type	Description
Result	A result containing the published nf:// file URL. Invalid URLs,
Result	unknown remotes, unsafe paths, and synchronization failures produce
Result	failed results.

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

@Task(
    input=ResolveGitUrlInput,
    output=ResolveGitUrlResult,
    fastapi={"methods": ["POST"]},
    agent={"enabled": False},
    mcp={
        "annotations": {
            "title": "Resolve Git URL",
            "readOnlyHint": False,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def resolve_git_url(self, job: Job, url: str) -> Result:
    """Synchronize a Git remote and resolve its file URL to an nf:// URL.

    Args:
        job: Active worker job used by the Git synchronization task.
        url: File URL in ``git://<remote-name>/<path>`` format.

    Returns:
        A result containing the published ``nf://`` file URL. Invalid URLs,
        unknown remotes, unsafe paths, and synchronization failures produce
        failed results.
    """
    remote_path = url.removeprefix("git://").replace("\\", "/")
    name, separator, file_path = remote_path.partition("/")
    if not url.startswith("git://") or not name or not separator or not file_path:
        return Result(failed=True, errors=[f"'{url}' - invalid Git URL format"])

    remote = self.remotes.get(name)
    if remote is None:
        return Result(failed=True, errors=[f"Remote '{name}' is not configured"])
    if remote["type"] != "git":
        return Result(failed=True, errors=[f"Remote '{name}' is not a git remote"])

    mount_path = self._safe_path(remote["mount"])
    resolved_path = os.path.abspath(os.path.join(mount_path, *file_path.split("/")))
    if os.path.commonpath([mount_path, resolved_path]) != mount_path:
        return Result(failed=True, errors=[f"'{url}' - invalid Git URL path"])

    sync_result = self.git_clone(job, name)
    if sync_result.failed:
        return Result(failed=True, errors=sync_result.errors)
    if not os.path.isfile(resolved_path):
        return Result(failed=True, errors=[f"'{url}' file not found"])

    published_path = os.path.relpath(resolved_path, mount_path).replace(os.sep, "/")
    resolved_url = f"{remote['mount'].rstrip('/')}/{published_path}"
    return Result(result=resolved_url)