External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-03-13 15:09:32 -02:30
Code Issues Packages Projects Releases Wiki Activity

Remove sneeringer's old unused decorator code

Browse Source
This commit is contained in:
Matthew Jones
2016-12-12 16:13:56 -05:00
parent 4063ef4353
commit e1942083e3
4 changed files with 8 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
awx/api/utils/__init__.py
View File

82
awx/api/utils/decorators.py
View File

@@ -1,82 +0,0 @@
# Copyright (c) 2015 Ansible, Inc.
# All Rights Reserved.
from collections import OrderedDict
import copy
import functools
from rest_framework.response import Response
from rest_framework.settings import api_settings
from rest_framework import status
def paginated(method):
"""Given an method with a Django REST Framework API method signature
(e.g. `def get(self, request, ...):`), abstract out boilerplate pagination
duties.
This causes the method to receive two additional keyword arguments:
`limit`, and `offset`. The method expects a two-tuple to be
returned, with a result list as the first item, and the total number
of results (across all pages) as the second item.
"""
@functools.wraps(method)
def func(self, request, *args, **kwargs):
# Manually spin up pagination.
# How many results do we show?
paginator_class = api_settings.DEFAULT_PAGINATION_CLASS
limit = paginator_class.page_size
if request.query_params.get(paginator_class.page_size_query_param, False):
limit = request.query_params[paginator_class.page_size_query_param]
if paginator_class.max_page_size:
limit = min(paginator_class.max_page_size, limit)
limit = int(limit)
# Get the order parameter if it's given
if request.query_params.get("ordering", False):
ordering = request.query_params["ordering"]
else:
ordering = None
# What page are we on?
page = int(request.query_params.get('page', 1))
offset = (page - 1) * limit
# Add the limit, offset, page, and order variables to the keyword arguments
# being sent to the underlying method.
kwargs['limit'] = limit
kwargs['offset'] = offset
kwargs['ordering'] = ordering
# Okay, call the underlying method.
results, count, stat = method(self, request, *args, **kwargs)
if stat is None:
stat = status.HTTP_200_OK
if stat == status.HTTP_200_OK:
# Determine the next and previous pages, if any.
prev, next_ = None, None
if page > 1:
get_copy = copy.copy(request.GET)
get_copy['page'] = page - 1
prev = '%s?%s' % (request.path, get_copy.urlencode())
if count > offset + limit:
get_copy = copy.copy(request.GET)
get_copy['page'] = page + 1
next_ = '%s?%s' % (request.path, get_copy.urlencode())
# Compile the results into a dictionary with pagination
# information.
answer = OrderedDict((
('count', count),
('next', next_),
('previous', prev),
('results', results),
))
else:
answer = results
# Okay, we're done; return response data.
return Response(answer, status=stat)
return func

14
awx/api/views.py
View File

@@ -60,7 +60,6 @@ from awx.main.tasks import send_notifications
from awx.main.access import get_user_queryset
from awx.main.ha import is_ha_environment
from awx.api.authentication import TaskAuthentication, TokenGetAuthentication
from awx.api.utils.decorators import paginated
from awx.api.generics import get_view_name
from awx.api.generics import * # noqa
from awx.conf.license import get_license, feature_enabled, feature_exists, LicenseForbids
@@ -3435,8 +3434,10 @@ class JobJobPlaysList(BaseJobEventsList):
view_name = _('Job Plays List')
new_in_200 = True
@paginated
def get(self, request, limit, offset, ordering, *args, **kwargs):
def get(self, request, *args, **kwargs):
limit = kwargs.get('limit', 20)
ordering = kwargs.get('ordering', None)
offset = kwargs.get('offset', 0)
all_plays = []
job = Job.objects.filter(pk=self.kwargs['pk'])
if not job.exists():
@@ -3510,14 +3511,15 @@ class JobJobTasksList(BaseJobEventsList):
view_name = _('Job Play Tasks List')
new_in_200 = True
@paginated
def get(self, request, limit, offset, ordering, *args, **kwargs):
def get(self, request, *args, **kwargs):
"""Return aggregate data about each of the job tasks that is:
- an immediate child of the job event
- corresponding to the spinning up of a new task or playbook
"""
results = []
limit = kwargs.get('limit', 20)
ordering = kwargs.get('ordering', None)
offset = kwargs.get('offset', 0)
# Get the job and the parent task.
# If there's no event ID specified, this will return a 404.
job = Job.objects.filter(pk=self.kwargs['pk'])

83
awx/main/tests/unit/api/decorator_paginated.py
View File

@@ -1,83 +0,0 @@
# Copyright (c) 2015 Ansible, Inc.
# All Rights Reserved.
import json
from django.test import TestCase
from rest_framework.permissions import AllowAny
from rest_framework.test import APIRequestFactory
from rest_framework.views import APIView
from awx.api.utils.decorators import paginated
class PaginatedDecoratorTests(TestCase):
"""A set of tests for ensuring that the "paginated" decorator works
in the way we expect.
"""
def setUp(self):
self.rf = APIRequestFactory()
# Define an uninteresting view that we can use to test
# that the paginator wraps in the way we expect.
class View(APIView):
permission_classes = (AllowAny,)
@paginated
def get(self, request, limit, ordering, offset):
return ['a', 'b', 'c', 'd', 'e'], 26, None
self.view = View.as_view()
def test_implicit_first_page(self):
"""Establish that if we get an implicit request for the first page
(e.g. no page provided), that it is returned appropriately.
"""
# Create a request, and run the paginated function.
request = self.rf.get('/dummy/', {'page_size': 5})
response = self.view(request)
# Ensure the response looks like what it should.
r = json.loads(response.rendered_content)
self.assertEqual(r['count'], 26)
self.assertIn(r['next'],
(u'/dummy/?page=2&page_size=5',
u'/dummy/?page_size=5&page=2'))
self.assertEqual(r['previous'], None)
self.assertEqual(r['results'], ['a', 'b', 'c', 'd', 'e'])
def test_mid_page(self):
"""Establish that if we get a request for a page in the middle, that
the paginator causes next and prev to be set appropriately.
"""
# Create a request, and run the paginated function.
request = self.rf.get('/dummy/', {'page': 3, 'page_size': 5})
response = self.view(request)
# Ensure the response looks like what it should.
r = json.loads(response.rendered_content)
self.assertEqual(r['count'], 26)
self.assertIn(r['next'],
(u'/dummy/?page=4&page_size=5',
u'/dummy/?page_size=5&page=4'))
self.assertIn(r['previous'],
(u'/dummy/?page=2&page_size=5',
u'/dummy/?page_size=5&page=2'))
self.assertEqual(r['results'], ['a', 'b', 'c', 'd', 'e'])
def test_last_page(self):
"""Establish that if we get a request for the last page, that the
paginator picks up on it and sets `next` to None.
"""
# Create a request, and run the paginated function.
request = self.rf.get('/dummy/', {'page': 6, 'page_size': 5})
response = self.view(request)
# Ensure the response looks like what it should.
r = json.loads(response.rendered_content)
self.assertEqual(r['count'], 26)
self.assertEqual(r['next'], None)
self.assertIn(r['previous'],
(u'/dummy/?page=5&page_size=5',
u'/dummy/?page_size=5&page=5'))
self.assertEqual(r['results'], ['a', 'b', 'c', 'd', 'e'])