1 | # py312 |
2 | # requirements: |
3 | # pynetbox==7.9.0 |
4 |
|
5 |
|
6 | import re |
7 | from typing import Literal, Optional, TypedDict |
8 |
|
9 | import pynetbox |
10 |
|
11 |
|
12 | class netbox(TypedDict): |
13 | url: str |
14 | token: str |
15 |
|
16 |
|
17 | def api(auth: netbox): |
18 | # pynetbox adds /api itself; the resource URL may already end with it. |
19 | return pynetbox.api( |
20 | auth["url"].rstrip("/").removesuffix("/api"), token=auth["token"] |
21 | ) |
22 |
|
23 |
|
24 | def main( |
25 | auth: netbox, |
26 | method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"], |
27 | path: str, |
28 | query: Optional[dict] = None, |
29 | body: Optional[dict] = None, |
30 | ): |
31 | """ |
32 | Call NetBox API |
33 | Call any NetBox REST endpoint: path relative to /api (e.g. dcim/cables/), query params, and a JSON body (object, or a list for bulk PATCH/POST/DELETE). |
34 | """ |
35 | # Any endpoint, so this one uses pynetbox's HTTP session directly. |
36 | nb = api(auth) |
37 | params = { |
38 | k: [x for x in (v if isinstance(v, list) else [v]) if x not in (None, "")] |
39 | for k, v in (query or {}).items() |
40 | } |
41 | response = nb.http_session.request( |
42 | method, |
43 | f"{nb.base_url}/{re.sub(r'^/?(api/)?', '', path)}", |
44 | params=params, |
45 | json=body if method != "GET" and body else None, |
46 | headers={ |
47 | "Authorization": f"{'Bearer' if auth['token'].startswith('nbt_') else 'Token'} {auth['token']}", |
48 | "Accept": "application/json", |
49 | }, |
50 | ) |
51 | if not response.ok: |
52 | raise Exception(f"{response.status_code} {response.text}") |
53 | if response.status_code == 204: |
54 | return {"success": True} |
55 | return response.json() |
56 | |