Merge pull request #6678 from netbox-community/2007-graphql

Closes #2007: Implement GraphQL API
This commit is contained in:
Jeremy Stretch 2021-06-30 10:42:49 -04:00 committed by GitHub
commit 57fc6efd4c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
45 changed files with 1456 additions and 10 deletions

View File

@ -18,6 +18,10 @@ django-debug-toolbar
# https://github.com/carltongibson/django-filter
django-filter
# Django debug toolbar extension with support for GraphiQL
# https://github.com/flavors/django-graphiql-debug-toolbar/
django-graphiql-debug-toolbar
# Modified Preorder Tree Traversal (recursive nesting of objects)
# https://github.com/django-mptt/django-mptt
django-mptt
@ -54,6 +58,10 @@ djangorestframework
# https://github.com/axnsan12/drf-yasg
drf-yasg[validation]
# Django wrapper for Graphene (GraphQL support)
# https://github.com/graphql-python/graphene-django
graphene_django
# WSGI HTTP server
# https://gunicorn.org/
gunicorn

View File

@ -201,6 +201,14 @@ EXEMPT_VIEW_PERMISSIONS = ['*']
---
## GRAPHQL_ENABLED
Default: True
Setting this to False will disable the GraphQL API.
---
## HTTP_PROXIES
Default: None

View File

@ -0,0 +1,70 @@
# GraphQL API Overview
NetBox provides a read-only [GraphQL](https://graphql.org/) API to complement its REST API. This API is powered by the [Graphene](https://graphene-python.org/) library and [Graphene-Django](https://docs.graphene-python.org/projects/django/en/latest/).
## Queries
GraphQL enables the client to specify an arbitrary nested list of fields to include in the response. All queries are made to the root `/graphql` API endpoint. For example, to return the circuit ID and provider name of each circuit with an active status, you can issue a request such as the following:
```
curl -H "Authorization: Token $TOKEN" \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
http://netbox/graphql/ \
--data '{"query": "query {circuits(status:\"active\" {cid provider {name}}}"}'
```
The response will include the requested data formatted as JSON:
```json
{
"data": {
"circuits": [
{
"cid": "1002840283",
"provider": {
"name": "CenturyLink"
}
},
{
"cid": "1002840457",
"provider": {
"name": "CenturyLink"
}
}
]
}
}
```
!!! note
It's recommended to pass the return data through a JSON parser such as `jq` for better readability.
NetBox provides both a singular and plural query field for each object type:
* `$OBJECT`: Returns a single object. Must specify the object's unique ID as `(id: 123)`.
* `$OBJECT_list`: Returns a list of objects, optionally filtered by given parameters.
For example, query `device(id:123)` to fetch a specific device (identified by its unique ID), and query `device_list` (with an optional set of fitlers) to fetch all devices.
For more detail on constructing GraphQL queries, see the [Graphene documentation](https://docs.graphene-python.org/en/latest/).
## Filtering
The GraphQL API employs the same filtering logic as the UI and REST API. Filters can be specified as key-value pairs within parentheses immediately following the query name. For example, the following will return only sites within the North Carolina region with a status of active:
```
{"query": "query {sites(region:\"north-carolina\", status:\"active\") {name}}"}
```
## Authentication
NetBox's GraphQL API uses the same API authentication tokens as its REST API. Authentication tokens are included with requests by attaching an `Authorization` HTTP header in the following form:
```
Authorization: Token $TOKEN
```
## Disabling the GraphQL API
If not needed, the GraphQL API can be disabled by setting the [`GRAPHQL_ENABLED`](../configuration/optional-settings.md#graphql_enabled) configuration parameter to False and restarting NetBox.

View File

@ -87,6 +87,8 @@ nav:
- Overview: 'rest-api/overview.md'
- Filtering: 'rest-api/filtering.md'
- Authentication: 'rest-api/authentication.md'
- GraphQL API:
- Overview: 'graphql-api/overview.md'
- Development:
- Introduction: 'development/index.md'
- Getting Started: 'development/getting-started.md'

View File

View File

@ -0,0 +1,21 @@
import graphene
from netbox.graphql.fields import ObjectField, ObjectListField
from .types import *
class CircuitsQuery(graphene.ObjectType):
circuit = ObjectField(CircuitType)
circuit_list = ObjectListField(CircuitType)
circuit_termination = ObjectField(CircuitTerminationType)
circuit_termination_list = ObjectListField(CircuitTerminationType)
circuit_type = ObjectField(CircuitTypeType)
circuit_type_list = ObjectListField(CircuitTypeType)
provider = ObjectField(ProviderType)
provider_list = ObjectListField(ProviderType)
provider_network = ObjectField(ProviderNetworkType)
provider_network_list = ObjectListField(ProviderNetworkType)

View File

@ -0,0 +1,50 @@
from circuits import filtersets, models
from netbox.graphql.types import BaseObjectType, ObjectType, TaggedObjectType
__all__ = (
'CircuitTerminationType',
'CircuitType',
'CircuitTypeType',
'ProviderType',
'ProviderNetworkType',
)
class CircuitTerminationType(BaseObjectType):
class Meta:
model = models.CircuitTermination
fields = '__all__'
filterset_class = filtersets.CircuitTerminationFilterSet
class CircuitType(TaggedObjectType):
class Meta:
model = models.Circuit
fields = '__all__'
filterset_class = filtersets.CircuitFilterSet
class CircuitTypeType(ObjectType):
class Meta:
model = models.CircuitType
fields = '__all__'
filterset_class = filtersets.CircuitTypeFilterSet
class ProviderType(TaggedObjectType):
class Meta:
model = models.Provider
fields = '__all__'
filterset_class = filtersets.ProviderFilterSet
class ProviderNetworkType(TaggedObjectType):
class Meta:
model = models.ProviderNetwork
fields = '__all__'
filterset_class = filtersets.ProviderNetworkFilterSet

View File

View File

@ -0,0 +1,105 @@
import graphene
from netbox.graphql.fields import ObjectField, ObjectListField
from .types import *
class DCIMQuery(graphene.ObjectType):
cable = ObjectField(CableType)
cable_list = ObjectListField(CableType)
console_port = ObjectField(ConsolePortType)
console_port_list = ObjectListField(ConsolePortType)
console_port_template = ObjectField(ConsolePortTemplateType)
console_port_template_list = ObjectListField(ConsolePortTemplateType)
console_server_port = ObjectField(ConsoleServerPortType)
console_server_port_list = ObjectListField(ConsoleServerPortType)
console_server_port_template = ObjectField(ConsoleServerPortTemplateType)
console_server_port_template_list = ObjectListField(ConsoleServerPortTemplateType)
device = ObjectField(DeviceType)
device_list = ObjectListField(DeviceType)
device_bay = ObjectField(DeviceBayType)
device_bay_list = ObjectListField(DeviceBayType)
device_bay_template = ObjectField(DeviceBayTemplateType)
device_bay_template_list = ObjectListField(DeviceBayTemplateType)
device_role = ObjectField(DeviceRoleType)
device_role_list = ObjectListField(DeviceRoleType)
device_type = ObjectField(DeviceTypeType)
device_type_list = ObjectListField(DeviceTypeType)
front_port = ObjectField(FrontPortType)
front_port_list = ObjectListField(FrontPortType)
front_port_template = ObjectField(FrontPortTemplateType)
front_port_template_list = ObjectListField(FrontPortTemplateType)
interface = ObjectField(InterfaceType)
interface_list = ObjectListField(InterfaceType)
interface_template = ObjectField(InterfaceTemplateType)
interface_template_list = ObjectListField(InterfaceTemplateType)
inventory_item = ObjectField(InventoryItemType)
inventory_item_list = ObjectListField(InventoryItemType)
location = ObjectField(LocationType)
location_list = ObjectListField(LocationType)
manufacturer = ObjectField(ManufacturerType)
manufacturer_list = ObjectListField(ManufacturerType)
platform = ObjectField(PlatformType)
platform_list = ObjectListField(PlatformType)
power_feed = ObjectField(PowerFeedType)
power_feed_list = ObjectListField(PowerFeedType)
power_outlet = ObjectField(PowerOutletType)
power_outlet_list = ObjectListField(PowerOutletType)
power_outlet_template = ObjectField(PowerOutletTemplateType)
power_outlet_template_list = ObjectListField(PowerOutletTemplateType)
power_panel = ObjectField(PowerPanelType)
power_panel_list = ObjectListField(PowerPanelType)
power_port = ObjectField(PowerPortType)
power_port_list = ObjectListField(PowerPortType)
power_port_template = ObjectField(PowerPortTemplateType)
power_port_template_list = ObjectListField(PowerPortTemplateType)
rack = ObjectField(RackType)
rack_list = ObjectListField(RackType)
rack_reservation = ObjectField(RackReservationType)
rack_reservation_list = ObjectListField(RackReservationType)
rack_role = ObjectField(RackRoleType)
rack_role_list = ObjectListField(RackRoleType)
rear_port = ObjectField(RearPortType)
rear_port_list = ObjectListField(RearPortType)
rear_port_template = ObjectField(RearPortTemplateType)
rear_port_template_list = ObjectListField(RearPortTemplateType)
region = ObjectField(RegionType)
region_list = ObjectListField(RegionType)
site = ObjectField(SiteType)
site_list = ObjectListField(SiteType)
site_group = ObjectField(SiteGroupType)
site_group_list = ObjectListField(SiteGroupType)
virtual_chassis = ObjectField(VirtualChassisType)
virtual_chassis_list = ObjectListField(VirtualChassisType)

View File

@ -0,0 +1,353 @@
from dcim import filtersets, models
from netbox.graphql.types import BaseObjectType, ObjectType, TaggedObjectType
__all__ = (
'CableType',
'ConsolePortType',
'ConsolePortTemplateType',
'ConsoleServerPortType',
'ConsoleServerPortTemplateType',
'DeviceType',
'DeviceBayType',
'DeviceBayTemplateType',
'DeviceRoleType',
'DeviceTypeType',
'FrontPortType',
'FrontPortTemplateType',
'InterfaceType',
'InterfaceTemplateType',
'InventoryItemType',
'LocationType',
'ManufacturerType',
'PlatformType',
'PowerFeedType',
'PowerOutletType',
'PowerOutletTemplateType',
'PowerPanelType',
'PowerPortType',
'PowerPortTemplateType',
'RackType',
'RackReservationType',
'RackRoleType',
'RearPortType',
'RearPortTemplateType',
'RegionType',
'SiteType',
'SiteGroupType',
'VirtualChassisType',
)
class CableType(TaggedObjectType):
class Meta:
model = models.Cable
fields = '__all__'
filterset_class = filtersets.CableFilterSet
def resolve_type(self, info):
return self.type or None
def resolve_length_unit(self, info):
return self.length_unit or None
class ConsolePortType(TaggedObjectType):
class Meta:
model = models.ConsolePort
exclude = ('_path',)
filterset_class = filtersets.ConsolePortFilterSet
def resolve_type(self, info):
return self.type or None
class ConsolePortTemplateType(BaseObjectType):
class Meta:
model = models.ConsolePortTemplate
fields = '__all__'
filterset_class = filtersets.ConsolePortTemplateFilterSet
def resolve_type(self, info):
return self.type or None
class ConsoleServerPortType(TaggedObjectType):
class Meta:
model = models.ConsoleServerPort
exclude = ('_path',)
filterset_class = filtersets.ConsoleServerPortFilterSet
def resolve_type(self, info):
return self.type or None
class ConsoleServerPortTemplateType(BaseObjectType):
class Meta:
model = models.ConsoleServerPortTemplate
fields = '__all__'
filterset_class = filtersets.ConsoleServerPortTemplateFilterSet
def resolve_type(self, info):
return self.type or None
class DeviceType(TaggedObjectType):
class Meta:
model = models.Device
fields = '__all__'
filterset_class = filtersets.DeviceFilterSet
def resolve_face(self, info):
return self.face or None
class DeviceBayType(TaggedObjectType):
class Meta:
model = models.DeviceBay
fields = '__all__'
filterset_class = filtersets.DeviceBayFilterSet
class DeviceBayTemplateType(BaseObjectType):
class Meta:
model = models.DeviceBayTemplate
fields = '__all__'
filterset_class = filtersets.DeviceBayTemplateFilterSet
class DeviceRoleType(ObjectType):
class Meta:
model = models.DeviceRole
fields = '__all__'
filterset_class = filtersets.DeviceRoleFilterSet
class DeviceTypeType(TaggedObjectType):
class Meta:
model = models.DeviceType
fields = '__all__'
filterset_class = filtersets.DeviceTypeFilterSet
def resolve_subdevice_role(self, info):
return self.subdevice_role or None
class FrontPortType(TaggedObjectType):
class Meta:
model = models.FrontPort
fields = '__all__'
filterset_class = filtersets.FrontPortFilterSet
class FrontPortTemplateType(BaseObjectType):
class Meta:
model = models.FrontPortTemplate
fields = '__all__'
filterset_class = filtersets.FrontPortTemplateFilterSet
class InterfaceType(TaggedObjectType):
class Meta:
model = models.Interface
exclude = ('_path',)
filterset_class = filtersets.InterfaceFilterSet
def resolve_mode(self, info):
return self.mode or None
class InterfaceTemplateType(BaseObjectType):
class Meta:
model = models.InterfaceTemplate
fields = '__all__'
filterset_class = filtersets.InterfaceTemplateFilterSet
class InventoryItemType(TaggedObjectType):
class Meta:
model = models.InventoryItem
fields = '__all__'
filterset_class = filtersets.InventoryItemFilterSet
class LocationType(ObjectType):
class Meta:
model = models.Location
fields = '__all__'
filterset_class = filtersets.LocationFilterSet
class ManufacturerType(ObjectType):
class Meta:
model = models.Manufacturer
fields = '__all__'
filterset_class = filtersets.ManufacturerFilterSet
class PlatformType(ObjectType):
class Meta:
model = models.Platform
fields = '__all__'
filterset_class = filtersets.PlatformFilterSet
class PowerFeedType(TaggedObjectType):
class Meta:
model = models.PowerFeed
exclude = ('_path',)
filterset_class = filtersets.PowerFeedFilterSet
class PowerOutletType(TaggedObjectType):
class Meta:
model = models.PowerOutlet
exclude = ('_path',)
filterset_class = filtersets.PowerOutletFilterSet
def resolve_feed_leg(self, info):
return self.feed_leg or None
def resolve_type(self, info):
return self.type or None
class PowerOutletTemplateType(BaseObjectType):
class Meta:
model = models.PowerOutletTemplate
fields = '__all__'
filterset_class = filtersets.PowerOutletTemplateFilterSet
def resolve_feed_leg(self, info):
return self.feed_leg or None
def resolve_type(self, info):
return self.type or None
class PowerPanelType(TaggedObjectType):
class Meta:
model = models.PowerPanel
fields = '__all__'
filterset_class = filtersets.PowerPanelFilterSet
class PowerPortType(TaggedObjectType):
class Meta:
model = models.PowerPort
exclude = ('_path',)
filterset_class = filtersets.PowerPortFilterSet
def resolve_type(self, info):
return self.type or None
class PowerPortTemplateType(BaseObjectType):
class Meta:
model = models.PowerPortTemplate
fields = '__all__'
filterset_class = filtersets.PowerPortTemplateFilterSet
def resolve_type(self, info):
return self.type or None
class RackType(TaggedObjectType):
class Meta:
model = models.Rack
fields = '__all__'
filterset_class = filtersets.RackFilterSet
def resolve_type(self, info):
return self.type or None
def resolve_outer_unit(self, info):
return self.outer_unit or None
class RackReservationType(TaggedObjectType):
class Meta:
model = models.RackReservation
fields = '__all__'
filterset_class = filtersets.RackReservationFilterSet
class RackRoleType(ObjectType):
class Meta:
model = models.RackRole
fields = '__all__'
filterset_class = filtersets.RackRoleFilterSet
class RearPortType(TaggedObjectType):
class Meta:
model = models.RearPort
fields = '__all__'
filterset_class = filtersets.RearPortFilterSet
class RearPortTemplateType(BaseObjectType):
class Meta:
model = models.RearPortTemplate
fields = '__all__'
filterset_class = filtersets.RearPortTemplateFilterSet
class RegionType(ObjectType):
class Meta:
model = models.Region
fields = '__all__'
filterset_class = filtersets.RegionFilterSet
class SiteType(TaggedObjectType):
class Meta:
model = models.Site
fields = '__all__'
filterset_class = filtersets.SiteFilterSet
class SiteGroupType(ObjectType):
class Meta:
model = models.SiteGroup
fields = '__all__'
filterset_class = filtersets.SiteGroupFilterSet
class VirtualChassisType(TaggedObjectType):
class Meta:
model = models.VirtualChassis
fields = '__all__'
filterset_class = filtersets.VirtualChassisFilterSet

View File

@ -25,6 +25,7 @@ __all__ = (
'Interface',
'InterfaceTemplate',
'InventoryItem',
'Location',
'Manufacturer',
'Platform',
'PowerFeed',
@ -34,7 +35,6 @@ __all__ = (
'PowerPort',
'PowerPortTemplate',
'Rack',
'Location',
'RackReservation',
'RackRole',
'RearPort',

View File

View File

@ -0,0 +1,30 @@
import graphene
from netbox.graphql.fields import ObjectField, ObjectListField
from .types import *
class ExtrasQuery(graphene.ObjectType):
config_context = ObjectField(ConfigContextType)
config_context_list = ObjectListField(ConfigContextType)
custom_field = ObjectField(CustomFieldType)
custom_field_list = ObjectListField(CustomFieldType)
custom_link = ObjectField(CustomLinkType)
custom_link_list = ObjectListField(CustomLinkType)
export_template = ObjectField(ExportTemplateType)
export_template_list = ObjectListField(ExportTemplateType)
image_attachment = ObjectField(ImageAttachmentType)
image_attachment_list = ObjectListField(ImageAttachmentType)
journal_entry = ObjectField(JournalEntryType)
journal_entry_list = ObjectListField(JournalEntryType)
tag = ObjectField(TagType)
tag_list = ObjectListField(TagType)
webhook = ObjectField(WebhookType)
webhook_list = ObjectListField(WebhookType)

View File

@ -0,0 +1,77 @@
from extras import filtersets, models
from netbox.graphql.types import BaseObjectType
__all__ = (
'ConfigContextType',
'CustomFieldType',
'CustomLinkType',
'ExportTemplateType',
'ImageAttachmentType',
'JournalEntryType',
'TagType',
'WebhookType',
)
class ConfigContextType(BaseObjectType):
class Meta:
model = models.ConfigContext
fields = '__all__'
filterset_class = filtersets.ConfigContextFilterSet
class CustomFieldType(BaseObjectType):
class Meta:
model = models.CustomField
fields = '__all__'
filterset_class = filtersets.CustomFieldFilterSet
class CustomLinkType(BaseObjectType):
class Meta:
model = models.CustomLink
fields = '__all__'
filterset_class = filtersets.CustomLinkFilterSet
class ExportTemplateType(BaseObjectType):
class Meta:
model = models.ExportTemplate
fields = '__all__'
filterset_class = filtersets.ExportTemplateFilterSet
class ImageAttachmentType(BaseObjectType):
class Meta:
model = models.ImageAttachment
fields = '__all__'
filterset_class = filtersets.ImageAttachmentFilterSet
class JournalEntryType(BaseObjectType):
class Meta:
model = models.JournalEntry
fields = '__all__'
filterset_class = filtersets.JournalEntryFilterSet
class TagType(BaseObjectType):
class Meta:
model = models.Tag
exclude = ('extras_taggeditem_items',)
filterset_class = filtersets.TagFilterSet
class WebhookType(BaseObjectType):
class Meta:
model = models.Webhook
fields = '__all__'
filterset_class = filtersets.WebhookFilterSet

View File

@ -270,7 +270,8 @@ class TagTest(APIViewTestCases.APIViewTestCase):
class ImageAttachmentTest(
APIViewTestCases.GetObjectViewTestCase,
APIViewTestCases.ListObjectsViewTestCase,
APIViewTestCases.DeleteObjectViewTestCase
APIViewTestCases.DeleteObjectViewTestCase,
APIViewTestCases.GraphQLTestCase
):
model = ImageAttachment
brief_fields = ['display', 'id', 'image', 'name', 'url']

View File

View File

@ -0,0 +1,36 @@
import graphene
from netbox.graphql.fields import ObjectField, ObjectListField
from .types import *
class IPAMQuery(graphene.ObjectType):
aggregate = ObjectField(AggregateType)
aggregate_list = ObjectListField(AggregateType)
ip_address = ObjectField(IPAddressType)
ip_address_list = ObjectListField(IPAddressType)
prefix = ObjectField(PrefixType)
prefix_list = ObjectListField(PrefixType)
rir = ObjectField(RIRType)
rir_list = ObjectListField(RIRType)
role = ObjectField(RoleType)
role_list = ObjectListField(RoleType)
route_target = ObjectField(RouteTargetType)
route_target_list = ObjectListField(RouteTargetType)
service = ObjectField(ServiceType)
service_list = ObjectListField(ServiceType)
vlan = ObjectField(VLANType)
vlan_list = ObjectListField(VLANType)
vlan_group = ObjectField(VLANGroupType)
vlan_group_list = ObjectListField(VLANGroupType)
vrf = ObjectField(VRFType)
vrf_list = ObjectListField(VRFType)

View File

@ -0,0 +1,98 @@
from ipam import filtersets, models
from netbox.graphql.types import ObjectType, TaggedObjectType
__all__ = (
'AggregateType',
'IPAddressType',
'PrefixType',
'RIRType',
'RoleType',
'RouteTargetType',
'ServiceType',
'VLANType',
'VLANGroupType',
'VRFType',
)
class AggregateType(TaggedObjectType):
class Meta:
model = models.Aggregate
fields = '__all__'
filterset_class = filtersets.AggregateFilterSet
class IPAddressType(TaggedObjectType):
class Meta:
model = models.IPAddress
fields = '__all__'
filterset_class = filtersets.IPAddressFilterSet
def resolve_role(self, info):
return self.role or None
class PrefixType(TaggedObjectType):
class Meta:
model = models.Prefix
fields = '__all__'
filterset_class = filtersets.PrefixFilterSet
class RIRType(ObjectType):
class Meta:
model = models.RIR
fields = '__all__'
filterset_class = filtersets.RIRFilterSet
class RoleType(ObjectType):
class Meta:
model = models.Role
fields = '__all__'
filterset_class = filtersets.RoleFilterSet
class RouteTargetType(TaggedObjectType):
class Meta:
model = models.RouteTarget
fields = '__all__'
filterset_class = filtersets.RouteTargetFilterSet
class ServiceType(TaggedObjectType):
class Meta:
model = models.Service
fields = '__all__'
filterset_class = filtersets.ServiceFilterSet
class VLANType(TaggedObjectType):
class Meta:
model = models.VLAN
fields = '__all__'
filterset_class = filtersets.VLANFilterSet
class VLANGroupType(ObjectType):
class Meta:
model = models.VLANGroup
fields = '__all__'
filterset_class = filtersets.VLANGroupFilterSet
class VRFType(TaggedObjectType):
class Meta:
model = models.VRF
fields = '__all__'
filterset_class = filtersets.VRFFilterSet

View File

@ -8,3 +8,7 @@ class ServiceUnavailable(APIException):
class SerializerNotFound(Exception):
pass
class GraphQLTypeNotFound(Exception):
pass

View File

@ -149,6 +149,9 @@ EXEMPT_VIEW_PERMISSIONS = [
# 'ipam.prefix',
]
# Enable the GraphQL API
GRAPHQL_ENABLED = True
# HTTP proxies NetBox should use when sending outbound HTTP requests (e.g. for webhooks).
# HTTP_PROXIES = {
# 'http': 'http://10.10.1.10:3128',

View File

@ -0,0 +1,22 @@
import graphene
from graphene_django.converter import convert_django_field
from taggit.managers import TaggableManager
from dcim.fields import MACAddressField
from ipam.fields import IPAddressField, IPNetworkField
@convert_django_field.register(TaggableManager)
def convert_field_to_tags_list(field, registry=None):
"""
Register conversion handler for django-taggit's TaggableManager
"""
return graphene.List(graphene.String)
@convert_django_field.register(IPAddressField)
@convert_django_field.register(IPNetworkField)
@convert_django_field.register(MACAddressField)
def convert_field_to_string(field, registry=None):
# TODO: Update to use get_django_field_description under django_graphene v3.0
return graphene.String(description=field.help_text, required=not field.null)

View File

@ -0,0 +1,65 @@
from functools import partial
import graphene
from graphene_django import DjangoListField
from .utils import get_graphene_type
__all__ = (
'ObjectField',
'ObjectListField',
)
class ObjectField(graphene.Field):
"""
Retrieve a single object, identified by its numeric ID.
"""
def __init__(self, *args, **kwargs):
if 'id' not in kwargs:
kwargs['id'] = graphene.Int(required=True)
super().__init__(*args, **kwargs)
@staticmethod
def object_resolver(django_object_type, root, info, **args):
"""
Return an object given its numeric ID.
"""
manager = django_object_type._meta.model._default_manager
queryset = django_object_type.get_queryset(manager, info)
return queryset.get(**args)
def get_resolver(self, parent_resolver):
return partial(self.object_resolver, self._type)
class ObjectListField(DjangoListField):
"""
Retrieve a list of objects, optionally filtered by one or more FilterSet filters.
"""
def __init__(self, _type, *args, **kwargs):
assert hasattr(_type._meta, 'filterset_class'), "DjangoFilterListField must define filterset_class under Meta"
filterset_class = _type._meta.filterset_class
# Get FilterSet kwargs
filter_kwargs = {}
for filter_name, filter_field in filterset_class.get_filters().items():
field_type = get_graphene_type(type(filter_field))
filter_kwargs[filter_name] = graphene.Argument(field_type)
super().__init__(_type, args=filter_kwargs, *args, **kwargs)
@staticmethod
def list_resolver(django_object_type, resolver, default_manager, root, info, **args):
# Get the QuerySet from the object type
queryset = django_object_type.get_queryset(default_manager, info)
# Instantiate and apply the FilterSet
filterset_class = django_object_type._meta.filterset_class
filterset = filterset_class(data=args, queryset=queryset, request=info.context)
return filterset.qs

View File

@ -0,0 +1,25 @@
import graphene
from circuits.graphql.schema import CircuitsQuery
from dcim.graphql.schema import DCIMQuery
from extras.graphql.schema import ExtrasQuery
from ipam.graphql.schema import IPAMQuery
from tenancy.graphql.schema import TenancyQuery
from users.graphql.schema import UsersQuery
from virtualization.graphql.schema import VirtualizationQuery
class Query(
CircuitsQuery,
DCIMQuery,
ExtrasQuery,
IPAMQuery,
TenancyQuery,
UsersQuery,
VirtualizationQuery,
graphene.ObjectType
):
pass
schema = graphene.Schema(query=Query, auto_camelcase=False)

View File

@ -0,0 +1,64 @@
import graphene
from django.contrib.contenttypes.models import ContentType
from graphene.types.generic import GenericScalar
from graphene_django import DjangoObjectType
__all__ = (
'BaseObjectType',
'ObjectType',
'TaggedObjectType',
)
#
# Base types
#
class BaseObjectType(DjangoObjectType):
"""
Base GraphQL object type for all NetBox objects
"""
class Meta:
abstract = True
@classmethod
def get_queryset(cls, queryset, info):
# Enforce object permissions on the queryset
return queryset.restrict(info.context.user, 'view')
class ObjectType(BaseObjectType):
"""
Extends BaseObjectType with support for custom field data.
"""
custom_fields = GenericScalar()
class Meta:
abstract = True
def resolve_custom_fields(self, info):
return self.custom_field_data
class TaggedObjectType(ObjectType):
"""
Extends ObjectType with support for Tags
"""
tags = graphene.List(graphene.String)
class Meta:
abstract = True
def resolve_tags(self, info):
return self.tags.all()
#
# Miscellaneous types
#
class ContentTypeType(DjangoObjectType):
class Meta:
model = ContentType
fields = ('id', 'app_label', 'model')

View File

@ -0,0 +1,25 @@
import graphene
from django_filters import filters
def get_graphene_type(filter_cls):
"""
Return the appropriate Graphene scalar type for a django_filters Filter
"""
if issubclass(filter_cls, filters.BooleanFilter):
field_type = graphene.Boolean
elif issubclass(filter_cls, filters.NumberFilter):
# TODO: Floats? BigInts?
field_type = graphene.Int
elif issubclass(filter_cls, filters.DateFilter):
field_type = graphene.Date
elif issubclass(filter_cls, filters.DateTimeFilter):
field_type = graphene.DateTime
else:
field_type = graphene.String
# Multi-value filters should be handled as lists
if issubclass(filter_cls, filters.MultipleChoiceFilter):
return graphene.List(field_type)
return field_type

View File

@ -0,0 +1,40 @@
from django.conf import settings
from django.contrib.auth.views import redirect_to_login
from django.http import HttpResponseNotFound, HttpResponseForbidden
from django.urls import reverse
from graphene_django.views import GraphQLView as GraphQLView_
from rest_framework.exceptions import AuthenticationFailed
from netbox.api.authentication import TokenAuthentication
class GraphQLView(GraphQLView_):
"""
Extends graphene_django's GraphQLView to support DRF's token-based authentication.
"""
def dispatch(self, request, *args, **kwargs):
# Enforce GRAPHQL_ENABLED
if not settings.GRAPHQL_ENABLED:
return HttpResponseNotFound("The GraphQL API is not enabled.")
# Attempt to authenticate the user using a DRF token, if provided
if not request.user.is_authenticated:
authenticator = TokenAuthentication()
try:
auth_info = authenticator.authenticate(request)
if auth_info is not None:
request.user = auth_info[0] # User object
except AuthenticationFailed as exc:
return HttpResponseForbidden(exc.detail)
# Enforce LOGIN_REQUIRED
if settings.LOGIN_REQUIRED and not request.user.is_authenticated:
# If this is a human user, send a redirect to the login page
if self.request_wants_html(request):
return redirect_to_login(reverse('graphql'))
return HttpResponseForbidden("No credentials provided.")
return super().dispatch(request, *args, **kwargs)

View File

@ -24,7 +24,8 @@ class LoginRequiredMiddleware(object):
if settings.LOGIN_REQUIRED and not request.user.is_authenticated:
# Determine exempt paths
exempt_paths = [
reverse('api-root')
reverse('api-root'),
reverse('graphql'),
]
if settings.METRICS_ENABLED:
exempt_paths.append(reverse('prometheus-django-metrics'))

View File

@ -83,6 +83,7 @@ DOCS_ROOT = getattr(configuration, 'DOCS_ROOT', os.path.join(os.path.dirname(BAS
EMAIL = getattr(configuration, 'EMAIL', {})
ENFORCE_GLOBAL_UNIQUE = getattr(configuration, 'ENFORCE_GLOBAL_UNIQUE', False)
EXEMPT_VIEW_PERMISSIONS = getattr(configuration, 'EXEMPT_VIEW_PERMISSIONS', [])
GRAPHQL_ENABLED = getattr(configuration, 'GRAPHQL_ENABLED', True)
HTTP_PROXIES = getattr(configuration, 'HTTP_PROXIES', None)
INTERNAL_IPS = getattr(configuration, 'INTERNAL_IPS', ('127.0.0.1', '::1'))
LOGGING = getattr(configuration, 'LOGGING', {})
@ -282,9 +283,11 @@ INSTALLED_APPS = [
'cacheops',
'corsheaders',
'debug_toolbar',
'graphiql_debug_toolbar',
'django_filters',
'django_tables2',
'django_prometheus',
'graphene_django',
'mptt',
'rest_framework',
'taggit',
@ -303,7 +306,7 @@ INSTALLED_APPS = [
# Middleware
MIDDLEWARE = [
'debug_toolbar.middleware.DebugToolbarMiddleware',
'graphiql_debug_toolbar.middleware.DebugToolbarMiddleware',
'django_prometheus.middleware.PrometheusBeforeMiddleware',
'corsheaders.middleware.CorsMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
@ -494,6 +497,17 @@ REST_FRAMEWORK = {
}
#
# Graphene
#
GRAPHENE = {
# Avoids naming collision on models with 'type' field; see
# https://github.com/graphql-python/graphene-django/issues/185
'DJANGO_CHOICE_FIELD_ENUM_V3_NAMING': True,
}
#
# drf_yasg (OpenAPI/Swagger)
#

View File

@ -0,0 +1,36 @@
from django.test import override_settings
from django.urls import reverse
from utilities.testing import disable_warnings, TestCase
class GraphQLTestCase(TestCase):
@override_settings(GRAPHQL_ENABLED=False)
def test_graphql_enabled(self):
"""
The /graphql URL should return a 404 when GRAPHQL_ENABLED=False
"""
url = reverse('graphql')
response = self.client.get(url)
self.assertHttpStatus(response, 404)
@override_settings(LOGIN_REQUIRED=True)
def test_graphiql_interface(self):
"""
Test rendering of the GraphiQL interactive web interface
"""
url = reverse('graphql')
header = {
'HTTP_ACCEPT': 'text/html',
}
# Authenticated request
response = self.client.get(url, **header)
self.assertHttpStatus(response, 200)
# Non-authenticated request
self.client.logout()
response = self.client.get(url, **header)
with disable_warnings('django.request'):
self.assertHttpStatus(response, 302) # Redirect to login page

View File

@ -7,6 +7,8 @@ from drf_yasg.views import get_schema_view
from extras.plugins.urls import plugin_admin_patterns, plugin_patterns, plugin_api_patterns
from netbox.api.views import APIRootView, StatusView
from netbox.graphql.schema import schema
from netbox.graphql.views import GraphQLView
from netbox.views import HomeView, StaticMediaFailureView, SearchView
from users.views import LoginView, LogoutView
from .admin import admin_site
@ -60,6 +62,9 @@ _patterns = [
path('api/redoc/', schema_view.with_ui('redoc'), name='api_redocs'),
re_path(r'^api/swagger(?P<format>.json|.yaml)$', schema_view.without_ui(), name='schema_swagger'),
# GraphQL
path('graphql/', GraphQLView.as_view(graphiql=True, schema=schema), name='graphql'),
# Serving static media in Django to pipe it through LoginRequiredMiddleware
path('media/<path:path>', serve, {'document_root': settings.MEDIA_ROOT}),
path('media-failure/', StaticMediaFailureView.as_view(), name='media_failure'),

@ -0,0 +1 @@
Subproject commit 942aa8c7bd506fb88b7c669cab173bc319eca309

View File

View File

@ -0,0 +1,12 @@
import graphene
from netbox.graphql.fields import ObjectField, ObjectListField
from .types import *
class TenancyQuery(graphene.ObjectType):
tenant = ObjectField(TenantType)
tenant_list = ObjectListField(TenantType)
tenant_group = ObjectField(TenantGroupType)
tenant_group_list = ObjectListField(TenantGroupType)

View File

@ -0,0 +1,23 @@
from tenancy import filtersets, models
from netbox.graphql.types import ObjectType, TaggedObjectType
__all__ = (
'TenantType',
'TenantGroupType',
)
class TenantType(TaggedObjectType):
class Meta:
model = models.Tenant
fields = '__all__'
filterset_class = filtersets.TenantFilterSet
class TenantGroupType(ObjectType):
class Meta:
model = models.TenantGroup
fields = '__all__'
filterset_class = filtersets.TenantGroupFilterSet

View File

View File

@ -0,0 +1,12 @@
import graphene
from netbox.graphql.fields import ObjectField, ObjectListField
from .types import *
class UsersQuery(graphene.ObjectType):
group = ObjectField(GroupType)
group_list = ObjectListField(GroupType)
user = ObjectField(UserType)
user_list = ObjectListField(UserType)

View File

@ -0,0 +1,37 @@
from django.contrib.auth.models import Group, User
from graphene_django import DjangoObjectType
from users import filtersets
from utilities.querysets import RestrictedQuerySet
__all__ = (
'GroupType',
'UserType',
)
class GroupType(DjangoObjectType):
class Meta:
model = Group
fields = ('id', 'name')
filterset_class = filtersets.GroupFilterSet
@classmethod
def get_queryset(cls, queryset, info):
return RestrictedQuerySet(model=Group)
class UserType(DjangoObjectType):
class Meta:
model = User
fields = (
'id', 'username', 'password', 'first_name', 'last_name', 'email', 'is_staff', 'is_active', 'date_joined',
'groups',
)
filterset_class = filtersets.UserFilterSet
@classmethod
def get_queryset(cls, queryset, info):
return RestrictedQuerySet(model=User)

View File

@ -75,7 +75,14 @@ class GroupTest(APIViewTestCases.APIViewTestCase):
Group.objects.bulk_create(users)
class TokenTest(APIViewTestCases.APIViewTestCase):
class TokenTest(
# No GraphQL support for Token
APIViewTestCases.GetObjectViewTestCase,
APIViewTestCases.ListObjectsViewTestCase,
APIViewTestCases.CreateObjectViewTestCase,
APIViewTestCases.UpdateObjectViewTestCase,
APIViewTestCases.DeleteObjectViewTestCase
):
model = Token
brief_fields = ['display', 'id', 'key', 'url', 'write_enabled']
bulk_update_data = {
@ -138,7 +145,14 @@ class TokenTest(APIViewTestCases.APIViewTestCase):
self.assertEqual(response.status_code, 403)
class ObjectPermissionTest(APIViewTestCases.APIViewTestCase):
class ObjectPermissionTest(
# No GraphQL support for ObjectPermission
APIViewTestCases.GetObjectViewTestCase,
APIViewTestCases.ListObjectsViewTestCase,
APIViewTestCases.CreateObjectViewTestCase,
APIViewTestCases.UpdateObjectViewTestCase,
APIViewTestCases.DeleteObjectViewTestCase
):
model = ObjectPermission
brief_fields = ['actions', 'display', 'enabled', 'groups', 'id', 'name', 'object_types', 'url', 'users']

View File

@ -7,7 +7,7 @@ from django.urls import reverse
from rest_framework import status
from rest_framework.utils import formatting
from netbox.api.exceptions import SerializerNotFound
from netbox.api.exceptions import GraphQLTypeNotFound, SerializerNotFound
from .utils import dynamic_import
@ -24,10 +24,25 @@ def get_serializer_for_model(model, prefix=''):
return dynamic_import(serializer_name)
except AttributeError:
raise SerializerNotFound(
"Could not determine serializer for {}.{} with prefix '{}'".format(app_name, model_name, prefix)
f"Could not determine serializer for {app_name}.{model_name} with prefix '{prefix}'"
)
def get_graphql_type_for_model(model):
"""
Return the GraphQL type class for the given model.
"""
app_name, model_name = model._meta.label.split('.')
# Object types for Django's auth models are in the users app
if app_name == 'auth':
app_name = 'users'
class_name = f'{app_name}.graphql.types.{model_name}Type'
try:
return dynamic_import(class_name)
except AttributeError:
raise GraphQLTypeNotFound(f"Could not find GraphQL type for {app_name}.{model_name}")
def is_api_request(request):
"""
Return True of the request is being made via the REST API.

View File

@ -1,14 +1,18 @@
import json
from django.conf import settings
from django.contrib.auth.models import User
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.test import override_settings
from graphene.types.dynamic import Dynamic
from rest_framework import status
from rest_framework.test import APIClient
from extras.choices import ObjectChangeActionChoices
from extras.models import ObjectChange
from users.models import ObjectPermission, Token
from utilities.api import get_graphql_type_for_model
from .base import ModelTestCase
from .utils import disable_warnings
@ -20,7 +24,7 @@ __all__ = (
#
# REST API Tests
# REST/GraphQL API Tests
#
class APITestCase(ModelTestCase):
@ -421,11 +425,99 @@ class APIViewTestCases:
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(self._get_queryset().count(), initial_count - 3)
class GraphQLTestCase(APITestCase):
def _get_graphql_base_name(self):
"""
Return graphql_base_name, if set. Otherwise, construct the base name for the query
field from the model's verbose name.
"""
base_name = self.model._meta.verbose_name.lower().replace(' ', '_')
return getattr(self, 'graphql_base_name', base_name)
def _build_query(self, name, **filters):
type_class = get_graphql_type_for_model(self.model)
if filters:
filter_string = ', '.join(f'{k}:{v}' for k, v in filters.items())
filter_string = f'({filter_string})'
else:
filter_string = ''
# Compile list of fields to include
fields_string = ''
for field_name, field in type_class._meta.fields.items():
if type(field) is Dynamic:
# Dynamic fields must specify a subselection
fields_string += f'{field_name} {{ id }}\n'
else:
fields_string += f'{field_name}\n'
query = f"""
{{
{name}{filter_string} {{
{fields_string}
}}
}}
"""
return query
@override_settings(LOGIN_REQUIRED=True)
def test_graphql_get_object(self):
url = reverse('graphql')
field_name = self._get_graphql_base_name()
object_id = self._get_queryset().first().pk
query = self._build_query(field_name, id=object_id)
# Non-authenticated requests should fail
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(url, data={'query': query}), status.HTTP_403_FORBIDDEN)
# Add object-level permission
obj_perm = ObjectPermission(
name='Test permission',
actions=['view']
)
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))
response = self.client.post(url, data={'query': query}, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
data = json.loads(response.content)
self.assertNotIn('errors', data)
@override_settings(LOGIN_REQUIRED=True)
def test_graphql_list_objects(self):
url = reverse('graphql')
field_name = f'{self._get_graphql_base_name()}_list'
query = self._build_query(field_name)
# Non-authenticated requests should fail
with disable_warnings('django.request'):
self.assertHttpStatus(self.client.post(url, data={'query': query}), status.HTTP_403_FORBIDDEN)
# Add object-level permission
obj_perm = ObjectPermission(
name='Test permission',
actions=['view']
)
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))
response = self.client.post(url, data={'query': query}, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
data = json.loads(response.content)
self.assertNotIn('errors', data)
self.assertGreater(len(data['data'][field_name]), 0)
class APIViewTestCase(
GetObjectViewTestCase,
ListObjectsViewTestCase,
CreateObjectViewTestCase,
UpdateObjectViewTestCase,
DeleteObjectViewTestCase
DeleteObjectViewTestCase,
GraphQLTestCase
):
pass

View File

@ -0,0 +1,21 @@
import graphene
from netbox.graphql.fields import ObjectField, ObjectListField
from .types import *
class VirtualizationQuery(graphene.ObjectType):
cluster = ObjectField(ClusterType)
cluster_list = ObjectListField(ClusterType)
cluster_group = ObjectField(ClusterGroupType)
cluster_group_list = ObjectListField(ClusterGroupType)
cluster_type = ObjectField(ClusterTypeType)
cluster_type_list = ObjectListField(ClusterTypeType)
virtual_machine = ObjectField(VirtualMachineType)
virtual_machine_list = ObjectListField(VirtualMachineType)
vm_interface = ObjectField(VMInterfaceType)
vm_interface_list = ObjectListField(VMInterfaceType)

View File

@ -0,0 +1,53 @@
from virtualization import filtersets, models
from netbox.graphql.types import ObjectType, TaggedObjectType
__all__ = (
'ClusterType',
'ClusterGroupType',
'ClusterTypeType',
'VirtualMachineType',
'VMInterfaceType',
)
class ClusterType(TaggedObjectType):
class Meta:
model = models.Cluster
fields = '__all__'
filterset_class = filtersets.ClusterFilterSet
class ClusterGroupType(ObjectType):
class Meta:
model = models.ClusterGroup
fields = '__all__'
filterset_class = filtersets.ClusterGroupFilterSet
class ClusterTypeType(ObjectType):
class Meta:
model = models.ClusterType
fields = '__all__'
filterset_class = filtersets.ClusterTypeFilterSet
class VirtualMachineType(TaggedObjectType):
class Meta:
model = models.VirtualMachine
fields = '__all__'
filterset_class = filtersets.VirtualMachineFilterSet
class VMInterfaceType(TaggedObjectType):
class Meta:
model = models.VMInterface
fields = '__all__'
filterset_class = filtersets.VMInterfaceFilterSet
def resolve_mode(self, info):
return self.mode or None

View File

@ -211,6 +211,7 @@ class VMInterfaceTest(APIViewTestCases.APIViewTestCase):
bulk_update_data = {
'description': 'New description',
}
graphql_base_name = 'vm_interface'
@classmethod
def setUpTestData(cls):

View File

@ -3,6 +3,7 @@ django-cacheops==6.0
django-cors-headers==3.7.0
django-debug-toolbar==3.2.1
django-filter==2.4.0
django-graphiql-debug-toolbar==0.1.4
django-mptt==0.12.0
django-pglocks==1.0.4
django-prometheus==2.1.0
@ -12,6 +13,7 @@ django-taggit==1.4.0
django-timezone-field==4.1.2
djangorestframework==3.12.4
drf-yasg[validation]==1.20.0
graphene_django==2.15.0
gunicorn==20.1.0
Jinja2==3.0.1
Markdown==3.3.4