Netbox Create Prefix Task

task api name: create_prefix

Allocates the next available child prefix from a parent prefix, or updates an existing matching prefix. By default it creates a /30, which is useful for point-to-point subnets.

The task can also be called from Nornir templates through the netbox.create_prefix Jinja2 filter.

Output

Returns the created or updated prefix data. In dry-run mode, returns the candidate prefix that would be allocated. The preview uses the first available range that can fit the requested prefix length.

{
    "prefix": "10.0.0.0/30",
    "description": "leaf-1 to spine-1",
    "status": "active",
    "...": "...",
}

Notes / Gotchas

Warning

create_prefix uses the description argument to find existing prefixes. Use the same description for repeated calls that should refer to the same prefix.

• parent can be a prefix string such as 10.0.0.0/24, a parent prefix description, or a dictionary of pynetbox prefix filters.
• To associate an existing VLAN, supply vlan (VID, 1-4094) and vlan_group (group name) together. A VID and site alone are unsupported; site scopes the prefix and does not identify the VLAN. Missing or ambiguous VLAN/group lookups fail before allocation. Omission preserves an existing association.
• VLAN changes appear in diff.vlan as old/new VID (ID) values, including dry-run previews. Changing VLAN updates the prefix selected by the existing description/scope filters rather than allocating a replacement.
• If vrf is provided, the parent prefix must belong to the same VRF.
• When updating an existing prefix, new tags and new items in list-valued custom fields are added without removing current values. Supplying [] leaves an existing custom-field list unchanged. For custom fields, null clears the value, and a new scalar value replaces the old one.
• Object and multiobject custom fields accept related object names or IDs. The task resolves names using each field's NetBox definition; missing or ambiguous names fail.
• Branch writes require the NetBox Branching Plugin.

Examples

CLIPython

Allocate the next /30 from a parent prefix:

nf#netbox create prefix parent 10.0.0.0/24 description "leaf-1 to spine-1"

Allocate a /31 in a VRF:

nf#netbox create prefix parent 10.0.0.0/24 description "leaf-1 to spine-1" prefixlen 31 vrf PROD

Associate VLAN 100 from an existing group:

nf#netbox create prefix parent 10.0.0.0/24 description "users" vlan 100 vlan_group "LAB VLANS"

Preview the allocation:

nf#netbox create prefix parent 10.0.0.0/24 description "leaf-1 to spine-1" dry-run

Write into a NetBox branch:

nf#netbox create prefix parent 10.0.0.0/24 description "leaf-1 to spine-1" branch my-branch

from norfab.core.nfapi import NorFab

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

# allocate next available /30; VLAN 100 in LAB VLANS must already exist
result = client.run_job(
    "netbox",
    "create_prefix",
    workers="any",
    kwargs={
        "parent": "10.0.0.0/24",
        "description": "leaf-1 to spine-1",
        "vlan": 100,
        "vlan_group": "LAB VLANS",
    },
)

# allocate from a filtered parent prefix
result = client.run_job(
    "netbox",
    "create_prefix",
    workers="any",
    kwargs={
        "parent": {"prefix": "10.0.0.0/24", "site": "lab"},
        "description": "leaf-2 to spine-1",
        "prefixlen": 31,
        "vrf": "PROD",
        "tags": ["automation-managed"],
    },
)

nf.destroy()

NORFAB Netbox Create Prefix Command Shell Reference

NorFab shell supports these command options for Netbox create_prefix task:

nf# man tree netbox.create.prefix
root
└── netbox:    Netbox service
    └── create:    Create objects in Netbox
        └── prefix:    Allocate next available prefix from parent prefix
            ├── instance:    Netbox instance name to target
            ├── dry-run:    Do not commit to database
            ├── parent:    Parent prefix to allocate new prefix from
            ├── description:    Description for new prefix
            ├── prefixlen:    The prefix length of the new prefix, default '30'
            ├── vrf:    Name of the VRF to associate with the prefix
            ├── tags:    List of tags to assign to the prefix
            ├── tenant:    Name of the tenant to associate with the prefix
            ├── comments:    Comments for the prefix
            ├── role:    Role to assign to the prefix
            ├── site:    Name of the site to associate with the prefix
            ├── vlan:    Existing VLAN VID; requires vlan_group
            ├── vlan_group:    Existing VLAN group name; requires vlan
            ├── status:    Status of the prefix
            ├── branch:    Branching plugin branch name to use
            ├── timeout:    Job timeout
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            └── progress:    Display progress events, default 'True'
nf#

Python API Reference

Creates a new IP prefix in NetBox or updates an existing one.

Parameters:

Name	Type	Description	Default
job	Job	NorFab job supplying progress events and execution context.	required
parent	Union[str, dict]	Parent prefix to allocate new prefix from, could be: IPv4 prefix string e.g. 10.0.0.0/24 IPv6 prefix string e.g. 2001::/64 Prefix description string to filter by Dictionary with prefix filters for pynetbox prefixes.get method e.g. {"prefix": "10.0.0.0/24", "site": "foo"}	required
custom_fields	dict	Custom-field names and values to PATCH on the allocated prefix. New list items are added to existing custom-field lists; scalar values and null replace old values. Object and multiobject fields accept related object names or IDs. Definitions must already exist in NetBox. The result diff lists changed field names only, not their values.	None
description	str	Description for the new prefix, prefix description used for deduplication to source existing prefixes.	None
prefixlen	int	The prefix length of the new prefix to create, by default allocates next available /30 point-to-point prefix.	30
vrf	str	Name of the VRF to associate with the prefix.	None
tags	Union[None, list]	Tag names to add to the prefix.	None
tenant	str	Name of the tenant to associate with the prefix.	None
comments	str	Comments for the prefix.	None
role	str	Role to assign to the prefix.	None
site	str	Name of the site to associate with the prefix.	None
vlan	int	Existing VLAN VID (1-4094), identified together with vlan_group. Omission preserves the current association. Site cannot substitute for the VLAN group.	None
vlan_group	str	Existing VLAN group name. Must be supplied together with vlan; neither the group nor VLAN is created.	None
status	str	Status of the prefix.	None
instance	Union[None, str]	NetBox instance identifier.	None
dry_run	bool	If True, simulates the creation without making changes.	False
branch	str	Branch name to use, need to have branching plugin installed, automatically creates branch if it does not exist in Netbox.	None

Returns:

Name	Type	Description
Result	Result	Prefix details, status, resources, and changed fields. VLAN changes use old/new "VID (ID)" values in diff["vlan"]. Dry runs report the planned association without writing it.

Raises:

Type	Description
NetboxAllocationError	The parent or requested VLAN/group cannot be resolved uniquely, allocation fails, or an existing prefix has a different length. VLAN lookup failures occur before allocation.

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

@Task(
    input=CreatePrefixInput,
    output=CreatePrefixResult,
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    mcp={
        "annotations": {
            "title": "Create Prefix",
            "readOnlyHint": False,
            "destructiveHint": False,
            "idempotentHint": False,
            "openWorldHint": True,
        }
    },
)
def create_prefix(
    self,
    job: Job,
    parent: Union[str, dict],
    description: str = None,
    prefixlen: int = 30,
    vrf: str = None,
    tags: Union[None, list] = None,
    tenant: str = None,
    comments: str = None,
    role: str = None,
    site: str = None,
    status: str = None,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    branch: str = None,
    custom_fields: Union[None, dict] = None,
    vlan: Union[None, int] = None,
    vlan_group: Union[None, str] = None,
) -> Result:
    """
    Creates a new IP prefix in NetBox or updates an existing one.

    Args:
        job: NorFab job supplying progress events and execution context.
        parent (Union[str, dict]): Parent prefix to allocate new prefix from, could be:

            - IPv4 prefix string e.g. 10.0.0.0/24
            - IPv6 prefix string e.g. 2001::/64
            - Prefix description string to filter by
            - Dictionary with prefix filters for `pynetbox` prefixes.get method
                e.g. `{"prefix": "10.0.0.0/24", "site": "foo"}`

        custom_fields (dict, optional): Custom-field names and values to PATCH
            on the allocated prefix. New list items are added to existing
            custom-field lists; scalar values and null replace old values.
            Object and multiobject fields accept related object names or IDs.
            Definitions must already exist in NetBox. The result diff lists
            changed field names only, not their values.
        description (str): Description for the new prefix, prefix description used for
            deduplication to source existing prefixes.
        prefixlen (int, optional): The prefix length of the new prefix to create, by default
            allocates next available /30 point-to-point prefix.
        vrf (str, optional): Name of the VRF to associate with the prefix.
        tags (Union[None, list], optional): Tag names to add to the prefix.
        tenant (str, optional): Name of the tenant to associate with the prefix.
        comments (str, optional): Comments for the prefix.
        role (str, optional): Role to assign to the prefix.
        site (str, optional): Name of the site to associate with the prefix.
        vlan (int, optional): Existing VLAN VID (1-4094), identified together
            with vlan_group. Omission preserves the current association.
            Site cannot substitute for the VLAN group.
        vlan_group (str, optional): Existing VLAN group name. Must be supplied
            together with vlan; neither the group nor VLAN is created.
        status (str, optional): Status of the prefix.
        instance (Union[None, str], optional): NetBox instance identifier.
        dry_run (bool, optional): If True, simulates the creation without making changes.
        branch (str, optional): Branch name to use, need to have branching plugin installed,
            automatically creates branch if it does not exist in Netbox.

    Returns:
        Result: Prefix details, status, resources, and changed fields. VLAN
            changes use old/new "VID (ID)" values in diff["vlan"]. Dry runs
            report the planned association without writing it.

    Raises:
        NetboxAllocationError: The parent or requested VLAN/group cannot be
            resolved uniquely, allocation fails, or an existing prefix has
            a different length. VLAN lookup failures occur before allocation.
    """
    instance = instance or self.default_instance
    log.info(
        f"{self.name} - Create prefix: Creating /{prefixlen} prefix within '{parent}' from '{instance}' Netbox"
    )
    changed = {}
    ret = Result(
        task=f"{self.name}:create_prefix",
        result={},
        resources=[instance],
        diff=changed,
    )
    tags = tags or []
    nb_prefix = None
    nb = self._get_pynetbox(instance, branch=branch, job=job)
    nb_vlan = None
    if vlan is not None:
        try:
            group = nb.ipam.vlan_groups.get(name=vlan_group)
            if not group:
                raise ValueError(f"VLAN group '{vlan_group}' not found")
            nb_vlan = nb.ipam.vlans.get(group_id=group.id, vid=vlan)
            if not nb_vlan:
                raise ValueError(f"VLAN {vlan} not found in group '{vlan_group}'")
        except Exception as exc:
            raise NetboxAllocationError(
                f"Unable to resolve VLAN {vlan} in group '{vlan_group}': {exc}"
            ) from exc

    job.event(
        f"processing prefix create request within '{parent}' for '/{prefixlen}' subnet"
    )

    # source parent prefix from Netbox
    if isinstance(parent, str):
        # check if parent prefix is IP network or description
        try:
            _ = ipaddress.ip_network(parent)
            is_network = True
        except Exception:
            is_network = False
        if is_network is True and vrf:
            parent_filters = {"prefix": parent, "vrf__name": vrf}
        elif is_network is True:
            parent_filters = {"prefix": parent}
        elif is_network is False and vrf:
            parent_filters = {"description": parent, "vrf__name": vrf}
        elif is_network is False:
            parent_filters = {"description": parent}
    elif isinstance(parent, dict):
        parent_filters = dict(parent)
        parent_site = parent_filters.pop("site", None)
        if parent_site:
            nb_site = nb.dcim.sites.get(name=parent_site)
            if not nb_site:
                raise NetboxAllocationError(
                    f"Failed to get parent prefix site '{parent_site}' from NetBox"
                )
            parent_filters.update(scope_type="dcim.site", scope_id=nb_site.id)
    nb_parent_prefix = nb.ipam.prefixes.get(**parent_filters)
    if not nb_parent_prefix:
        job.event(f"parent prefix not found in NetBox: {parent}", severity="ERROR")
        raise NetboxAllocationError(
            f"Unable to source parent prefix from Netbox - {parent}"
        )

    # check that parent vrf and new prefix vrf are same
    if vrf and str(nb_parent_prefix.vrf) != vrf:
        raise NetboxAllocationError(
            f"Parent prefix vrf '{nb_parent_prefix.vrf}' not same as requested child prefix vrf '{vrf}'"
        )

    # try to source existing prefix from netbox
    prefix_filters = {}
    if vrf:
        prefix_filters["vrf__name"] = vrf
    if site:
        prefix_filters["site__name"] = site
    if role:
        prefix_filters["role__name"] = role
    if description:
        prefix_filters["description"] = description
    try:
        if prefix_filters:
            nb_prefix = nb.ipam.prefixes.get(
                within=nb_parent_prefix.prefix, **prefix_filters
            )
    except Exception as e:
        raise NetboxAllocationError(
            f"Failed to source existing prefix from Netbox using filters '{prefix_filters}', error: {e}"
        )

    # create new prefix
    if not nb_prefix:
        job.event(
            f"creating new '/{prefixlen}' prefix within '{nb_parent_prefix}' prefix"
        )
        # execute dry run on new prefix
        if dry_run is True:
            nb_prefixes = nb_parent_prefix.available_prefixes.list()
            if not nb_prefixes:
                raise NetboxAllocationError(
                    f"Parent prefix '{parent}' has no child prefixes available"
                )
            for pfx in nb_prefixes:
                # Skip ranges too small for the requested child prefix, then
                # select the first child within the first range that fits.
                network = ipaddress.ip_network(pfx.prefix)
                if network.prefixlen <= prefixlen:
                    nb_prefix = str(next(network.subnets(new_prefix=prefixlen)))
                    break
            else:
                raise NetboxAllocationError(
                    f"Parent prefix '{parent}' has no child prefixes available with '/{prefixlen}' prefix length"
                )
            ret.status = "unchanged"
            ret.dry_run = True
            if nb_vlan is not None:
                changed["vlan"] = {"-": None, "+": f"{nb_vlan.vid} ({nb_vlan.id})"}
                ret.diff = changed
            ret.result = {
                "prefix": nb_prefix,
                "description": description,
                "parent": nb_parent_prefix.prefix,
                "vrf": vrf,
                "site": site,
            }
            # add branch to results
            if branch is not None:
                ret.result["branch"] = branch
            job.event(f"dry-run: would create prefix '{nb_prefix}'")
            return ret
        # create new prefix
        else:
            try:
                nb_prefix = nb_parent_prefix.available_prefixes.create(
                    {"prefix_length": prefixlen}
                )
            except Exception as e:
                raise NetboxAllocationError(
                    f"Failed creating child prefix of '/{prefixlen}' prefix length "
                    f"within parent prefix '{str(nb_parent_prefix)}', error: {e}"
                )
        job.event(f"created new '{nb_prefix}' prefix within '{parent}' prefix")
        ret.status = "created"
    else:
        # check existing prefix length matching requested length
        if not nb_prefix.prefix.endswith(f"/{prefixlen}"):
            raise NetboxAllocationError(
                f"Found existing child prefix '{nb_prefix.prefix}' with mismatch "
                f"requested prefix length '/{prefixlen}'"
            )
        job.event(f"using existing prefix {nb_prefix}")

    # update prefix parameters
    if nb_vlan is not None:
        current_vlan_id = nb_prefix.vlan.id if nb_prefix.vlan else None
        if current_vlan_id != nb_vlan.id:
            changed["vlan"] = {
                "-": (
                    f"{nb_prefix.vlan.vid} ({current_vlan_id})"
                    if nb_prefix.vlan
                    else None
                ),
                "+": f"{nb_vlan.vid} ({nb_vlan.id})",
            }
            nb_prefix.vlan = nb_vlan.id
    if custom_fields is not None:
        reference_fields = self.bulk_filter(
            nb.extras.custom_fields,
            object_type="ipam.prefix",
            type=["object", "multiobject"],
        )
        # Normalize existing references to IDs before comparing the merged
        # values, so a repeat call does not report a false change.
        current_fields, merged_fields = merge_resolved_custom_fields(
            self,
            nb,
            nb_prefix,
            custom_fields,
            [field for field in reference_fields if field.type.value == "object"],
            [
                field
                for field in reference_fields
                if field.type.value == "multiobject"
            ],
        )
        changed_fields = [
            name
            for name in custom_fields
            if current_fields.get(name) != merged_fields.get(name)
        ]
        if changed_fields:
            changed["custom_fields"] = changed_fields
            if not dry_run:
                nb_prefix.update({"custom_fields": merged_fields})
    if description and description != nb_prefix.description:
        changed["description"] = {"-": str(nb_prefix.description), "+": description}
        nb_prefix.description = description
    if vrf and vrf != str(nb_prefix.vrf):
        changed["vrf"] = {"-": str(nb_prefix.vrf), "+": vrf}
        nb_prefix.vrf = {"name": vrf}
    if tenant and tenant != str(nb_prefix.tenant):
        changed["tenant"] = {
            "-": str(nb_prefix.tenant) if nb_prefix.tenant else None,
            "+": tenant,
        }
        nb_prefix.tenant = {"name": tenant}
    if site and str(nb_prefix.scope) != site:
        nb_site = nb.dcim.sites.get(name=site)
        if not nb_site:
            raise NetboxAllocationError(f"Failed to get '{site}' site from Netbox")
        changed["site"] = {
            "-": str(nb_prefix.scope) if nb_prefix.scope else None,
            "+": nb_site.name,
        }
        nb_prefix.scope_type = "dcim.site"
        nb_prefix.scope_id = nb_site.id
    if status and status.lower() != nb_prefix.status:
        changed["status"] = {"-": str(nb_prefix.status), "+": status.title()}
        nb_prefix.status = status.lower()
    if comments and comments != nb_prefix.comments:
        changed["comments"] = {"-": str(nb_prefix.comments), "+": comments}
        nb_prefix.comments = comments
    if role and role != str(nb_prefix.role):
        changed["role"] = {"-": str(nb_prefix.role), "+": role}
        nb_prefix.role = {"name": role}
    existing_tags = [tag.name for tag in nb_prefix.tags]
    merged_tags = merge_array_values(nb_prefix.tags, tags, attribute="name")
    merged_names = [tag["name"] for tag in merged_tags]
    if merged_names != existing_tags:
        changed["tags"] = {
            "-": existing_tags,
            "+": merged_names,
        }
        if not dry_run:
            nb_prefix.tags = merged_tags

    # save prefix into Netbox
    if dry_run is True:
        ret.status = "unchanged"
        ret.dry_run = True
        ret.diff = changed
        job.event(
            f"dry-run: prefix '{nb_prefix.prefix}' has {len(changed)} pending change(s)"
        )
    elif changed:
        ret.diff = changed
        nb_prefix.save()
        if ret.status != "created":
            ret.status = "updated"
        job.event(
            f"updated prefix '{nb_prefix.prefix}' with {len(changed)} change(s)"
        )
    else:
        ret.status = "unchanged"
        job.event(f"prefix '{nb_prefix.prefix}' already in desired state")

    # source vrf name
    vrf_name = None
    if nb_prefix.vrf:
        if isinstance(nb_prefix.vrf, dict):
            vrf_name = nb_prefix.vrf["name"]
        else:
            vrf_name = nb_prefix.vrf.name

    # form and return results
    ret.result = {
        "prefix": nb_prefix.prefix,
        "description": nb_prefix.description,
        "vrf": vrf_name,
        "site": str(nb_prefix.scope) if nb_prefix.scope else site,
        "parent": nb_parent_prefix.prefix,
    }
    # add branch to results
    if branch is not None:
        ret.result["branch"] = branch

    job.event(f"prefix task complete for '{nb_prefix.prefix}'")
    return ret