-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathtunnel_termination.py
More file actions
90 lines (76 loc) · 2.65 KB
/
Copy pathtunnel_termination.py
File metadata and controls
90 lines (76 loc) · 2.65 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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
"""
TunnelTermination entity examples for the Diode Python SDK.
This module demonstrates three patterns for ingesting TunnelTermination entities:
- tunnel_termination_minimal: Required fields only
- tunnel_termination_extended: Common optional fields
- tunnel_termination_explicit: Fully nested objects with all fields
"""
from netboxlabs.diode.sdk import DiodeClient
from netboxlabs.diode.sdk.ingester import (
Entity,
IPAddress,
Tag,
Tunnel,
TunnelTermination,
)
TARGET = "grpc://localhost:8080/diode"
APP_NAME = "tunnel_termination-example"
APP_VERSION = "1.0.0"
CLIENT_ID = "diode"
CLIENT_SECRET = "changeme"
def main():
"""Main execution - demonstrates ingesting a TunnelTermination entity."""
with DiodeClient(
target=TARGET,
app_name=APP_NAME,
app_version=APP_VERSION,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
) as client:
# Choose one of the three patterns:
tunnel_termination = tunnel_termination_minimal()
# tunnel_termination = tunnel_termination_extended()
# tunnel_termination = tunnel_termination_explicit()
response = client.ingest(
entities=[Entity(tunnel_termination=tunnel_termination)]
)
if response.errors:
print(f"Errors: {response.errors}")
else:
print("TunnelTermination ingested successfully")
def tunnel_termination_minimal() -> TunnelTermination:
"""Create a TunnelTermination with only required fields using flat strings."""
return TunnelTermination(
tunnel="Example Tunnel", # flat string -> Tunnel
role="hub",
metadata={"source": "example"},
)
def tunnel_termination_extended() -> TunnelTermination:
"""Create a TunnelTermination with common optional fields."""
return TunnelTermination(
tunnel="Example Tunnel",
role="hub",
metadata={"source": "example", "custom_key": "custom_value"},
)
def tunnel_termination_explicit() -> TunnelTermination:
"""Create a TunnelTermination with fully nested objects and all common fields."""
return TunnelTermination(
tunnel=Tunnel(
name="Example Name",
status="active",
encapsulation="gre",
metadata={"source": "example"},
),
role="hub",
metadata={
"source": "example",
"custom_key": "custom_value",
"collected_at": "2024-01-15T10:30:00Z",
},
outside_ip=IPAddress(
address="192.0.2.1/32", status="active", metadata={"source": "example"}
),
tags=[Tag(name="production")],
)
if __name__ == "__main__":
main()