NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-29 11:56:25 -06:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Misc cleanup

Browse Source
This commit is contained in:
Jeremy Stretch 2024-11-26 09:27:04 -05:00
parent 1982741e5b
commit a0fceefb68
2 changed files with 29 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
netbox/core/api/schema.py
View File

@ -158,11 +158,9 @@ class NetBoxAutoSchema(AutoSchema):
fields = {} if hasattr(serializer, 'child') else serializer.fields fields = {} if hasattr(serializer, 'child') else serializer.fields
remove_fields = [] remove_fields = []
""" # If you get a failure here for "AttributeError: 'cached_property' object has no attribute 'items'"
If you get a failure here for: AttributeError: 'cached_property' object has no attribute 'items' # it is probably because you are using a viewsets.ViewSet for the API View and are defining a
it is probably because you are using a viewsets.ViewSet for the API View and are defining a # serializer_class. You will also need to define a get_serializer() method like for GenericAPIView.
serializer_class. You will also need to define a get_serializer function like GenericAPIView.
"""
for child_name, child in fields.items(): for child_name, child in fields.items():
# read_only fields don't need to be in writable (write only) serializers # read_only fields don't need to be in writable (write only) serializers
if 'read_only' in dir(child) and child.read_only: if 'read_only' in dir(child) and child.read_only:

40
netbox/core/utils.py
View File

@ -14,6 +14,28 @@ from rq.registry import (
StartedJobRegistry, StartedJobRegistry,
) )
__all__ = (
'delete_rq_job',
'enqueue_rq_job',
'get_rq_jobs',
'get_rq_jobs_from_status',
'requeue_rq_job',
'stop_rq_job',
)
def get_rq_jobs():
"""
Return a list of all RQ jobs.
"""
jobs = set()
for queue in QUEUES_LIST:
queue = get_queue(queue['name'])
jobs.update(queue.get_jobs())
return list(jobs)
def get_rq_jobs_from_status(queue, status): def get_rq_jobs_from_status(queue, status):
""" """
@ -51,19 +73,9 @@ def get_rq_jobs_from_status(queue, status):
return jobs return jobs
def get_rq_jobs():
jobs = set()
for queue in QUEUES_LIST:
queue = get_queue(queue['name'])
jobs.update(queue.get_jobs())
return list(jobs)
def delete_rq_job(job_id): def delete_rq_job(job_id):
""" """
Deletes the specified RQ job. Delete the specified RQ job.
""" """
config = QUEUES_LIST[0] config = QUEUES_LIST[0]
try: try:
@ -81,7 +93,7 @@ def delete_rq_job(job_id):
def requeue_rq_job(job_id): def requeue_rq_job(job_id):
""" """
Requeues the specified RQ job. Requeue the specified RQ job.
""" """
config = QUEUES_LIST[0] config = QUEUES_LIST[0]
try: try:
@ -97,7 +109,7 @@ def requeue_rq_job(job_id):
def enqueue_rq_job(job_id): def enqueue_rq_job(job_id):
""" """
Enqueues the specified RQ job. Enqueue the specified RQ job.
""" """
config = QUEUES_LIST[0] config = QUEUES_LIST[0]
try: try:
@ -129,7 +141,7 @@ def enqueue_rq_job(job_id):
def stop_rq_job(job_id): def stop_rq_job(job_id):
""" """
Stops the specified RQ job. Stop the specified RQ job.
""" """
config = QUEUES_LIST[0] config = QUEUES_LIST[0]
try: try: