Python rest_framework.response 模块，Response() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用rest_framework.response.Response()。
def employee_location_list(request):
"""
Returns employee location full list
---
serializer: employees.serializers.EmployeeLocationListSerializer
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
"""
if request.method == 'GET':
location_list = get_list_or_404(Location)
serializer = EmployeeLocationListSerializer(location_list, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def employee_role_list(request):
"""
Returns employee role full list
---
serializer: employees.serializers.EmployeeRoleListSerializer
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
"""
if request.method == 'GET':
role_list = get_list_or_404(Role)
serializer = EmployeeRoleListSerializer(role_list, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def get(self, request, format=None):
"""
List all badges
---
serializer: administrator.serializers.BadgeSerializer
parameters:
- name: pagination
required: false
type: string
paramType: query
"""
badges = get_list_or_404(Badge)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(badges, request)
serializer = BadgeSerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = BadgeSerializer(badges, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def SwaggerRequestMethodMaker(model = None):
def model_handler(self, request, *args, **kwargs):
data = kwargs.get('data', None)
resp = model
return Response(resp, status = status.HTTP_200_OK)
def empty_handler(self, request, *args, **kwargs):
data = kwargs.get('data', None)
resp = None
return Response(resp, status = status.HTTP_200_OK)
if model:
return model_handler
else:
return empty_handler
# make named APIView class with specified methods
def get(self, request, **kwargs):
name = kwargs['name']
try:
project = models.Project.objects.get(name=name)
s = self.serializer_class(project)
return Response(s.data)
except:
pass
qs = models.Project.objects.filter(name__icontains=name)
if qs.count() > 0:
s = self.serializer_class(qs, many=True)
return Response(s.data)
return Response(status=404)
def post(self, request): # noqa
username = request.data.get('username')
password = request.data.get('password')
device_id = request.data.get('device_id') or ''
if not username or not password:
return Response(
{'error': 'Missing username or password'},
status=status.HTTP_400_BAD_REQUEST
)
user = authenticate(
username=username.lower(), password=password
)
if not user:
raise InvalidEmailOrPasswordAPIException()
auth_token, _ = AuthToken.objects.get_or_create(
user=user, device_id=device_id
)
return Response({'token': auth_token.key})
def time_entry_item(request, id):
try:
item = TimeEntry.objects.get(user=request.user, id=id)
except NoCurrentEntry:
return Response(status=status.HTTP_404_NOT_FOUND)
user = request.user
if request.method == 'GET':
serializer = TimeEntrySerializer(request.user, item)
return Response(serializer.data)
elif request.method == 'PUT':
serializer = TimeEntrySerializer(request.user, item, data=request.data)
if (serializer.is_valid()):
serializer.save()
return Response(serializer.data)
return Response(serializer.errors)
elif request.method == 'DELETE':
pass
def view_autocomplete(self, request, group, **kwargs):
field = request.GET.get('autocomplete_field')
query = request.GET.get('autocomplete_query')
if field != 'issue_id' or not query:
return Response({'issue_id': []})
repo = self.get_option('repo', group.project)
client = self.get_client(request.user)
try:
response = client.search_issues(
query=(u'repo:%s %s' % (repo, query)).encode('utf-8'),
)
except Exception as e:
return self.handle_api_error(e)
issues = [
{
'text': '(#%s) %s' % (i['number'], i['title']),
'id': i['number']
} for i in response.get('items', [])
]
return Response({field: issues})
def view_autocomplete(self, request, group, **kwargs):
field = request.GET.get('autocomplete_field')
query = request.GET.get('autocomplete_query')
if field != 'issue_id' or not query:
return Response({'issue_id': []})
query = query.encode('utf-8')
_url = '%s?%s' % (self.build_api_url(group, 'search'), urlencode({'query': query}))
try:
req = self.make_api_request(group.project, _url)
body = safe_urlread(req)
except (requests.RequestException, PluginError) as e:
return self.handle_api_error(e)
try:
json_resp = json.loads(body)
except ValueError as e:
return self.handle_api_error(e)
resp = json_resp.get('stories', {})
stories = resp.get('stories', [])
issues = [{'text': '(#%s) %s' % (i['id'], i['name']), 'id': i['id']} for i in stories]
return Response({field: issues})
def put(self, request, id, format=None):
"""Summary
Args:
request (TYPE): Description
id (TYPE): Description
format (None, optional): Description
Returns:
TYPE: Description
"""
process = self.get_object(id)
serializer = serializers.ProcessSerializer(process, data=request.data)
if serializer.is_valid():
serializer.save()
return Response(serializer.data)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
def delete(self, request, id, format=None):
"""Summary
Args:
request (TYPE): Description
id (TYPE): Description
format (None, optional): Description
Returns:
TYPE: Description
"""
process = self.get_object(id)
process.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
# FBV
# class MyOpenAPIRenderer(OpenAPIRenderer):
# def get_customizations(self):
# data = super(MyOpenAPIRenderer, self).get_customizations()
# data['paths'] = custom_data['paths']
# data['info'] = custom_data['info']
# data['basePath'] = custom_data['basePath']
# return data
def upload_avatar(request):
"""
Sets new user's avatar.
"""
with transaction.atomic():
profile_user = get_object_or_404(TheUser, auth_token=request.data.get('user_token'))
try:
profile_user.user_photo.save('user_{}.png'.format(profile_user.id), request.data.get('file'))
profile_user.save()
logger.info("User '{}' changed his avatar.".format(profile_user))
resize_image(profile_user.user_photo.path, AVATAR_WIDTH)
logger.info("Image '{}' successfully resized!".format(profile_user.user_photo.path))
return Response({'status': 200,
'detail': 'successful',
'data': {'profile_image': profile_user.user_photo.url}})
except ValidationError:
logger.info("User '{}' tried to upload not an image as avatar!".format(profile_user))
return Response({'status': 404,
'detail': 'tried to upload not an image',
'data': {}})
def user_login(request):
"""
Authenticates user and returns the token which uses to access to the API.
"""
user = authenticate(username=request.data.get('username'),
password=request.data.get('password'))
if user:
user_token = TheUser.objects.get(id_user=user).auth_token
login(request, user)
logger.info("User '{}' logged in.".format(user.username))
return Response({'status': 200,
'detail': 'successful',
'data': {'token': user_token}})
return Response({'status': 404,
'detail': 'not authenticated',
'data': {'token': None}})
# ----------------------------------------------------------------------------------------------------------------------
def is_user_exists(request):
"""
Checks if user is exists. If exists return True, else False.
"""
try:
User.objects.get(username=request.data.get('username'))
return Response({'status': 200,
'detail': 'successful',
'data': {'user': True}})
except ObjectDoesNotExist:
return Response({'status': 200,
'detail': 'successful',
'data': {'user': False}})
# ----------------------------------------------------------------------------------------------------------------------
def is_mail_exists(request):
"""
Checks if mail is exists. If exists return True, else False.
"""
try:
User.objects.get(email=request.data.get('email'))
return Response({'status': 200,
'detail': 'successful',
'data': {'email': True}})
except ObjectDoesNotExist:
return Response({'status': 200,
'detail': 'successful',
'data': {'email': False}})
# ----------------------------------------------------------------------------------------------------------------------
def selected_category(request):
"""
Returns books from selected category.
"""
user = get_object_or_404(TheUser, auth_token=request.data.get('user_token'))
category = get_object_or_404(Category, id=request.data.get('category_id'))
books = Book.objects.filter(id_category=category).order_by('book_name')
filtered_books = Book.exclude_private_books(user.id_user, books)
paginator = Paginator(filtered_books, OUTPUT_BOOKS_PER_PAGE)
page = paginator.page(request.data.get('page'))
next_page = page.has_next()
page_books = page.object_list
return Response({'status': 200,
'detail': 'successful',
'data': {'books': [BookSerializer(book).data for book in page_books],
'next_page': page.next_page_number() if next_page else 0}})
# ----------------------------------------------------------------------------------------------------------------------
def find_book(request):
"""
Generates list with books of data which user entered. At first it check full equality in name,
after tries to check if contains some part of entered data.
"""
user = get_object_or_404(TheUser, auth_token=request.data.get('user_token'))
search_data = request.data.get('search_term')
filtered_books = Book.exclude_private_books(user.id_user, Book.fetch_books(search_data))
paginator = Paginator(filtered_books, OUTPUT_BOOKS_PER_PAGE)
page = paginator.page(request.data.get('page'))
next_page = page.has_next()
page_books = page.object_list
return Response({'status': 200,
'detail': 'successful',
'data': {'books': [BookSerializer(book).data for book in page_books],
'next_page': page.next_page_number() if next_page else 0}})
def add_book_to_home(request):
"""
Adds book to list of user's added books.
"""
user = get_object_or_404(TheUser, auth_token=request.data.get('user_token'))
book = get_object_or_404(Book, id=request.data.get('book_id'))
if book.private_book and book.who_added != user:
return Response({}, status=404)
if AddedBook.objects.filter(id_user=user, id_book=book).exists():
return Response({}, status=404)
AddedBook.objects.create(id_user=user, id_book=book)
logger.info("User '{}' added book with id: '{}' to his own library.".format(user.id_user.id, book.id))
return Response({'status': 200,
'detail': 'success',
'data': {}})
# ----------------------------------------------------------------------------------------------------------------------
def open_book(request):
"""
Returns the book of last readed page.
"""
user = get_object_or_404(TheUser, auth_token=request.data.get('user_token'))
book = get_object_or_404(Book, id=request.data.get('book_id'))
added_book = get_object_or_404(AddedBook, id_book=book, id_user=user)
added_book.last_read = added_book.last_read.now()
added_book.save()
logger.info("User '{}' opened book with id: '{}'.".format(user, book.id))
return Response({'status': 200,
'detail': 'successful',
'data': {'last_page': added_book.last_page}})
# ----------------------------------------------------------------------------------------------------------------------
def set_current_page(request):
"""
Changes current readed page for book of the user.
"""
user = get_object_or_404(TheUser, auth_token=request.data.get('user_token'))
book = get_object_or_404(Book, id=request.data.get('book_id'))
current_page = request.data.get('current_page')
if not isinstance(current_page, int):
return Response({'status': 400,
'detail': 'current page not a number',
'data': {}}, status=400)
added_book = AddedBook.objects.get(id_book=book, id_user=user)
added_book.last_page = current_page
added_book.save()
logger.info("User '{}' on book with id: '{}' changed page to: '{}'."
.format(user, book.id, current_page))
return Response({'status': 200,
'detail': 'successful',
'data': {}})
def save_support_message(request):
"""
Saves a support message if validated.
"""
message = request.data.get('text')
if len(message) > 5000:
return Response({'status': 400,
'detail': 'message length more than 5000 symbols',
'data': {}})
SupportMessage.objects.create(email=request.data.get('email'), text=message)
return Response({'status': 200,
'detail': 'successful',
'data': {}})
def category_detail(request, category_id):
"""
Category detail
---
serializer: categories.serializers.CategorySerializer
responseMessages:
- code: 400
message: Bad request.
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
"""
category = get_object_or_404(Category, pk=category_id)
if request.method == 'GET':
serializer = CategorySerializer(category)
return Response(serializer.data, status=status.HTTP_200_OK)
def keyword_create(request):
"""
Creates a new keyword
---
serializer: categories.serializers.KeywordSerializer
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
"""
if request.method == 'POST':
serializer = KeywordSerializer(data=request.data)
if serializer.is_valid():
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
def keyword_detail(request, keyword_id):
"""
Keyword detail
---
serializer: categories.serializers.KeywordSerializer
responseMessages:
- code: 400
message: Bad request.
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
"""
keyword = get_object_or_404(Keyword, pk=keyword_id)
if request.method == 'GET':
serializer = KeywordSerializer(keyword)
return Response(serializer.data, status=status.HTTP_200_OK)
def employee_activate(request, employee_id, action):
"""
Activate employee account, action could be true or false
---
response_serializer: employees.serializers.EmployeeSerializer
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
"""
if request.method == 'PATCH':
employee = get_object_or_404(Employee, pk=employee_id)
if action == 'true':
employee.is_active = True
elif action == 'false':
employee.is_active = False
else:
pass
employee.save()
serializer = EmployeeSerializer(employee)
return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
def employee_block(request, employee_id, action):
"""
Block employee account, action could be true or false
---
response_serializer: employees.serializers.EmployeeSerializer
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
"""
if request.method == 'PATCH':
employee = get_object_or_404(Employee, pk=employee_id)
if action == 'true':
employee.is_blocked = True
elif action == 'false':
employee.is_blocked = False
else:
pass
employee.save()
serializer = EmployeeSerializer(employee)
return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
def employee_logout(request):
"""
Logout employee
---
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
- code: 500
message: Internal Server Error
"""
if request.method == 'GET':
employee = request.user
try:
devices = EmployeeDevice.objects.filter(username=employee)
for device in devices:
device.delete()
except:
pass
logout(request)
content = {'detail': config.USER_LOGOUT}
return Response(content, status=status.HTTP_202_ACCEPTED)
def employee_admin(request, employee_id, action):
"""
Set or unset admin permission to employee, action could be true or false
---
response_serializer: employees.serializers.EmployeeSerializer
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden.
- code: 404
message: Not found
"""
if request.method == 'PATCH':
employee = get_object_or_404(Employee, pk=employee_id)
if action == 'true':
employee.is_staff = True
elif action == 'false':
employee.is_staff = False
else:
pass
employee.save()
serializer = EmployeeSerializer(employee)
return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
def star(request, star_id):
"""
Returns star detail
---
serializer: stars.serializers.StarSerializer
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden, authentication credentials were not provided
- code: 404
message: Not found
"""
if request.method == 'GET':
star = get_object_or_404(Star, pk=star_id)
serializer = StarSerializer(star)
return Response(serializer.data, status=status.HTTP_200_OK)
def get(self, request, format=None):
"""
List all categories
---
parameters:
- name: pagination
required: false
type: string
paramType: query
"""
categories = get_list_or_404(Category)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(categories, request)
serializer = CategorySerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = CategorySerializer(categories, many=True)
return Response(serializer.data, status=status.HTTP_400_BAD_REQUEST)
def get(self, request, format=None):
"""
List all events
---
serializer: administrator.serializers.EventSerializer
parameters:
- name: pagination
required: false
type: string
paramType: query
"""
events = get_list_or_404(Event)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(events, request)
serializer = EventSerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = EventSerializer(events, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def get(self, request, event_id, format=None):
"""
List all activities for an event
---
serializer: administrator.serializers.EventActivitySerializer
parameters:
- name: pagination
required: false
type: string
paramType: query
"""
event = get_object_or_404(Event, pk=event_id)
activities = EventActivity.objects.filter(event=event)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(activities, request)
serializer = EventActivitySerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = EventActivitySerializer(activities, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def get(self, request, format=None):
"""
List all employee positions
---
serializer: administrator.serializers.LocationSerializer
parameters:
- name: pagination
required: false
type: string
paramType: query
"""
locations = get_list_or_404(Location)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(locations, request)
serializer = LocationSerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = LocationSerializer(locations, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def get(self, request, format=None):
"""
List all messages
---
serializer: administrator.serializers.MessageSerializer
parameters:
- name: pagination
required: false
type: string
paramType: query
"""
messages = get_list_or_404(Message)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(messages, request)
serializer = MessageSerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = MessageSerializer(messages, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def get(self, request, employee_id, format=None):
"""
List all messages from employee
---
serializer: administrator.serializers.MessageSerializer
parameters:
- name: pagination
required: false
type: string
paramType: query
"""
employee = get_object_or_404(Employee, pk=employee_id)
messages = get_list_or_404(Message, from_user=employee)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(messages, request)
serializer = MessageSerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = MessageSerializer(messages, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def get(self, request, format=None):
"""
List all roles
---
serializer: administrator.serializers.RoleSerializer
parameters:
- name: pagination
required: false
type: string
paramType: query
"""
roles = get_list_or_404(Role)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(roles, request)
serializer = RoleSerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = RoleSerializer(roles, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def delete(self, request, id, kind, format=None):
"""
WARNING: Force delete
"""
if kind == 'badge':
kind = get_object_or_404(Badge, pk=id)
elif kind == 'category':
kind = get_object_or_404(Category, pk=id)
elif kind == 'event':
kind = get_object_or_404(Event, pk=id)
elif kind == 'keyword':
kind = get_object_or_404(Keyword, pk=id)
elif kind == 'location':
kind = get_object_or_404(Location, pk=id)
elif kind == 'position':
kind = get_object_or_404(Position, pk=id)
elif kind == 'role':
kind = get_object_or_404(Role, pk=id)
else:
return Response(status=status.HTTP_404_NOT_FOUND)
kind.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
def get(self, request, format=None):
"""
Get site info
---
serializer: administrator.serializers.SiteInfoSerializer
"""
email_domain = settings.EMAIL_DOMAIN_LIST[0]
current_site = Site.objects.get_current()
version = config.VERSION
data = {'site': current_site.domain,
'email_domain': email_domain,
'backend_version': version}
serializer = SiteInfoSerializer(data)
return Response(serializer.data, status=status.HTTP_200_OK)
def retrieve(self, request, parent_lookup_torrent=None, *args, **kwargs):
"""
Retrieve a file of the user.
:return: Response
"""
if parent_lookup_torrent:
# Retrieve a file of the torrent.
torrent = get_object_or_404(Torrent, pk=parent_lookup_torrent)
if not torrent.finished:
return Response({
'detail': "The torrent hasn't finished downloading yet.",
'progress': torrent.progress
}, status=status.HTTP_400_BAD_REQUEST)
file_obj = get_object_or_404(File, torrent__pk=parent_lookup_torrent, pk=kwargs.get('pk'))
serializer = self.serializer_class(file_obj)
return Response(serializer.data)
return super(FileViewSet, self).retrieve(request, *args, **kwargs)
def get_paginated_response(self, data):
"""
Return response with some additional fields.
:param data: data queryset.
:type data: django.db.models.query.QuerySet.
:return: extended django rest framework response.
:rtype: rest_framework.response.Response.
"""
return Response(OrderedDict([
("page", self.page.number),
("num_pages", self.page.paginator.num_pages),
("count", self.page.paginator.count),
("next", self.get_next_link()),
("next_page_number", self.page.next_page_number() if self.page.has_next() else None),
("previous", self.get_previous_link()),
("previous_page_number", self.page.previous_page_number() if self.page.has_previous() else None),
("request", self.request.GET),
("results", data),
]))
def my(self, request, **kwargs):
"""
Return only user RSVP.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=request.user.rsvp.all() if request.user.is_authenticated else RSVP.objects.none())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def my(self, request, **kwargs):
"""
Return only user owned groups.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=request.user.owned.all() if request.user.is_authenticated else Group.objects.none())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def active(self, request, **kwargs):
"""
Return only active groups.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=Group.objects.active())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def inactive(self, request, **kwargs):
"""
Return only inactive groups.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=Group.objects.inactive())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def my__inactive(self, request, **kwargs):
"""
Return only inactive user owned groups.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=request.user.owned.inactive() if request.user.is_authenticated else Group.objects.none())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def active(self, request, **kwargs):
"""
Return only active memberships.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=Membership.objects.active())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def inactive(self, request, **kwargs):
"""
Return only inactive memberships.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=Membership.objects.inactive())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def my__active(self, request, **kwargs):
"""
Return only active user memberships.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=request.user.membership.active() if request.user.is_authenticated else Membership.objects.none())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def my__inactive(self, request, **kwargs):
"""
Return only inactive user memberships.
:param request: django request instance.
:type request: django.http.request.HttpRequest.
:param kwargs: additional args.
:type kwargs: dict.
:return: serialized custom queryset response.
:rtype: rest_framework.response.Response.
"""
queryset = self.filter_queryset(queryset=request.user.membership.inactive() if request.user.is_authenticated else Membership.objects.none())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def getNewsByKey(request):
"""
?????????id????????????
:param request:
{
"_id": "??",
"start": 0,
"size": 6
}
:return:
"""
size = 6
start = 0
keyword = ""
if request.method == 'GET':
size = int(request.GET.get('size', 6))
start = int(request.GET.get('start', 0))
keyword = request.GET.get('_id', 0)
if request.method == 'POST':
size = int(request.data.get('size', 6))
start = int(request.data.get('start', 0)) * size
keyword = request.data.get('_id', "")
resultList = list(newsCol.find({"keywords": keyword}))
return Response(resultList[start : start + size])