Netbox CRUD Tasks¤

task api names: crud_list_objects, crud_search, crud_read, crud_create, crud_update, crud_delete, crud_get_changelogs

Seven generic CRUD tasks for any NetBox object type. Designed for AI agents, CLI automation, and programmatic workflows where the full flexibility of the NetBox REST API is needed without hard-coding object-type-specific helper tasks.

Typical workflow:

crud_list_objects — discover what object types exist
crud_search — free-text search to locate objects by name or description
crud_read — fetch full objects by ID or filter
crud_create / crud_update / crud_delete — mutate objects (dry_run=True first)
crud_get_changelogs — verify what changed and when

List Objects¤

task api name: crud_list_objects

Lists all available NetBox object types extracted from the OpenAPI schema. Results are cached for 24 hours (key netbox_{instance}_openapi_objects).

Outputs¤

When include_metadata=True (default):

{
  "dcim": {
    "devices": {
      "path": "/api/dcim/devices/",
      "object_type": "devices",
      "methods": ["GET", "POST", "PUT", "PATCH", "DELETE"],
      "schema_name": "WritableDevice",
      "description": "..."
    }
  }
}

When include_metadata=False:

{
  "dcim": ["cables", "devices", "interfaces", "sites"],
  "ipam": ["ip-addresses", "prefixes"]
}

NORFAB Netbox CRUD List Objects Command Shell Reference¤

NorFab shell supports these command options for Netbox crud_list_objects task:

nf#man tree netbox.crud.list-objects
root
└── netbox:    Netbox service
    └── crud:    Generic CRUD operations on NetBox objects
        └── list-objects:    List available NetBox object types from OpenAPI schema
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── app-filter:    Filter by app(s) e.g. dcim or dcim,ipam (comma-separated)
            └── no-include-metadata:    Return object names only (omit path/methods/schema)
nf#

Examples¤

CLIPython

List all object types with full metadata:

nf#netbox crud list-objects

List object names only (no metadata), filtered to dcim and ipam apps:

nf#netbox crud list-objects no-include-metadata app-filter dcim,ipam

from norfab.core.nfapi import NorFab

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

# all object types with metadata
result = client.run_job(
    "netbox",
    "crud_list_objects",
    workers="any",
)

# names only, filtered to dcim app
result = client.run_job(
    "netbox",
    "crud_list_objects",
    workers="any",
    kwargs={
        "app_filter": ["dcim"],
        "include_metadata": False,
    },
)

nf.destroy()

Python API Reference¤

List all available NetBox object types extracted from OpenAPI schema.

Parameters:

Name	Type	Description	Default
`job`	`Job`	NorFab Job object	required
`instance`	`Union[None, str]`	NetBox instance name; uses default if omitted	`None`
`app_filter`	`Union[None, str, list]`	str or list to filter by app(s) e.g. "dcim" or ["dcim", "ipam"]	`None`
`include_metadata`	`bool`	if False returns object names only; if True includes path, methods (GET/POST/PATCH/PUT/DELETE), schema_name, description	`True`

Returns:

Type	Description
`Result`	dict keyed by app → object_type → {path, object_type, methods, schema_name, description}

Source code in norfab\workers\netbox_worker\netbox_crud.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def crud_list_objects(
    self,
    job: Job,
    instance: Union[None, str] = None,
    app_filter: Union[None, str, list] = None,
    include_metadata: bool = True,
) -> Result:
    """
    List all available NetBox object types extracted from OpenAPI schema.

    Args:
        job: NorFab Job object
        instance: NetBox instance name; uses default if omitted
        app_filter: str or list to filter by app(s) e.g. "dcim" or ["dcim", "ipam"]
        include_metadata: if False returns object names only; if True includes path,
            methods (GET/POST/PATCH/PUT/DELETE), schema_name, description

    Returns:
        dict keyed by app → object_type → {path, object_type, methods, schema_name, description}
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:crud_list_objects",
        result={},
        resources=[instance],
    )
    cache_key = f"netbox_{instance}_openapi_objects"

    # return cached result if available
    if cache_key in self.cache:
        schema_data = self.cache[cache_key]
        log.info(
            f"{self.name} - Serving OpenAPI objects list from cache for '{instance}'"
        )
    else:
        job.event(
            f"extracting object types from OpenAPI schema for instance '{instance}'"
        )
        nb = self._get_pynetbox(instance)
        schema = nb.openapi()
        schema_data: Dict[str, Dict[str, Any]] = {}

        for path, path_spec in schema.get("paths", {}).items():
            match = _OPENAPI_PATH_RE.match(path)
            if not match:
                continue

            app = match.group(1)
            object_type = match.group(2)
            methods = [
                m.upper()
                for m in ("get", "post", "put", "patch", "delete")
                if m in path_spec
            ]
            schema_name = _schema_name_from_path_spec(path_spec)
            description = path_spec.get("get", {}).get("description", "")

            schema_data.setdefault(app, {})[object_type] = {
                "path": path,
                "object_type": object_type,
                "methods": methods,
                "schema_name": schema_name,
                "description": description,
            }

        count = sum(len(obj_types) for obj_types in schema_data.values())
        job.event(f"retrieved {count} object types")
        log.info(
            f"{self.name} - Retrieved {count} object types from OpenAPI schema"
            f" for '{instance}'"
        )

        # cache for 24 hours
        self.cache.set(cache_key, schema_data, expire=OPENAPI_CACHE_TTL)

    # apply app_filter
    if app_filter:
        if isinstance(app_filter, str):
            app_filter = [app_filter]
        schema_data = {
            app: obj_types
            for app, obj_types in schema_data.items()
            if app in app_filter
        }

    # sort alphabetically by app then object_type
    result: Dict[str, Any] = {}
    for app in sorted(schema_data.keys()):
        result[app] = dict(sorted(schema_data[app].items()))

    # strip metadata if not requested
    if not include_metadata:
        for app in result:
            result[app] = list(result[app].keys())

    ret.result = result
    return ret

Search¤

task api name: crud_search

Free-text search using the q parameter across multiple object types simultaneously. Results are grouped by object type. Per-type errors are logged as warnings and the search continues — the task is error-resilient across object types.

Default object types searched when object_types is omitted: dcim.devices, dcim.sites, ipam.ip-addresses, ipam.prefixes, dcim.interfaces, circuits.circuits, tenancy.tenants, virtualization.virtual-machines.

Outputs¤

{
  "dcim.devices": [
    {"id": 1, "name": "ceos1", "status": {"value": "active"}}
  ],
  "ipam.prefixes": []
}

NORFAB Netbox CRUD Search Command Shell Reference¤

NorFab shell supports these command options for Netbox crud_search task:

nf#man tree netbox.crud.search
root
└── netbox:    Netbox service
    └── crud:    Generic CRUD operations on NetBox objects
        └── search:    Free-text search across NetBox object types
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── *query:    Search term
            ├── object-types:    Comma-separated app.resource types to search e.g. dcim.devices,ipam.prefixes
            ├── fields:    Comma-separated fields to return
            ├── brief:    Return brief representation
            └── limit:    Max results per object type, default 10
nf#

Examples¤

CLIPython

Search for "ceos" across default object types:

nf#netbox crud search ceos

Search within specific object types and return selected fields:

nf#netbox crud search ceos object-types dcim.devices,dcim.interfaces fields id,name limit 5

from norfab.core.nfapi import NorFab

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

# search across default object types
result = client.run_job(
    "netbox",
    "crud_search",
    workers="any",
    kwargs={"query": "ceos"},
)

# search in specific types, restrict fields and limit results
result = client.run_job(
    "netbox",
    "crud_search",
    workers="any",
    kwargs={
        "query": "ceos",
        "object_types": ["dcim.devices", "dcim.interfaces"],
        "fields": ["id", "name"],
        "limit": 5,
    },
)

nf.destroy()

Python API Reference¤

Free-text search using the 'q' parameter across multiple object types.

Results are grouped by object_type. Per-type errors are logged as warnings and the search continues (error-resilient).

Parameters:

Name	Type	Description	Default
`job`	`Job`	NorFab Job object	required
`query`	`str`	search term; NetBox searches across all indexed fields	required
`object_types`	`Union[None, list]`	list of "app.resource" strings to search; uses a sensible default set when omitted	`None`
`fields`	`Union[None, list]`	list of specific fields to return; ignored when brief=True	`None`
`brief`	`bool`	if True adds brief=1 to request	`False`
`limit`	`int`	max results per object type (1-100)	`10`
`instance`	`Union[None, str]`	NetBox instance name; uses default if omitted	`None`

Returns:

Type	Description
`Result`	dict keyed by "app.resource" → list of matching objects

Source code in norfab\workers\netbox_worker\netbox_crud.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def crud_search(
    self,
    job: Job,
    query: str,
    object_types: Union[None, list] = None,
    fields: Union[None, list] = None,
    brief: bool = False,
    limit: int = 10,
    instance: Union[None, str] = None,
) -> Result:
    """
    Free-text search using the 'q' parameter across multiple object types.

    Results are grouped by object_type.  Per-type errors are logged as
    warnings and the search continues (error-resilient).

    Args:
        job: NorFab Job object
        query: search term; NetBox searches across all indexed fields
        object_types: list of "app.resource" strings to search; uses a
            sensible default set when omitted
        fields: list of specific fields to return; ignored when brief=True
        brief: if True adds brief=1 to request
        limit: max results per object type (1-100)
        instance: NetBox instance name; uses default if omitted

    Returns:
        dict keyed by "app.resource" → list of matching objects
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:crud_search",
        result={},
        resources=[instance],
    )

    if object_types is None:
        object_types = list(self._CRUD_SEARCH_DEFAULT_OBJECT_TYPES)

    job.event(f"searching '{query}' across {len(object_types)} object type(s)")

    nb = self._get_pynetbox(instance)

    for object_type in object_types:
        try:
            accessor = _get_pynetbox_accessor(nb, object_type)
            params: Dict[str, Any] = {"q": query, "limit": limit}
            if brief:
                params["brief"] = 1
            elif fields:
                params["fields"] = ",".join(fields)

            found = list(itertools.islice(accessor.filter(**params), limit))
            ret.result[object_type] = [dict(obj) for obj in found]
        except Exception as exc:
            log.warning(
                f"{self.name} - crud_search: failed to search '{object_type}':"
                f" {exc}"
            )
            ret.result[object_type] = []

    total = sum(len(v) for v in ret.result.values())
    job.event(f"search completed: {total} matches found")
    log.info(f"{self.name} - crud_search '{query}' completed: {total} matches")

    return ret

Get (Read)¤

task api name: crud_read

Retrieve NetBox objects by ID(s) or filter dict(s). When multiple filter dicts are provided, results are merged and de-duplicated by id.

Outputs¤

{
  "count": 2,
  "next": null,
  "previous": null,
  "results": [
    {"id": 1, "name": "ceos1", "status": {"value": "active"}},
    {"id": 2, "name": "ceos2", "status": {"value": "active"}}
  ]
}

NORFAB Netbox CRUD Get Command Shell Reference¤

NorFab shell supports these command options for Netbox crud_read task:

nf#man tree netbox.crud.get
root
└── netbox:    Netbox service
    └── crud:    Generic CRUD operations on NetBox objects
        └── get:    Retrieve NetBox objects by ID or filter
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── *object-type:    Object type e.g. "dcim.devices"
            ├── object-id:    Comma-separated integer ID(s)
            ├── filters:    JSON filter dict or list of dicts e.g. '{"name":"ceos1"}'
            ├── fields:    Comma-separated fields to return
            ├── brief:    Return brief representation
            ├── limit:    Page size, default 50
            ├── offset:    Pagination skip count, default 0
            └── ordering:    Comma-separated ordering fields, prefix "-" for descending
nf#

Examples¤

CLIPython

Retrieve a single device by name filter:

nf#netbox crud get object-type dcim.devices filters '{"name":"ceos1"}' fields id,name,status

Retrieve multiple devices by ID:

nf#netbox crud get object-type dcim.devices object-id 1,2,3

Paginate through all devices sorted by name:

nf#netbox crud get object-type dcim.devices limit 50 offset 0 ordering name

from norfab.core.nfapi import NorFab

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

# single filter dict
result = client.run_job(
    "netbox",
    "crud_read",
    workers="any",
    kwargs={
        "object_type": "dcim.devices",
        "filters": {"name": "ceos1"},
        "fields": ["id", "name", "status", "site"],
    },
)

# multiple IDs
result = client.run_job(
    "netbox",
    "crud_read",
    workers="any",
    kwargs={
        "object_type": "dcim.devices",
        "object_id": [1, 2, 3],
    },
)

nf.destroy()

Python API Reference¤

Retrieve NetBox objects by ID(s) or filter dict(s).

Parameters:

Name	Type	Description	Default
`job`	`Job`	NorFab Job object	required
`object_type`	`str`	"app.resource" e.g. "dcim.devices"	required
`object_id`	`Union[None, int, list]`	int or list[int]; when set, filters is ignored	`None`
`filters`	`Union[None, dict, list]`	dict or list[dict]; list runs multiple queries merged into one result set	`None`
`fields`	`Union[None, list]`	list of specific fields to return; ignored when brief=True	`None`
`brief`	`bool`	if True adds brief=1	`False`
`limit`	`int`	page size (1-1000)	`50`
`offset`	`int`	pagination skip count	`0`
`ordering`	`Union[None, str, list]`	str or list[str]; prefix with '-' for descending	`None`

Returns:

Type	Description
`Result`	{count, next, previous, results: [...]}

Source code in norfab\workers\netbox_worker\netbox_crud.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def crud_read(
    self,
    job: Job,
    object_type: str,
    object_id: Union[None, int, list] = None,
    filters: Union[None, dict, list] = None,
    fields: Union[None, list] = None,
    brief: bool = False,
    limit: int = 50,
    offset: int = 0,
    ordering: Union[None, str, list] = None,
    instance: Union[None, str] = None,
) -> Result:
    """
    Retrieve NetBox objects by ID(s) or filter dict(s).

    Args:
        job: NorFab Job object
        object_type: "app.resource" e.g. "dcim.devices"
        object_id: int or list[int]; when set, filters is ignored
        filters: dict or list[dict]; list runs multiple queries merged into one result set
        fields: list of specific fields to return; ignored when brief=True
        brief: if True adds brief=1
        limit: page size (1-1000)
        offset: pagination skip count
        ordering: str or list[str]; prefix with '-' for descending

    Returns:
        {count, next, previous, results: [...]}
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:crud_read",
        result={},
        resources=[instance],
    )

    nb = self._get_pynetbox(instance)
    accessor = _get_pynetbox_accessor(nb, object_type)

    # build common params
    params: Dict[str, Any] = {"limit": limit, "offset": offset}
    if brief:
        params["brief"] = 1
    elif fields:
        params["fields"] = ",".join(fields)
    if ordering:
        if isinstance(ordering, list):
            params["ordering"] = ",".join(ordering)
        else:
            params["ordering"] = ordering

    results = []

    if object_id is not None:
        if isinstance(object_id, int):
            job.event(f"retrieving {object_type} by ID(s)")
            obj = accessor.get(object_id)
            results = [dict(obj)] if obj else []
        else:
            job.event(f"retrieving {object_type} by ID(s)")
            found = list(accessor.filter(id=list(object_id), **params))
            results = [dict(o) for o in found]
    elif filters is not None:
        if isinstance(filters, dict):
            job.event(f"retrieving {object_type} with 1 filter(s)")
            found = list(accessor.filter(**filters, **params))
            results = [dict(o) for o in found]
        else:
            # list of dicts — run one filter per dict and merge
            job.event(f"retrieving {object_type} with {len(filters)} filter(s)")
            seen_ids: set = set()
            for flt in filters:
                for obj in accessor.filter(**flt, **params):
                    obj_dict = dict(obj)
                    obj_id_ = obj_dict.get("id")
                    if obj_id_ not in seen_ids:
                        seen_ids.add(obj_id_)
                        results.append(obj_dict)
    else:
        job.event(f"retrieving {object_type} with 0 filter(s)")
        found = list(accessor.filter(**params))
        results = [dict(o) for o in found]

    job.event(f"retrieved {len(results)} total objects")
    log.info(
        f"{self.name} - crud_read '{object_type}': retrieved {len(results)} objects"
    )

    ret.result = {
        "count": len(results),
        "next": None,
        "previous": None,
        "results": results,
    }
    return ret

Create¤

task api name: crud_create

Create one or multiple NetBox objects in a single request. Pass a single dict for one object or a list of dicts for bulk creation. Use dry_run=True to preview the payload without writing to NetBox.

Outputs¤

Normal:

{
  "created": 2,
  "objects": [
    {"id": 10, "name": "Acme", "slug": "acme"},
    {"id": 11, "name": "Initech", "slug": "initech"}
  ]
}

Dry run:

{
  "dry_run": true,
  "count": 2,
  "preview": [
    {"name": "Acme", "slug": "acme"},
    {"name": "Initech", "slug": "initech"}
  ]
}

NORFAB Netbox CRUD Create Command Shell Reference¤

NorFab shell supports these command options for Netbox crud_create task:

nf#man tree netbox.crud.create
root
└── netbox:    Netbox service
    └── crud:    Generic CRUD operations on NetBox objects
        └── create:    Create one or multiple NetBox objects
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── *object-type:    Object type e.g. "dcim.manufacturers"
            ├── *data:    JSON dict or list of dicts with object field values
            └── dry-run:    Preview without creating
nf#

Examples¤

CLIPython

Create a single manufacturer (dry run first):

nf#netbox crud create object-type dcim.manufacturers data '{"name":"Acme","slug":"acme"}' dry-run
nf#netbox crud create object-type dcim.manufacturers data '{"name":"Acme","slug":"acme"}'

Bulk-create interfaces:

nf#netbox crud create object-type dcim.interfaces data '[{"device":1,"name":"eth0","type":"1000base-t"},{"device":1,"name":"eth1","type":"1000base-t"}]'

result = client.run_job(
    "netbox",
    "crud_create",
    workers="any",
    kwargs={
        "object_type": "dcim.manufacturers",
        "data": {"name": "Acme", "slug": "acme"},
        "dry_run": True,
    },
)

Python API Reference¤

Create one or multiple NetBox objects in a single request.

Parameters:

Name	Type	Description	Default
`job`	`Job`	NorFab Job object	required
`object_type`	`str`	"app.resource" e.g. "dcim.interfaces"	required
`data`	`Union[dict, list]`	dict (single) or list[dict] (bulk); normalized to list internally	required
`instance`	`Union[None, str]`	NetBox instance name; uses default if omitted	`None`
`dry_run`	`bool`	if True returns input data without calling NetBox	`False`

Returns:

Type	Description
`Result`	Normal: {created: N, objects: [...]}
`Result`	dry_run: {dry_run: True, count: N, preview: [...input data...]}

Source code in norfab\workers\netbox_worker\netbox_crud.py

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def crud_create(
    self,
    job: Job,
    object_type: str,
    data: Union[dict, list],
    instance: Union[None, str] = None,
    dry_run: bool = False,
) -> Result:
    """
    Create one or multiple NetBox objects in a single request.

    Args:
        job: NorFab Job object
        object_type: "app.resource" e.g. "dcim.interfaces"
        data: dict (single) or list[dict] (bulk); normalized to list internally
        instance: NetBox instance name; uses default if omitted
        dry_run: if True returns input data without calling NetBox

    Returns:
        - Normal: {created: N, objects: [...]}
        - dry_run: {dry_run: True, count: N, preview: [...input data...]}
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:crud_create",
        result={},
        resources=[instance],
    )

    # normalise to list
    if isinstance(data, dict):
        data_list = [data]
    else:
        data_list = list(data)

    if dry_run:
        job.event(f"dry-run: would create {len(data_list)} {object_type}(s)")
        log.info(
            f"{self.name} - crud_create dry-run: {len(data_list)}"
            f" {object_type}(s) would be created"
        )
        ret.result = {
            "dry_run": True,
            "count": len(data_list),
            "preview": data_list,
        }
        return ret

    job.event(f"creating {len(data_list)} {object_type}(s)")

    nb = self._get_pynetbox(instance)
    accessor = _get_pynetbox_accessor(nb, object_type)

    created = accessor.create(data_list)
    # pynetbox returns a Record for single or list for bulk
    if not isinstance(created, list):
        created = [created]

    created_dicts = [dict(obj) for obj in created]
    log.info(
        f"{self.name} - crud_create '{object_type}':"
        f" created {len(created_dicts)} object(s)"
    )

    ret.result = {"created": len(created_dicts), "objects": created_dicts}
    return ret

Update¤

task api name: crud_update

Update one or multiple NetBox objects. Each item in data must contain an "id" field. By default uses PATCH (partial update — only specified fields are changed). Set partial=False to use PUT (full replace — omitted fields revert to their defaults).

Use dry_run=True to compute field-level diffs without modifying anything.

Outputs¤

Normal:

{
  "updated": 1,
  "objects": [
    {"id": 10, "name": "Acme Corp", "slug": "acme"}
  ]
}

Dry run:

{
  "dry_run": true,
  "count": 1,
  "changes": [
    {
      "id": 10,
      "changes": {
        "name": {"old": "Acme", "new": "Acme Corp"}
      }
    }
  ]
}

NORFAB Netbox CRUD Update Command Shell Reference¤

NorFab shell supports these command options for Netbox crud_update task:

nf#man tree netbox.crud.update
root
└── netbox:    Netbox service
    └── crud:    Generic CRUD operations on NetBox objects
        └── update:    Update one or multiple NetBox objects
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── *object-type:    Object type e.g. "dcim.manufacturers"
            ├── *data:    JSON dict or list of dicts; each must contain "id"
            ├── no-partial:    Use PUT (full replace) instead of PATCH (partial update)
            └── dry-run:    Show diffs without updating
nf#

Examples¤

CLIPython

Preview changes before applying:

nf#netbox crud update object-type dcim.manufacturers data '{"id":10,"name":"Acme Corp"}' dry-run
nf#netbox crud update object-type dcim.manufacturers data '{"id":10,"name":"Acme Corp"}'

# dry run first
result = client.run_job(
    "netbox",
    "crud_update",
    workers="any",
    kwargs={
        "object_type": "dcim.manufacturers",
        "data": {"id": 10, "name": "Acme Corp"},
        "dry_run": True,
    },
)

# apply
result = client.run_job(
    "netbox",
    "crud_update",
    workers="any",
    kwargs={
        "object_type": "dcim.manufacturers",
        "data": {"id": 10, "name": "Acme Corp"},
    },
)

Python API Reference¤

Update one or multiple NetBox objects. Each item in data MUST include "id".

Parameters:

Name	Type	Description	Default
`job`	`Job`	NorFab Job object	required
`object_type`	`str`	"app.resource"	required
`data`	`Union[dict, list]`	dict (single) or list[dict]; each must contain "id" field	required
`partial`	`bool`	True → PATCH (only specified fields); False → PUT (full replace)	`True`
`instance`	`Union[None, str]`	NetBox instance name; uses default if omitted	`None`
`dry_run`	`bool`	if True fetches current state, computes diffs, returns without modifying	`False`

Returns:

Type	Description
`Result`	Normal: {updated: N, objects: [...]}
`Result`	dry_run: {dry_run: True, count: N, changes: [{id, changes: {field: {old, new}}}]}

Source code in norfab\workers\netbox_worker\netbox_crud.py

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def crud_update(
    self,
    job: Job,
    object_type: str,
    data: Union[dict, list],
    partial: bool = True,
    instance: Union[None, str] = None,
    dry_run: bool = False,
) -> Result:
    """
    Update one or multiple NetBox objects. Each item in data MUST include "id".

    Args:
        job: NorFab Job object
        object_type: "app.resource"
        data: dict (single) or list[dict]; each must contain "id" field
        partial: True → PATCH (only specified fields); False → PUT (full replace)
        instance: NetBox instance name; uses default if omitted
        dry_run: if True fetches current state, computes diffs, returns without modifying

    Returns:
        - Normal: {updated: N, objects: [...]}
        - dry_run: {dry_run: True, count: N, changes: [{id, changes: {field: {old, new}}}]}
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:crud_update",
        result={},
        resources=[instance],
    )

    # normalise to list
    if isinstance(data, dict):
        data_list = [data]
    else:
        data_list = list(data)

    # validate that each item has an "id"
    for item in data_list:
        if "id" not in item:
            raise ValueError(
                f"crud_update: each data item must contain 'id', got: {item}"
            )

    nb = self._get_pynetbox(instance)
    accessor = _get_pynetbox_accessor(nb, object_type)

    if dry_run:
        job.event(f"dry-run: computing diffs for {len(data_list)} {object_type}(s)")
        diffs = []
        for item in data_list:
            obj_id = item["id"]
            current = accessor.get(obj_id)
            current_dict = dict(current) if current else {}
            changes: Dict[str, Any] = {}
            for field, new_val in item.items():
                if field == "id":
                    continue
                old_val = current_dict.get(field)
                if old_val != new_val:
                    changes[field] = {"old": old_val, "new": new_val}
            diffs.append({"id": obj_id, "changes": changes})
        log.info(
            f"{self.name} - crud_update dry-run '{object_type}':"
            f" computed diffs for {len(diffs)} object(s)"
        )
        ret.result = {
            "dry_run": True,
            "count": len(diffs),
            "changes": diffs,
        }
        return ret

    job.event(f"updating {len(data_list)} {object_type}(s)")

    updated_dicts = []
    for item in data_list:
        obj_id = item["id"]
        obj = accessor.get(obj_id)
        if obj:
            if partial:
                obj.update(item)
            else:
                # full PUT: save the entire item dict
                obj.update(item)
                obj.save()
            updated_dicts.append(dict(obj))

    log.info(
        f"{self.name} - crud_update '{object_type}':"
        f" updated {len(updated_dicts)} object(s)"
    )

    ret.result = {"updated": len(updated_dicts), "objects": updated_dicts}
    return ret

Delete¤

task api name: crud_delete

Delete one or multiple NetBox objects by ID. Use dry_run=True to preview which objects would be deleted without removing them.

Outputs¤

Normal:

{
  "deleted": 2,
  "deleted_ids": [10, 11]
}

Dry run:

{
  "dry_run": true,
  "count": 2,
  "would_delete": [
    {"id": 10, "name": "Acme", "slug": "acme"},
    {"id": 11, "name": "Initech", "slug": "initech"}
  ]
}

NORFAB Netbox CRUD Delete Command Shell Reference¤

NorFab shell supports these command options for Netbox crud_delete task:

nf#man tree netbox.crud.delete
root
└── netbox:    Netbox service
    └── crud:    Generic CRUD operations on NetBox objects
        └── delete:    Delete one or multiple NetBox objects by ID
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── *object-type:    Object type e.g. "dcim.manufacturers"
            ├── *object-id:    Comma-separated integer ID(s) to delete
            └── dry-run:    Preview objects that would be deleted
nf#

Examples¤

CLIPython

Preview then delete:

nf#netbox crud delete object-type dcim.manufacturers object-id 10,11 dry-run
nf#netbox crud delete object-type dcim.manufacturers object-id 10,11

result = client.run_job(
    "netbox",
    "crud_delete",
    workers="any",
    kwargs={
        "object_type": "dcim.manufacturers",
        "object_id": [10, 11],
        "dry_run": True,
    },
)

Python API Reference¤

Delete one or multiple NetBox objects by ID.

Parameters:

Name	Type	Description	Default
`job`	`Job`	NorFab Job object	required
`object_type`	`str`	"app.resource"	required
`object_id`	`Union[int, list]`	int (single) or list[int] (bulk)	required
`instance`	`Union[None, str]`	NetBox instance name; uses default if omitted	`None`
`dry_run`	`bool`	if True fetches and returns objects that would be deleted	`False`

Returns:

Type	Description
`Result`	Normal: {deleted: N, deleted_ids: [...]}
`Result`	dry_run: {dry_run: True, count: N, would_delete: [...objects...]}

Source code in norfab\workers\netbox_worker\netbox_crud.py

@Task(
    fastapi={"methods": ["DELETE"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def crud_delete(
    self,
    job: Job,
    object_type: str,
    object_id: Union[int, list],
    instance: Union[None, str] = None,
    dry_run: bool = False,
) -> Result:
    """
    Delete one or multiple NetBox objects by ID.

    Args:
        job: NorFab Job object
        object_type: "app.resource"
        object_id: int (single) or list[int] (bulk)
        instance: NetBox instance name; uses default if omitted
        dry_run: if True fetches and returns objects that would be deleted

    Returns:
        - Normal: {deleted: N, deleted_ids: [...]}
        - dry_run: {dry_run: True, count: N, would_delete: [...objects...]}
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:crud_delete",
        result={},
        resources=[instance],
    )

    # normalise to list
    if isinstance(object_id, int):
        id_list = [object_id]
    else:
        id_list = list(object_id)

    nb = self._get_pynetbox(instance)
    accessor = _get_pynetbox_accessor(nb, object_type)

    if dry_run:
        job.event(f"dry-run: would delete {len(id_list)} {object_type}(s)")
        preview = []
        for oid in id_list:
            obj = accessor.get(oid)
            if obj:
                preview.append(dict(obj))
        log.info(
            f"{self.name} - crud_delete dry-run '{object_type}':"
            f" would delete {len(id_list)} object(s)"
        )
        ret.result = {
            "dry_run": True,
            "count": len(preview),
            "would_delete": preview,
        }
        return ret

    job.event(f"deleting {len(id_list)} {object_type}(s)")

    deleted_ids = []
    for oid in id_list:
        obj = accessor.get(oid)
        if obj:
            obj.delete()
            deleted_ids.append(oid)

    log.info(
        f"{self.name} - crud_delete '{object_type}':"
        f" deleted {len(deleted_ids)} object(s)"
    )

    ret.result = {"deleted": len(deleted_ids), "deleted_ids": deleted_ids}
    return ret

Get Changelogs¤

task api name: crud_get_changelogs

Retrieve NetBox change history from the object-changes endpoint. Supports NetBox 4.0+ (endpoint moved from extras to core) with automatic fallback to older versions.

Outputs¤

{
  "count": 1,
  "next": null,
  "previous": null,
  "results": [
    {
      "id": 42,
      "user_name": "admin",
      "request_id": "abc-123",
      "action": {"value": "create", "label": "Created"},
      "changed_object_type": "dcim.manufacturer",
      "changed_object_id": 10,
      "object_repr": "Acme",
      "time": "2026-04-06T12:00:00Z",
      "prechange_data": null,
      "postchange_data": {"name": "Acme", "slug": "acme"}
    }
  ]
}

NORFAB Netbox CRUD Changelogs Command Shell Reference¤

NorFab shell supports these command options for Netbox crud_get_changelogs task:

nf#man tree netbox.crud.changelogs
root
└── netbox:    Netbox service
    └── crud:    Generic CRUD operations on NetBox objects
        └── changelogs:    Retrieve NetBox change history
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── filters:    JSON filter dict or list of dicts
            ├── fields:    Comma-separated fields to return
            ├── limit:    Page size, default 50
            └── offset:    Pagination skip count, default 0
nf#

Examples¤

CLIPython

Recent create actions:

nf#netbox crud changelogs filters '{"action":"create"}' limit 20

Changes to a specific object:

nf#netbox crud changelogs filters '{"changed_object_id":10}' fields id,action,time,object_repr

result = client.run_job(
    "netbox",
    "crud_get_changelogs",
    workers="any",
    kwargs={
        "filters": {"action": "create"},
        "fields": ["id", "action", "time", "object_repr", "user_name"],
        "limit": 20,
    },
)

Python API Reference¤

Retrieve NetBox change history from the extras/object-changes endpoint.

Parameters:

Name	Type	Description	Default
`job`	`Job`	NorFab Job object	required
`filters`	`Union[None, dict, list]`	dict or list[dict]; supported filter keys include user_id, user, changed_object_id, changed_object_type_id, object_repr, action, time_before, time_after (ISO-8601), q	`None`
`fields`	`Union[None, list]`	list of fields to return	`None`
`limit`	`int`	page size (1-1000)	`50`
`offset`	`int`	pagination skip count	`0`
`instance`	`Union[None, str]`	NetBox instance name; uses default if omitted	`None`

Returns:

Type	Description
`Result`	{count, next, previous, results: [{id, user, user_name, request_id, action,
`Result`	changed_object_type, changed_object_id, object_repr, time,
`Result`	prechange_data, postchange_data}]}

Source code in norfab\workers\netbox_worker\netbox_crud.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def crud_get_changelogs(
    self,
    job: Job,
    filters: Union[None, dict, list] = None,
    fields: Union[None, list] = None,
    limit: int = 50,
    offset: int = 0,
    instance: Union[None, str] = None,
) -> Result:
    """
    Retrieve NetBox change history from the extras/object-changes endpoint.

    Args:
        job: NorFab Job object
        filters: dict or list[dict]; supported filter keys include user_id, user,
            changed_object_id, changed_object_type_id, object_repr, action,
            time_before, time_after (ISO-8601), q
        fields: list of fields to return
        limit: page size (1-1000)
        offset: pagination skip count
        instance: NetBox instance name; uses default if omitted

    Returns:
        {count, next, previous, results: [{id, user, user_name, request_id, action,
        changed_object_type, changed_object_id, object_repr, time,
        prechange_data, postchange_data}]}
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:crud_get_changelogs",
        result={},
        resources=[instance],
    )

    filter_count = (
        1 if isinstance(filters, dict) else (len(filters) if filters else 0)
    )
    job.event(f"retrieving changelogs with {filter_count} filter(s)")

    nb = self._get_pynetbox(instance)

    # build base params
    base_params: Dict[str, Any] = {"limit": limit, "offset": offset}
    if fields:
        base_params["fields"] = ",".join(fields)

    results = []

    if filters is None or isinstance(filters, dict):
        params = {**base_params}
        if filters:
            params.update(filters)
        try:
            # NetBox 4.0+: ObjectChange moved from extras to core
            found = list(nb.core.object_changes.filter(**params))
        except Exception:
            found = list(nb.extras.object_changes.filter(**params))
        results = [dict(obj) for obj in found]
    else:
        # list of filter dicts — run multiple queries
        seen_ids: set = set()
        for flt in filters:
            params = {**base_params, **flt}
            try:
                changelog_iter = nb.core.object_changes.filter(**params)
            except Exception:
                changelog_iter = nb.extras.object_changes.filter(**params)
            for obj in changelog_iter:
                obj_dict = dict(obj)
                oid = obj_dict.get("id")
                if oid not in seen_ids:
                    seen_ids.add(oid)
                    results.append(obj_dict)

    job.event(f"retrieved {len(results)} changelog entries")
    log.info(f"{self.name} - crud_get_changelogs: retrieved {len(results)} entries")

    ret.result = {
        "count": len(results),
        "next": None,
        "previous": None,
        "results": results,
    }
    return ret