Skip to content

allow for better performance when not using interface auto descriptions #103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
loulecrivain wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

allow for better performance when not using interface auto descriptions #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b0c5f46
separate graphql queries from source code
loulecrivain Feb 23, 2026
05756fb
add @without_feature decorator for tests
loulecrivain Feb 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
351 changes: 11 additions & 340 deletions cosmo/clients/netbox_v4.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,12 @@
from abc import ABC, abstractmethod
from builtins import map
from multiprocessing import Manager
from string import Template
from os import PathLike
from pathlib import Path

from cosmo.clients import get_client_mp_context
from cosmo.clients.netbox_client import NetboxAPIClient
from cosmo.common import FileTemplate


class ParallelQuery(ABC):
Expand All @@ -16,6 +18,10 @@ def __init__(self, client: NetboxAPIClient, **kwargs):
self.data_promise = None
self.kwargs = kwargs

@staticmethod
def file_template(relpath: str | PathLike):
return FileTemplate(Path(__file__).parent.joinpath(Path(relpath)))

def fetch_data(self, pool):
return pool.apply_async(self._fetch_data, args=(self.kwargs, pool))

Expand Down Expand Up @@ -43,43 +49,7 @@ def _fetch_data(self, kwargs, pool):
if self.netbox_43_query_syntax
else 'tag: "bgp_cpe"'
)
query_template = Template(
"""
query {
interface_list(filters: { $tag_filter }) {
__typename
id,
parent {
__typename
id,
connected_endpoints {
... on InterfaceType {
__typename
name
device {
name
__typename
primary_ip4 {
__typename
address
}
interfaces {
id
name
__typename
ip_addresses {
__typename
address
}
}
}
}
}
}
}
}
"""
)
query_template = self.file_template("queries/connected_devices.graphql")

return self.client.query(
query_template.substitute(tag_filter=tag_filter), "connected_devices_query"
Expand Down Expand Up @@ -123,49 +93,7 @@ def _fetch_data(self, kwargs, pool):
# Note: This does not use the device list, because we can have other participating devices
# which are not in the same repository and thus are not appearing in device list.

query_template = Template(
"""
query{
interface_list(filters: {
name: {starts_with: "lo"}
}) {
__typename
name,
child_interfaces {
__typename
name,
vrf {
__typename
id
name
description
rd
export_targets {
__typename
name
}
import_targets {
__typename
name
}
},
ip_addresses {
__typename
address,
family {
__typename
value,
}
}
}
device{
__typename
name,
}
}
}
"""
)
query_template = self.file_template("queries/loopback.graphql")

return self.client.query(query_template.substitute(), "loopback_query")["data"]

Expand Down Expand Up @@ -205,74 +133,7 @@ def _merge_into(self, data: dict, query_data):

class L2VPNDataQuery(ParallelQuery):
def _fetch_data(self, kwargs, pool):
query_template = Template(
"""
query {
l2vpn_list (filters: {name: {starts_with: "WAN: "}}) {
__typename
id
name
type
identifier
terminations {
__typename
id
assigned_object {
__typename
... on VLANType {
__typename
id
name
interfaces_as_tagged {
id
name
__typename
device {
__typename
id
name
}
}
interfaces_as_untagged {
id
name
__typename
device {
__typename
id
name
}
}
}
... on InterfaceType {
__typename
id
name
custom_fields
untagged_vlan {
__typename
id
name
vid
}
tagged_vlans {
__typename
id
name
vid
}
device {
__typename
id
name
}
}
}
}
}
}
"""
)
query_template = self.file_template("queries/l2vpn.graphql")

return self.client.query(query_template.substitute(), "l2vpn_query")["data"]

Expand Down Expand Up @@ -428,197 +289,7 @@ def __init__(self, *args, multiple_mac_addresses=False, **kwargs):

def _fetch_data(self, kwargs, pool):
device = kwargs.get("device")
query_template = Template(
"""
query {
device_list(filters: {
name: { i_exact: $device },
}) {
__typename
id
name
custom_fields

device_type {
__typename
slug
}
platform {
__typename
manufacturer {
__typename
slug
}
slug
}
primary_ip4 {
__typename
address
}

interfaces {
__typename
id
name
enabled
type
mode
mtu
description
connected_endpoints {
... on ProviderNetworkType {
__typename
display
}
... on CircuitTerminationType {
__typename
display
}
... on VirtualCircuitTerminationType {
__typename
display
}
... on InterfaceType {
__typename
name
device {
__typename
name
}
}
... on FrontPortType {
__typename
name
device {
__typename
name
}
}
... on RearPortType {
__typename
name
device {
__typename
name
}
}
... on ConsolePortType {
__typename
name
device {
__typename
name
}
}
... on ConsoleServerPortType {
__typename
name
device {
__typename
name
}
}
}
link_peers {
... on CircuitTerminationType {
__typename
display
}
... on FrontPortType {
__typename
name
device {
__typename
name
}
}
... on RearPortType {
__typename
name
device {
__typename
name
}
}
... on ConsolePortType {
__typename
name
device {
__typename
name
}
}
... on ConsoleServerPortType {
__typename
name
device {
__typename
name
}
}
... on InterfaceType {
__typename
name
device {
__typename
name
}
}
}
vrf {
__typename
id
name
description
rd
export_targets {
__typename
name
}
import_targets {
__typename
name
}
}
lag {
__typename
id
name
}
ip_addresses {
__typename
address
role
}
untagged_vlan {
__typename
id
name
vid
}
tagged_vlans {
__typename
id
name
vid
}
tags {
__typename
id
name
slug
}
parent {
__typename
id
mtu
name
}
custom_fields
}
}
}"""
)
query_template = self.file_template("queries/device.graphql")

query = query_template.substitute(
device=json.dumps(device),
Expand Down
Loading