1
1
from __future__ import annotations
2
2
3
- from asyncio import run as aiorun
4
-
5
- from infrahub_sdk .async_typer import AsyncTyper
6
-
7
3
from typing import Annotated
8
4
9
- from pydantic import BaseModel , Field , ConfigDict
10
- from infrahub_sdk import InfrahubClient
5
+ from pydantic import ConfigDict , Field
11
6
from rich import print as rprint
12
- from infrahub_sdk .schema import InfrahubAttributeParam as AttrParam , InfrahubRelationshipParam as RelParam , AttributeKind , from_pydantic , NodeSchema , NodeModel , GenericSchema , GenericModel , RelationshipKind
13
7
8
+ from infrahub_sdk import InfrahubClient
9
+ from infrahub_sdk .async_typer import AsyncTyper
10
+ from infrahub_sdk .schema import (
11
+ GenericModel ,
12
+ GenericSchema ,
13
+ NodeModel ,
14
+ NodeSchema ,
15
+ RelationshipKind ,
16
+ from_pydantic ,
17
+ )
18
+ from infrahub_sdk .schema import (
19
+ InfrahubAttributeParam as AttrParam ,
20
+ )
21
+ from infrahub_sdk .schema import (
22
+ InfrahubRelationshipParam as RelParam ,
23
+ )
14
24
15
25
app = AsyncTyper ()
16
26
17
27
18
28
class Site (NodeModel ):
19
29
model_config = ConfigDict (
20
- node_schema = NodeSchema (name = "Site" , namespace = "Infra" , human_friendly_id = ["name__value" ], display_labels = ["name__value" ])
30
+ node_schema = NodeSchema (
31
+ name = "Site" , namespace = "Infra" , human_friendly_id = ["name__value" ], display_labels = ["name__value" ]
32
+ )
21
33
)
22
34
23
35
name : Annotated [str , AttrParam (unique = True )] = Field (description = "The name of the site" )
24
36
25
37
26
38
class Vlan (NodeModel ):
27
39
model_config = ConfigDict (
28
- node_schema = NodeSchema (name = "Vlan" , namespace = "Infra" , human_friendly_id = ["vlan_id__value" ], display_labels = ["vlan_id__value" ])
40
+ node_schema = NodeSchema (
41
+ name = "Vlan" , namespace = "Infra" , human_friendly_id = ["vlan_id__value" ], display_labels = ["vlan_id__value" ]
42
+ )
29
43
)
30
44
31
45
name : str
@@ -35,39 +49,45 @@ class Vlan(NodeModel):
35
49
36
50
class Device (NodeModel ):
37
51
model_config = ConfigDict (
38
- node_schema = NodeSchema (name = "Device" , namespace = "Infra" , human_friendly_id = ["name__value" ], display_labels = ["name__value" ])
52
+ node_schema = NodeSchema (
53
+ name = "Device" , namespace = "Infra" , human_friendly_id = ["name__value" ], display_labels = ["name__value" ]
54
+ )
39
55
)
40
56
41
57
name : Annotated [str , AttrParam (unique = True )] = Field (description = "The name of the car" )
42
58
site : Annotated [Site , RelParam (kind = RelationshipKind .ATTRIBUTE , identifier = "device__site" )]
43
- interfaces : Annotated [list [Interface ], RelParam (kind = RelationshipKind .COMPONENT , identifier = "device__interfaces" )] = Field (default_factory = list )
59
+ interfaces : Annotated [
60
+ list [Interface ], RelParam (kind = RelationshipKind .COMPONENT , identifier = "device__interfaces" )
61
+ ] = Field (default_factory = list )
44
62
45
63
46
64
class Interface (GenericModel ):
47
65
model_config = ConfigDict (
48
- generic_schema = GenericSchema (name = "Interface" , namespace = "Infra" , human_friendly_id = ["device__name__value" , "name__value" ], display_labels = ["name__value" ])
66
+ generic_schema = GenericSchema (
67
+ name = "Interface" ,
68
+ namespace = "Infra" ,
69
+ human_friendly_id = ["device__name__value" , "name__value" ],
70
+ display_labels = ["name__value" ],
71
+ )
49
72
)
50
73
51
74
device : Annotated [Device , RelParam (kind = RelationshipKind .PARENT , identifier = "device__interfaces" )]
52
75
name : str
53
76
description : str | None = None
54
77
78
+
55
79
class L2Interface (Interface ):
56
- model_config = ConfigDict (
57
- node_schema = NodeSchema (name = "L2Interface" , namespace = "Infra" )
58
- )
59
-
80
+ model_config = ConfigDict (node_schema = NodeSchema (name = "L2Interface" , namespace = "Infra" ))
81
+
60
82
vlans : list [Vlan ] = Field (default_factory = list )
61
83
84
+
62
85
class LoopbackInterface (Interface ):
63
- model_config = ConfigDict (
64
- node_schema = NodeSchema (name = "LoopbackInterface" , namespace = "Infra" )
65
- )
66
-
86
+ model_config = ConfigDict (node_schema = NodeSchema (name = "LoopbackInterface" , namespace = "Infra" ))
67
87
68
88
69
89
@app .command ()
70
- async def load_schema ():
90
+ async def load_schema () -> None :
71
91
client = InfrahubClient ()
72
92
schema = from_pydantic (models = [Site , Device , Interface , L2Interface , LoopbackInterface , Vlan ])
73
93
rprint (schema .to_schema_dict ())
@@ -76,7 +96,7 @@ async def load_schema():
76
96
77
97
78
98
@app .command ()
79
- async def load_data ():
99
+ async def load_data () -> None :
80
100
client = InfrahubClient ()
81
101
82
102
atl = await client .create ("InfraSite" , name = "ATL" )
@@ -100,14 +120,15 @@ async def load_data():
100
120
101
121
102
122
@app .command ()
103
- async def query_data ():
123
+ async def query_data () -> None :
104
124
client = InfrahubClient ()
105
125
sites = await client .all (kind = Site )
126
+ rprint (sites )
106
127
107
- breakpoint ()
108
128
devices = await client .all (kind = Device )
109
129
for device in devices :
110
130
rprint (device )
111
131
132
+
112
133
if __name__ == "__main__" :
113
- app ()
134
+ app ()
0 commit comments