1 | # py312 |
2 | # requirements: |
3 | # pynetbox==7.9.0 |
4 |
|
5 |
|
6 | from typing import List, Literal, Optional, TypedDict |
7 |
|
8 | import pynetbox |
9 |
|
10 |
|
11 | class netbox(TypedDict): |
12 | url: str |
13 | token: str |
14 |
|
15 |
|
16 | DynSelect_site = str |
17 | DynSelect_role = str |
18 |
|
19 |
|
20 | def api(auth: netbox): |
21 | # pynetbox adds /api itself; the resource URL may already end with it. |
22 | return pynetbox.api( |
23 | auth["url"].rstrip("/").removesuffix("/api"), token=auth["token"] |
24 | ) |
25 |
|
26 |
|
27 | def keep(**fields): |
28 | # The run form sends "" / [] / {} for untouched fields: drop them. |
29 | return {k: v for k, v in fields.items() if v not in (None, "", [], {})} |
30 |
|
31 |
|
32 | def options(endpoint, key="id", **filters): |
33 | return [ |
34 | {"value": str(getattr(r, key)), "label": r.display} |
35 | for r in endpoint.filter(brief=1, limit=1000, offset=0, **filters) |
36 | ] |
37 |
|
38 |
|
39 | # Dropdown of the instance's sites, by slug: the first 1000 (NetBox MAX_PAGE_SIZE); set the value by expression beyond that. |
40 | def site(auth: netbox): |
41 | return options(api(auth).dcim.sites, "slug") |
42 |
|
43 |
|
44 | # Dropdown of the instance's device roles, by slug. |
45 | def role(auth: netbox): |
46 | return options(api(auth).dcim.device_roles, "slug") |
47 |
|
48 |
|
49 | DynMultiselect_device_ids = List[str] |
50 |
|
51 |
|
52 | # Candidate devices for the chosen site, role and status: the first 1000; use all_matching beyond that. |
53 | def device_ids( |
54 | auth: netbox, |
55 | site: DynSelect_site, |
56 | role: DynSelect_role, |
57 | status: Optional[List[str]] = None, |
58 | ): |
59 | if not site or not role: |
60 | return [] |
61 | return options(api(auth).dcim.devices, **keep(site=site, role=role, status=status)) |
62 |
|
63 |
|
64 | def main( |
65 | auth: netbox, |
66 | site: DynSelect_site, |
67 | role: DynSelect_role, |
68 | status: Optional[ |
69 | List[ |
70 | Literal[ |
71 | "offline", |
72 | "active", |
73 | "planned", |
74 | "staged", |
75 | "failed", |
76 | "inventory", |
77 | "decommissioning", |
78 | ] |
79 | ] |
80 | ] = None, |
81 | device_ids: Optional[DynMultiselect_device_ids] = None, |
82 | all_matching: bool = False, |
83 | ): |
84 | """ |
85 | Select Devices |
86 | Pick devices of a site and role (optionally by status) from a live list, or take all of them, and return their raw NetBox JSON, including platform.name and custom_fields. |
87 | """ |
88 | if not all_matching and not device_ids: |
89 | return [] |
90 | # pynetbox sends a None filter as "null", so drop the empty ones. |
91 | filters = keep( |
92 | site=site, role=role, status=status, id=None if all_matching else device_ids |
93 | ) |
94 |
|
95 | # Page by offset: pynetbox's full iteration requests NetBox's next links |
96 | # as is, and behind a proxy their host or scheme can be internal. |
97 | endpoint, devices = api(auth).dcim.devices, [] |
98 | while True: |
99 | records = endpoint.filter(**filters, limit=1000, offset=len(devices)) |
100 | batch = [dict(r) for r in records] |
101 | devices += batch |
102 | if not batch or len(devices) >= len(records): |
103 | return devices |
104 |
|