Skip to content

Filesharing Delete Git Remote Task¤

task api name: delete_remote_git

The delete_remote_git task unregisters a runtime Git remote and removes its private repository, staging folder, and shared mount.

Inputs¤

Parameter Required Description
name Yes Registered remote name to delete

Output¤

Successful deletion returns true. An unknown name returns a failed result with no boolean result.

true

Examples¤

nf# filesharing git delete-remote name automation-assets
from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()
    result = client.run_job(
        service="filesharing",
        task="delete_remote_git",
        workers="filesharing-worker-1",
        kwargs={"name": "automation-assets"},
    )
    print(result)

Notes¤

  • Deletion is limited to the registered repository path and safe nf:// mount.
  • Deleting an inventory remote affects the running worker only. It is registered again when the worker restarts.
  • The task waits for an active operation on the same remote, subject to the job timeout.

Task Command Shell Reference¤

nf#man tree filesharing.git.delete-remote

R - required field, M - supports multiline input, D - dynamic key

root
└── filesharing:    File sharing service
    └── git:    Manage Git remotes
        └── delete-remote:    Delete and unregister a Git remote
            ├── timeout:    Job timeout
            ├── workers:    Filter workers to target, default 'all'
            ├── verbose-result:    Control output details, default 'False'
            ├── nowait:    Do not wait for job to complete, default 'False'
            └── name (R):    Configured remote name
nf#

Python API Reference¤

Delete all worker-managed local data for a configured Git remote.

The runtime definition, private repository, shared nf://<mount>/ snapshot, and staging data are removed.

Parameters:

Name Type Description Default
job Job

Active worker job. Internal calls may pass None to avoid waiting when the remote lock is held.

required
name str

Name of a type: git remote defined in File Sharing inventory.

required

Returns:

Type Description
Result

A result containing True after the local data is removed.

Source code in norfab\workers\filesharing_worker\git_tasks.py
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
@Task(
    input=RemoteNameInput,
    output=DeleteRemoteGitResult,
    fastapi={"methods": ["DELETE"]},
    agent={"enabled": False},
    mcp={
        "annotations": {
            "title": "Delete Git Remote",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": False,
        }
    },
)
def delete_remote_git(self, job: Job, name: str) -> Result:
    """Delete all worker-managed local data for a configured Git remote.

    The runtime definition, private repository, shared
    ``nf://<mount>/`` snapshot, and staging data are removed.

    Args:
        job: Active worker job. Internal calls may pass ``None`` to avoid
            waiting when the remote lock is held.
        name: Name of a ``type: git`` remote defined in File Sharing
            inventory.

    Returns:
        A result containing ``True`` after the local data is removed.
    """
    msg = f"{self.name} - deleting Git remote '{name}'"
    log.info(msg)
    if job is not None:
        job.event(msg)

    if name not in self.remotes:
        msg = f"{self.name} - remote '{name}' is not configured"
        log.warning(msg)
        if job is not None:
            job.event(msg, severity="ERROR")
        return Result(failed=True, errors=[f"Remote '{name}' is not configured"])
    remote = self.remotes[name]
    if remote["type"] != "git":
        msg = f"{self.name} - remote '{name}' is not a git remote"
        log.warning(msg)
        if job is not None:
            job.event(msg, severity="ERROR")
        return Result(failed=True, errors=[f"Remote '{name}' is not a git remote"])

    lock = remote["lock"]
    if job is None:
        acquired = lock.acquire(blocking=False)
    else:
        acquired = lock.acquire(
            timeout=job.timeout if job.timeout is not None else -1
        )
    if not acquired:
        msg = f"{self.name} - remote '{name}' is busy"
        log.warning(msg)
        if job is not None:
            job.event(msg, severity="WARNING")
        return Result(failed=True, errors=[f"Remote '{name}' is busy"])

    try:
        # Remove both the public snapshot and the private Git working tree.
        shutil.rmtree(self._safe_path(remote["mount"]), ignore_errors=True)
        shutil.rmtree(os.path.dirname(remote["repository"]), ignore_errors=True)
    finally:
        lock.release()
    self.remotes.pop(name, None)
    msg = f"{self.name} - deleted Git remote '{name}'"
    log.info(msg)
    if job is not None:
        job.event(msg)
    return Result(result=True)