Skip to content

Nautobot App Livedata API Package

nautobot_app_livedata.api

REST API module for Nautobot App Livedata.

serializers

Nautobot App Livedata API Serializers.

LivedataJobResultSerializer

Bases: Serializer

Serializer for the Nautobot App Livedata API to return the Job Result.

This serializer is used to get the job result for the given jobresult_id.

Properties:

  • jobresult_id (UUID): The job result ID of the job that was enqueued.

Raises: - ValidationError: If the job result ID is not defined. - ValidationError: If the job result is not found.

Source code in nautobot_app_livedata/api/serializers.py 
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
91
class LivedataJobResultSerializer(serializers.Serializer):
    """Serializer for the Nautobot App Livedata API to return the Job Result.

    This serializer is used to get the job result for the given jobresult_id.

    Properties:

    - jobresult_id (UUID): The job result ID of the job that was enqueued.

    Raises:
    - ValidationError: If the job result ID is not defined.
    - ValidationError: If the job result is not found.

    """

    jobresult_id = serializers.UUIDField(required=True, allow_null=False)

    def validate(self, attrs):
        """Validate the job result ID."""
        if "jobresult_id" not in attrs:
            raise serializers.ValidationError("The job result ID is not defined", code="invalid")
        return attrs

    def create(self, validated_data):
        """Serializer does not create any objects."""
        raise NotImplementedError

    def update(self, instance, validated_data):
        """Serializer does not update any objects."""
        raise NotImplementedError
create(validated_data)

Serializer does not create any objects.

Source code in nautobot_app_livedata/api/serializers.py 
85
86
87
def create(self, validated_data):
    """Serializer does not create any objects."""
    raise NotImplementedError
update(instance, validated_data)

Serializer does not update any objects.

Source code in nautobot_app_livedata/api/serializers.py 
89
90
91
def update(self, instance, validated_data):
    """Serializer does not update any objects."""
    raise NotImplementedError
validate(attrs)

Validate the job result ID.

Source code in nautobot_app_livedata/api/serializers.py 
79
80
81
82
83
def validate(self, attrs):
    """Validate the job result ID."""
    if "jobresult_id" not in attrs:
        raise serializers.ValidationError("The job result ID is not defined", code="invalid")
    return attrs

LivedataSerializer

Bases: Serializer

Serializer for the Nautobot App Livedata API to return the Managed Device.

For more information on implementing jobs, refer to the Nautobot job documentation: https://docs.nautobot.com/projects/core/en/stable/development/jobs/

This serializer is used to get the managed device for the given object_type and ID.

Properties:

  • pk (UUID): The primary key of the object.
  • object_type (str): The object type to get the managed device for.

Raises: - ValidationError: If the object type is not defined. - ValidationError: If the object ID is not defined. - ValidationError: If the object_type is not valid. - ValidationError: If the object is not found. - ValidationError: If the object does not have a primary IP address. - ValidationError: If the object state is not Active.

Source code in nautobot_app_livedata/api/serializers.py 
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
class LivedataSerializer(serializers.Serializer):
    """Serializer for the Nautobot App Livedata API to return the Managed Device.

    For more information on implementing jobs, refer to the Nautobot job documentation:
    https://docs.nautobot.com/projects/core/en/stable/development/jobs/

    This serializer is used to get the managed device for the given object_type
    and ID.

    Properties:

    - pk (UUID): The primary key of the object.
    - object_type (str): The object type to get the managed device for.

    Raises:
    - ValidationError: If the object type is not defined.
    - ValidationError: If the object ID is not defined.
    - ValidationError: If the object_type is not valid.
    - ValidationError: If the object is not found.
    - ValidationError: If the object does not have a primary IP address.
    - ValidationError: If the object state is not Active.
    """

    pk = serializers.UUIDField(required=True, allow_null=False)
    object_type = serializers.CharField(required=True, allow_blank=False)

    queryset = Interface.objects.all()

    def validate(self, attrs):
        """Validate the object type and the device/interface ID."""
        # Check that the object_type is valid
        if "object_type" not in attrs:
            raise serializers.ValidationError("The object type is not defined", code="invalid")
        if "pk" not in attrs:
            raise serializers.ValidationError("The object ID is not defined", code="invalid")
        try:
            result = PrimaryDeviceUtils(object_type=attrs["object_type"], pk=attrs["pk"]).to_dict()
            attrs.update(result)
        except ValueError as err:
            raise serializers.ValidationError(str(err), code="invalid") from err
        return attrs

    def create(self, validated_data):
        """Serializer does not create any objects."""
        raise NotImplementedError

    def update(self, instance, validated_data):
        """Serializer does not update any objects."""
        raise NotImplementedError
create(validated_data)

Serializer does not create any objects.

Source code in nautobot_app_livedata/api/serializers.py 
53
54
55
def create(self, validated_data):
    """Serializer does not create any objects."""
    raise NotImplementedError
update(instance, validated_data)

Serializer does not update any objects.

Source code in nautobot_app_livedata/api/serializers.py 
57
58
59
def update(self, instance, validated_data):
    """Serializer does not update any objects."""
    raise NotImplementedError
validate(attrs)

Validate the object type and the device/interface ID.

Source code in nautobot_app_livedata/api/serializers.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
def validate(self, attrs):
    """Validate the object type and the device/interface ID."""
    # Check that the object_type is valid
    if "object_type" not in attrs:
        raise serializers.ValidationError("The object type is not defined", code="invalid")
    if "pk" not in attrs:
        raise serializers.ValidationError("The object ID is not defined", code="invalid")
    try:
        result = PrimaryDeviceUtils(object_type=attrs["object_type"], pk=attrs["pk"]).to_dict()
        attrs.update(result)
    except ValueError as err:
        raise serializers.ValidationError(str(err), code="invalid") from err
    return attrs

urls

Nautobot App Livedata API URLs.

views

Views for the Livedata API.

LivedataPrimaryDeviceApiView

Bases: GenericAPIView

Nautobot App Livedata API Primary Device view.

For more information on implementing jobs, refer to the Nautobot job documentation: https://docs.nautobot.com/projects/core/en/stable/development/jobs/

Source code in nautobot_app_livedata/api/views.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
class LivedataPrimaryDeviceApiView(GenericAPIView):
    """Nautobot App Livedata API Primary Device view.

    For more information on implementing jobs, refer to the Nautobot job documentation:
    https://docs.nautobot.com/projects/core/en/stable/development/jobs/
    """

    serializer_class = LivedataSerializer
    queryset = Device.objects.all()

    def get(
        self, request: Any, *args: Any, pk: Optional[int] = None, object_type: Optional[str] = None, **kwargs: Any
    ) -> Response:
        """Handle GET request for Livedata Primary Device API.

        Args:
            request (HttpRequest): The request object.
            pk: The primary key of the object.
            object_type (str): The object type.
            *args: Additional positional arguments.
            **kwargs: Additional keyword arguments.

        Returns:
            Response: The response object. "application/json"

                data = {
                    "object_type": "The object type to get the primary device for",
                    "pk": "The primary key",
                    "device": "The device ID of the device that is referred in object_type",
                    "interface": "The interface ID if the object type is 'dcim.interface'",
                    "virtual_chassis": "The virtual chassis ID if the object type is 'dcim.virtualchassis'",
                    "primary_device": "The primary device ID"
                }

        Raises:
            PermissionDenied: If the user does not have permission to access the object.
            NotFound: If the object does not exist.
        """
        data = request.data
        data["pk"] = pk
        data["object_type"] = object_type
        serializer = self.get_serializer(data=request.data)
        if not serializer.is_valid():
            # Return error response if serializer is not valid
            return Response(
                {
                    "error": "Invalid data provided",
                    "details": serializer.errors,
                },
                status=HTTPStatus.BAD_REQUEST,  # 400
            )
        try:
            result = serializer.validated_data
        except ValueError as error:
            return Response(
                f"Failed to get primary device: {error}",
                status=HTTPStatus.BAD_REQUEST,  # 400
            )
        return Response(data=result, status=HTTPStatus.OK)  # 200
get(request, *args, pk=None, object_type=None, **kwargs)

Handle GET request for Livedata Primary Device API.

Parameters:

Name Type Description Default
request HttpRequest

The request object.

 required
pk Optional[int]

The primary key of the object.

 None
object_type str

The object type.

 None
*args Any

Additional positional arguments.

 ()
**kwargs Any

Additional keyword arguments.

 {}

Returns:

Name Type Description
Response Response

The response object. "application/json"

data = { "object_type": "The object type to get the primary device for", "pk": "The primary key", "device": "The device ID of the device that is referred in object_type", "interface": "The interface ID if the object type is 'dcim.interface'", "virtual_chassis": "The virtual chassis ID if the object type is 'dcim.virtualchassis'", "primary_device": "The primary device ID" }

Raises:

Type Description
PermissionDenied

If the user does not have permission to access the object.
NotFound

If the object does not exist.
Source code in nautobot_app_livedata/api/views.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def get(
    self, request: Any, *args: Any, pk: Optional[int] = None, object_type: Optional[str] = None, **kwargs: Any
) -> Response:
    """Handle GET request for Livedata Primary Device API.

    Args:
        request (HttpRequest): The request object.
        pk: The primary key of the object.
        object_type (str): The object type.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Returns:
        Response: The response object. "application/json"

            data = {
                "object_type": "The object type to get the primary device for",
                "pk": "The primary key",
                "device": "The device ID of the device that is referred in object_type",
                "interface": "The interface ID if the object type is 'dcim.interface'",
                "virtual_chassis": "The virtual chassis ID if the object type is 'dcim.virtualchassis'",
                "primary_device": "The primary device ID"
            }

    Raises:
        PermissionDenied: If the user does not have permission to access the object.
        NotFound: If the object does not exist.
    """
    data = request.data
    data["pk"] = pk
    data["object_type"] = object_type
    serializer = self.get_serializer(data=request.data)
    if not serializer.is_valid():
        # Return error response if serializer is not valid
        return Response(
            {
                "error": "Invalid data provided",
                "details": serializer.errors,
            },
            status=HTTPStatus.BAD_REQUEST,  # 400
        )
    try:
        result = serializer.validated_data
    except ValueError as error:
        return Response(
            f"Failed to get primary device: {error}",
            status=HTTPStatus.BAD_REQUEST,  # 400
        )
    return Response(data=result, status=HTTPStatus.OK)  # 200

LivedataQueryApiView

Bases: GenericAPIView, ABC

Abstract Livedata Query API view.

Source code in nautobot_app_livedata/api/views.py 
 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
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class LivedataQueryApiView(GenericAPIView, ABC):
    """Abstract Livedata Query API view."""

    serializer_class = LivedataSerializer

    @abstractmethod
    def get_object_type(self) -> str:
        """Get the object type for the view.

        Returns:
            str: The object type.
        """

    @abstractmethod
    def get_commands(self, instance) -> list:
        """Get the commands to be executed for the given instance.

        Args:
            instance (Model): The model instance.

        Returns:
            list: The commands to be executed.
        """

    def get(self, request: Any, *args: Any, pk=None, **kwargs: Any) -> Response:  # pylint: disable=R0911
        """Handle GET request for Livedata Query API.

        The get method is used to enqueue the Livedata Query Job.

        To access the JobResult object, use the jobresult_id returned in the response
        and make a GET request to the JobResult endpoint.

        For Example:
            GET /api/extras/job-results/{jobresult_id}/

        Args:
            request (Request): The request object.
            pk (uuid): The primary key of the model instance.
            *args: Additional positional arguments.
            **kwargs: Additional keyword arguments.

        Returns:
            jobresult_id: The job result ID of the job that was enqueued.

        Raises:
            Response: If the user does not have permission to execute 'livedata' on the instance.
            Response: If the serializer is not valid.
            Response: If the job Livedata Api-Job is not found.
            Response: If the job failed to run.
        """
        logger.debug(
            "Resolved view=%s queryset_model=%s",
            self.__class__.__name__,
            getattr(getattr(self, "queryset", None), "model", None),
        )
        data = request.data
        data["pk"] = pk
        data["object_type"] = self.get_object_type()
        serializer = self.get_serializer(data=data)
        if not serializer.is_valid():
            return Response(
                serializer.errors,
                status=HTTPStatus.BAD_REQUEST,
            )
        # Check if user has permission to interact with the interface
        if not request.user.has_perm("dcim.can_interact_device"):
            return Response(
                {
                    "error": (
                        "You do not have the permission 'can_interact' for 'dcim.device'. Contact your administrator."
                    )
                },
                status=HTTPStatus.FORBIDDEN,  # 403
            )

        try:
            primary_device_info = serializer.validated_data
            if not primary_device_info:
                return Response(
                    {"error": "Primary device information is missing."},
                    status=HTTPStatus.BAD_REQUEST,
                )
            if data["object_type"] == "dcim.interface":
                self.queryset = Interface.objects.all()
                instance = Interface.objects.get(pk=pk)
            elif data["object_type"] == "dcim.device":
                self.queryset = Device.objects.all()
                instance = Device.objects.get(pk=primary_device_info["primary_device"])
            else:
                qs = self.get_queryset()
                instance = qs.get(pk=pk)
            show_commands_j2_array = self.get_commands(instance)
        except ValueError as error:
            logger.error("Error during Livedata Query API: %s", error)
            return Response(
                "An error occurred during the Livedata Query API request.",
                status=HTTPStatus.BAD_REQUEST,
            )
        except ObjectDoesNotExist:
            return Response(
                "An error occurred during the Livedata Query API request.",
                status=HTTPStatus.NOT_FOUND,
            )
        except Exception as error:  # pylint: disable=broad-exception-caught
            logger.error("Unexpected error during Livedata Query API: %s", error)
            return Response(
                "An unexpected error occurred during the Livedata Query API request.",
                status=HTTPStatus.INTERNAL_SERVER_ERROR,
            )

        job = Job.objects.filter(name=PLUGIN_SETTINGS["query_job_name"]).first()
        if job is None:
            return Response(
                f"{PLUGIN_SETTINGS['query_job_name']} not found",
                status=HTTPStatus.NOT_FOUND,  # 404
            )

        job_kwargs = {
            "commands_j2": show_commands_j2_array,
            "device_id": primary_device_info["device"],
            "interface_id": primary_device_info.get("interface"),
            "primary_device_id": primary_device_info["primary_device"],
            "remote_addr": request.META.get("REMOTE_ADDR"),
            "virtual_chassis_id": primary_device_info.get("virtual_chassis"),
            "x_forwarded_for": request.META.get("HTTP_X_FORWARDED_FOR"),
            "call_object_type": data["object_type"],
        }

        try:
            jobres = JobResult.enqueue_job(
                job,
                user=request.user,
                task_queue=PLUGIN_SETTINGS["query_job_task_queue"],
                **job_kwargs,
            )

            return Response(
                content_type="application/json",
                data={"jobresult_id": jobres.id},
                status=HTTPStatus.OK,  # 200
            )
        except RunJobTaskFailed as error:
            logger.error("Failed to run %s: %s", PLUGIN_SETTINGS["query_job_name"], error)

            return Response(
                "An internal error has occurred while running the job.",
                status=HTTPStatus.INTERNAL_SERVER_ERROR,  # 500
            )
get(request, *args, pk=None, **kwargs)

Handle GET request for Livedata Query API.

The get method is used to enqueue the Livedata Query Job.

To access the JobResult object, use the jobresult_id returned in the response and make a GET request to the JobResult endpoint.

For Example

GET /api/extras/job-results/{jobresult_id}/

Parameters:

Name Type Description Default
request Request

The request object.

 required
pk uuid

The primary key of the model instance.

 None
*args Any

Additional positional arguments.

 ()
**kwargs Any

Additional keyword arguments.

 {}

Returns:

Name Type Description
jobresult_id Response

The job result ID of the job that was enqueued.

Raises:

Type Description
Response

If the user does not have permission to execute 'livedata' on the instance.
Response

If the serializer is not valid.
Response

If the job Livedata Api-Job is not found.
Response

If the job failed to run.
Source code in nautobot_app_livedata/api/views.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def get(self, request: Any, *args: Any, pk=None, **kwargs: Any) -> Response:  # pylint: disable=R0911
    """Handle GET request for Livedata Query API.

    The get method is used to enqueue the Livedata Query Job.

    To access the JobResult object, use the jobresult_id returned in the response
    and make a GET request to the JobResult endpoint.

    For Example:
        GET /api/extras/job-results/{jobresult_id}/

    Args:
        request (Request): The request object.
        pk (uuid): The primary key of the model instance.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Returns:
        jobresult_id: The job result ID of the job that was enqueued.

    Raises:
        Response: If the user does not have permission to execute 'livedata' on the instance.
        Response: If the serializer is not valid.
        Response: If the job Livedata Api-Job is not found.
        Response: If the job failed to run.
    """
    logger.debug(
        "Resolved view=%s queryset_model=%s",
        self.__class__.__name__,
        getattr(getattr(self, "queryset", None), "model", None),
    )
    data = request.data
    data["pk"] = pk
    data["object_type"] = self.get_object_type()
    serializer = self.get_serializer(data=data)
    if not serializer.is_valid():
        return Response(
            serializer.errors,
            status=HTTPStatus.BAD_REQUEST,
        )
    # Check if user has permission to interact with the interface
    if not request.user.has_perm("dcim.can_interact_device"):
        return Response(
            {
                "error": (
                    "You do not have the permission 'can_interact' for 'dcim.device'. Contact your administrator."
                )
            },
            status=HTTPStatus.FORBIDDEN,  # 403
        )

    try:
        primary_device_info = serializer.validated_data
        if not primary_device_info:
            return Response(
                {"error": "Primary device information is missing."},
                status=HTTPStatus.BAD_REQUEST,
            )
        if data["object_type"] == "dcim.interface":
            self.queryset = Interface.objects.all()
            instance = Interface.objects.get(pk=pk)
        elif data["object_type"] == "dcim.device":
            self.queryset = Device.objects.all()
            instance = Device.objects.get(pk=primary_device_info["primary_device"])
        else:
            qs = self.get_queryset()
            instance = qs.get(pk=pk)
        show_commands_j2_array = self.get_commands(instance)
    except ValueError as error:
        logger.error("Error during Livedata Query API: %s", error)
        return Response(
            "An error occurred during the Livedata Query API request.",
            status=HTTPStatus.BAD_REQUEST,
        )
    except ObjectDoesNotExist:
        return Response(
            "An error occurred during the Livedata Query API request.",
            status=HTTPStatus.NOT_FOUND,
        )
    except Exception as error:  # pylint: disable=broad-exception-caught
        logger.error("Unexpected error during Livedata Query API: %s", error)
        return Response(
            "An unexpected error occurred during the Livedata Query API request.",
            status=HTTPStatus.INTERNAL_SERVER_ERROR,
        )

    job = Job.objects.filter(name=PLUGIN_SETTINGS["query_job_name"]).first()
    if job is None:
        return Response(
            f"{PLUGIN_SETTINGS['query_job_name']} not found",
            status=HTTPStatus.NOT_FOUND,  # 404
        )

    job_kwargs = {
        "commands_j2": show_commands_j2_array,
        "device_id": primary_device_info["device"],
        "interface_id": primary_device_info.get("interface"),
        "primary_device_id": primary_device_info["primary_device"],
        "remote_addr": request.META.get("REMOTE_ADDR"),
        "virtual_chassis_id": primary_device_info.get("virtual_chassis"),
        "x_forwarded_for": request.META.get("HTTP_X_FORWARDED_FOR"),
        "call_object_type": data["object_type"],
    }

    try:
        jobres = JobResult.enqueue_job(
            job,
            user=request.user,
            task_queue=PLUGIN_SETTINGS["query_job_task_queue"],
            **job_kwargs,
        )

        return Response(
            content_type="application/json",
            data={"jobresult_id": jobres.id},
            status=HTTPStatus.OK,  # 200
        )
    except RunJobTaskFailed as error:
        logger.error("Failed to run %s: %s", PLUGIN_SETTINGS["query_job_name"], error)

        return Response(
            "An internal error has occurred while running the job.",
            status=HTTPStatus.INTERNAL_SERVER_ERROR,  # 500
        )
get_commands(instance) abstractmethod

Get the commands to be executed for the given instance.

Parameters:

Name Type Description Default
instance Model

The model instance.

 required

Returns:

Name Type Description
list list

The commands to be executed.
Source code in nautobot_app_livedata/api/views.py 
62
63
64
65
66
67
68
69
70
71
@abstractmethod
def get_commands(self, instance) -> list:
    """Get the commands to be executed for the given instance.

    Args:
        instance (Model): The model instance.

    Returns:
        list: The commands to be executed.
    """
get_object_type() abstractmethod

Get the object type for the view.

Returns:

Name Type Description
str str

The object type.
Source code in nautobot_app_livedata/api/views.py 
54
55
56
57
58
59
60
@abstractmethod
def get_object_type(self) -> str:
    """Get the object type for the view.

    Returns:
        str: The object type.
    """

LivedataQueryDeviceApiView

Bases: LivedataQueryApiView

Livedata Query Device API view.

Source code in nautobot_app_livedata/api/views.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class LivedataQueryDeviceApiView(LivedataQueryApiView):
    """Livedata Query Device API view."""

    serializer_class = LivedataSerializer
    queryset = Device.objects.all()

    def get_queryset(self):
        # Restrict devices to those the user can interact with
        qs = super().get_queryset()
        return qs.restrict(self.request.user, action="can_interact_device")

    def get_object_type(self) -> str:
        return "dcim.device"

    def get_commands(self, instance) -> list:
        return get_livedata_commands_for_device(instance)

LivedataQueryInterfaceApiView

Bases: LivedataQueryApiView

Livedata Query Interface API view.

Source code in nautobot_app_livedata/api/views.py 
199
200
201
202
203
204
205
206
207
208
209
class LivedataQueryInterfaceApiView(LivedataQueryApiView):
    """Livedata Query Interface API view."""

    serializer_class = LivedataSerializer
    queryset = Interface.objects.all()

    def get_object_type(self) -> str:
        return "dcim.interface"

    def get_commands(self, instance) -> list:
        return get_livedata_commands_for_interface(instance)