Skip to content

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.

Inputs¤

Parameter Required Description
parent Yes Parent prefix as a network string, description string, or pynetbox filter dictionary
description No Description for the new prefix and deduplication key for existing prefixes
prefixlen No Length of the child prefix, default 30
vrf No VRF name to associate with the prefix
tags No Tags to assign to the prefix
tenant No Tenant name to associate with the prefix
comments No Prefix comments
role No Prefix role
site No Site name to associate with the prefix
status No Prefix status
instance No NetBox instance name to target
branch No NetBox Branching plugin branch name to write to
dry_run No Preview the candidate prefix without writing

Output¤

Returns the created or updated prefix data. In dry-run mode, returns the candidate prefix that would be allocated.

{
    "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.
  • If vrf is provided, the parent prefix must belong to the same VRF.
  • Branch writes require the NetBox Branching Plugin.

Examples¤

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

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
result = client.run_job(
    "netbox",
    "create_prefix",
    workers="any",
    kwargs={
        "parent": "10.0.0.0/24",
        "description": "leaf-1 to spine-1",
    },
)

# 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
            ├── 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
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
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]

List of tags to assign 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
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

An object containing the outcome, including status, details of the prefix, and resources used.
Source code in norfab\workers\netbox_worker\prefix_tasks.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
@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,
) -> Result:
    """
    Creates a new IP prefix in NetBox or updates an existing one.

    Args:
        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"}`

        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): List of tags to assign 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.
        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: An object containing the outcome, including status, details of the prefix, and resources used.
    """
    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.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 = parent
    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 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 '{parent}' 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:
                # parent prefix empty, can use first subnet as a child prefix
                if pfx.prefix == nb_parent_prefix.prefix:
                    nb_prefix = (
                        nb_parent_prefix.prefix.split("/")[0] + f"/{prefixlen}"
                    )
                    break
                # find child prefix by prefix length
                elif str(pfx).endswith(f"/{prefixlen}"):
                    nb_prefix = str(pfx)
                    break
            else:
                # For some reason Netbox does not always return /31 as available
                # subnet int he list, this is to check that child subnet fits into
                # first subnet available in Netbox
                if int(nb_prefixes[0].prefix.split("/")[1]) < prefixlen:
                    nb_prefix = (
                        nb_parent_prefix.prefix.split("/")[0] + f"/{prefixlen}"
                    )
                else:
                    raise NetboxAllocationError(
                        f"Parent prefix '{parent}' has no child prefixes available with '/{prefixlen}' prefix length"
                    )
            ret.status = "unchanged"
            ret.dry_run = True
            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 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 != nb_prefix.role:
        changed["role"] = {"-": str(nb_prefix.role), "+": role}
        nb_prefix.role = {"name": role}
    existing_tags = [str(t) for t in nb_prefix.tags]
    if tags and not any(t in existing_tags for t in tags):
        changed["tags"] = {
            "-": existing_tags,
            "+": [t for t in tags if t not in existing_tags] + existing_tags,
        }
        for t in tags:
            if t not in existing_tags:
                nb_prefix.tags.append({"name": t})

    # 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