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.

API endpoint for retrieving the primary device for a given object (device, interface, or virtual chassis). The primary device is the device with an active status and primary IP address that can be used to establish a connection for executing commands.

Returns information about the object hierarchy including device, interface, virtual chassis, and the determined primary device.

Source code in nautobot_app_livedata/api/views.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
class LivedataPrimaryDeviceApiView(GenericAPIView):
    """Nautobot App Livedata API Primary Device view.

    API endpoint for retrieving the primary device for a given object (device, interface, or virtual chassis).
    The primary device is the device with an active status and primary IP address that can be used
    to establish a connection for executing commands.

    Returns information about the object hierarchy including device, interface, virtual chassis,
    and the determined primary device.
    """

    serializer_class = LivedataSerializer
    queryset = Device.objects.all()
    permission_classes = []  # No DRF permission checks - this view doesn't require authentication

    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.
        """
        # Check if user has permission to interact with devices
        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
            )

        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 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
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.
    """
    # Check if user has permission to interact with devices
    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
        )

    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.

Base class for all Livedata Query API views. Provides common functionality for queuing livedata jobs to execute commands on network devices. Subclasses must implement get_object_type() and get_commands() methods.

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
197
198
199
200
201
202
203
204
205
206
207
208
class LivedataQueryApiView(GenericAPIView, ABC):
    """Abstract Livedata Query API view.

    Base class for all Livedata Query API views. Provides common functionality
    for queuing livedata jobs to execute commands on network devices.
    Subclasses must implement get_object_type() and get_commands() methods.
    """

    serializer_class = LivedataSerializer
    permission_classes = []  # Custom permission checking in get() method

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

        Must be implemented by subclasses to identify the type of object
        being queried (e.g., 'dcim.interface' or 'dcim.device').

        Returns:
            str: The object type in format 'app_label.model_name'.
        """

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

        Must be implemented by subclasses to retrieve the appropriate commands
        based on the instance type and its platform configuration.

        Args:
            instance (Model): The model instance (Device or Interface) to query.

        Returns:
            list[str]: List of command strings to be executed on the device.
        """

    def get(self, request: Any, *args: Any, pk: Optional[Any] = 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 
 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
197
198
199
200
201
202
203
204
205
206
207
208
def get(self, request: Any, *args: Any, pk: Optional[Any] = 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.

Must be implemented by subclasses to retrieve the appropriate commands based on the instance type and its platform configuration.

Parameters:

Name Type Description Default
instance Model

The model instance (Device or Interface) to query.

 required

Returns:

Type Description
list[str]

list[str]: List of command strings to be executed on the device.
Source code in nautobot_app_livedata/api/views.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
@abstractmethod
def get_commands(self, instance: Any) -> list[str]:
    """Get the commands to be executed for the given instance.

    Must be implemented by subclasses to retrieve the appropriate commands
    based on the instance type and its platform configuration.

    Args:
        instance (Model): The model instance (Device or Interface) to query.

    Returns:
        list[str]: List of command strings to be executed on the device.
    """
get_object_type() abstractmethod

Get the object type for the view.

Must be implemented by subclasses to identify the type of object being queried (e.g., 'dcim.interface' or 'dcim.device').

Returns:

Name Type Description
str str

The object type in format 'app_label.model_name'.
Source code in nautobot_app_livedata/api/views.py 
60
61
62
63
64
65
66
67
68
69
@abstractmethod
def get_object_type(self) -> str:
    """Get the object type for the view.

    Must be implemented by subclasses to identify the type of object
    being queried (e.g., 'dcim.interface' or 'dcim.device').

    Returns:
        str: The object type in format 'app_label.model_name'.
    """

LivedataQueryDeviceApiView

Bases: LivedataQueryApiView

Livedata Query Device API view.

API endpoint for executing live data commands on a specific device. Retrieves commands from the device platform's 'livedata_device_commands' custom field. Restricts access to devices the user has 'can_interact_device' permission for.

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
class LivedataQueryDeviceApiView(LivedataQueryApiView):
    """Livedata Query Device API view.

    API endpoint for executing live data commands on a specific device.
    Retrieves commands from the device platform's 'livedata_device_commands' custom field.
    Restricts access to devices the user has 'can_interact_device' permission for.
    """

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

    def get_queryset(self) -> Any:
        """Filter queryset to devices the user can interact with.

        Returns:
            QuerySet: Device queryset restricted by user's can_interact_device permission.
        """
        # 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 the object type for device queries.

        Returns:
            str: Always returns 'dcim.device'.
        """
        return "dcim.device"

    def get_commands(self, instance) -> list:
        """Get livedata commands for the device.

        Args:
            instance (Device): The device instance to query.

        Returns:
            list[str]: List of commands from the platform's livedata_device_commands field.
        """
        return get_livedata_commands_for_device(instance)
get_commands(instance)

Get livedata commands for the device.

Parameters:

Name Type Description Default
instance Device

The device instance to query.

 required

Returns:

Type Description
list

list[str]: List of commands from the platform's livedata_device_commands field.
Source code in nautobot_app_livedata/api/views.py 
270
271
272
273
274
275
276
277
278
279
def get_commands(self, instance) -> list:
    """Get livedata commands for the device.

    Args:
        instance (Device): The device instance to query.

    Returns:
        list[str]: List of commands from the platform's livedata_device_commands field.
    """
    return get_livedata_commands_for_device(instance)
get_object_type()

Return the object type for device queries.

Returns:

Name Type Description
str str

Always returns 'dcim.device'.
Source code in nautobot_app_livedata/api/views.py 
262
263
264
265
266
267
268
def get_object_type(self) -> str:
    """Return the object type for device queries.

    Returns:
        str: Always returns 'dcim.device'.
    """
    return "dcim.device"
get_queryset()

Filter queryset to devices the user can interact with.

Returns:

Name Type Description
QuerySet Any

Device queryset restricted by user's can_interact_device permission.
Source code in nautobot_app_livedata/api/views.py 
252
253
254
255
256
257
258
259
260
def get_queryset(self) -> Any:
    """Filter queryset to devices the user can interact with.

    Returns:
        QuerySet: Device queryset restricted by user's can_interact_device permission.
    """
    # Restrict devices to those the user can interact with
    qs = super().get_queryset()
    return qs.restrict(self.request.user, action="can_interact_device")

LivedataQueryInterfaceApiView

Bases: LivedataQueryApiView

Livedata Query Interface API view.

API endpoint for executing live data commands on a specific interface. Retrieves commands from the device platform's 'livedata_interface_commands' custom field.

Source code in nautobot_app_livedata/api/views.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
class LivedataQueryInterfaceApiView(LivedataQueryApiView):
    """Livedata Query Interface API view.

    API endpoint for executing live data commands on a specific interface.
    Retrieves commands from the device platform's 'livedata_interface_commands' custom field.
    """

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

    def get_object_type(self) -> str:
        """Return the object type for interface queries.

        Returns:
            str: Always returns 'dcim.interface'.
        """
        return "dcim.interface"

    def get_commands(self, instance) -> list:
        """Get livedata commands for the interface.

        Args:
            instance (Interface): The interface instance to query.

        Returns:
            list[str]: List of commands from the platform's livedata_interface_commands field.
        """
        return get_livedata_commands_for_interface(instance)
get_commands(instance)

Get livedata commands for the interface.

Parameters:

Name Type Description Default
instance Interface

The interface instance to query.

 required

Returns:

Type Description
list

list[str]: List of commands from the platform's livedata_interface_commands field.
Source code in nautobot_app_livedata/api/views.py 
229
230
231
232
233
234
235
236
237
238
def get_commands(self, instance) -> list:
    """Get livedata commands for the interface.

    Args:
        instance (Interface): The interface instance to query.

    Returns:
        list[str]: List of commands from the platform's livedata_interface_commands field.
    """
    return get_livedata_commands_for_interface(instance)
get_object_type()

Return the object type for interface queries.

Returns:

Name Type Description
str str

Always returns 'dcim.interface'.
Source code in nautobot_app_livedata/api/views.py 
221
222
223
224
225
226
227
def get_object_type(self) -> str:
    """Return the object type for interface queries.

    Returns:
        str: Always returns 'dcim.interface'.
    """
    return "dcim.interface"