Python库 | django-rest-framework-simplify-1.1.1.dev.tar.gz

import datetime import dateutil.parser import logging from decimal import Decimal from django.core.cache import cache from django.core.exceptions import ObjectDoesNotExist from django.db.models.fields.related import ForeignKey, OneToOneField from rest_framework import exceptions from rest_framework.response import Response from rest_framework.views import APIView from rest_framework import status from mongoengine.errors import DoesNotExist from rest_framework_simplify.exceptions import ParseException from rest_framework_simplify.mapper import Mapper from rest_framework_simplify.serializer import MongoEngineSerializer, SQLEngineSerializer from rest_framework_simplify.services.sql_executor.service import SQLExecutorService from rest_framework_simplify.errors import ErrorMessages class SimplifyView(APIView): def __init__(self, model, linked_objects=[], supported_methods=[], read_db='default', write_db='default'): self.read_db = read_db self.write_db = write_db self.model = model self.supported_methods = supported_methods self.linked_objects = linked_objects if self.get_db_engine() == 'mongo': self.DoesNotExist = DoesNotExist else: self.DoesNotExist = ObjectDoesNotExist if self.get_db_engine() == 'mongo': self.serializer = MongoEngineSerializer else: self.serializer = SQLEngineSerializer def get_db_engine(self): if hasattr(self.model, 'validate'): return 'mongo' else: return 'sql' def delete(self, request, pk=None, parent_resource=None, parent_pk=None): if parent_pk and parent_resource and self.linked_objects: if 'DELETE_SUB' not in self.supported_methods: return self.create_response(error_message=ErrorMessages.DELETE_SUB_NOT_SUPPORTED .format(self.model.__name__)) else: if 'DELETE' not in self.supported_methods: return self.create_response(error_message=ErrorMessages.DELETE_NOT_SUPPORTED.format(self.model.__name__)) try: obj = self.model.objects.using(self.read_db).get(pk=pk) except self.DoesNotExist: raise self.DoesNotExist(ErrorMessages.DOES_NOT_EXIST.format(self.model.__name__, pk)) # check query param to only delete linker delete_link_only = request.query_params.get('deleteLinkOnly', False) # further checks for linked_object in self.linked_objects: # we have a linking table we need to clear the linking table objects if linked_object['sub_resource_name'] and linked_object['linking_cls']: kwargs = { linked_object['sub_resource_name']: obj } # get the linking table items linked_objs = linked_object['linking_cls'].objects.using(self.read_db).filter(**kwargs) for linked_obj in linked_objs: linked_obj.delete(using=self.read_db) else: pass if not delete_link_only: obj.delete(using=self.write_db) return self.create_response() def get(self, request, pk=None, parent_resource=None, parent_pk=None): meta_request = request.query_params.get('meta', False) if meta_request: # grab models fields and return them as a dict meta_data = self.model.get_meta_data() return self.create_response(body=meta_data) # handle caching cache_key = None if hasattr(self.model, 'CACHE'): cache_key = request.get_full_path() result = cache.get(cache_key, None) if result: return self.create_response(body=result, using_cache=True, cache_key=cache_key) obj = None # if we have a primary key we are returning one result if pk: if 'GET' not in self.supported_methods: return self.create_response(error_message=ErrorMessages.GET_NOT_SUPPORTED.format(self.model.__name__)) # try to get the obj from db try: obj = self.model.objects.using(self.read_db).get(pk=pk) except self.DoesNotExist: raise self.DoesNotExist(ErrorMessages.DOES_NOT_EXIST.format(self.model.__name__, pk)) else: # we could be a sub resource so we need to check if a parent_resource was passed in if parent_pk and parent_resource and self.linked_objects: snake_cased_url_tail = Mapper.camelcase_to_underscore(request.get_full_path().split('/')[-1]) lives_on_parent_results = [i for i in self.linked_objects if 'lives_on_parent' in i and i['lives_on_parent'] and i['sub_resource_name'] == snake_cased_url_tail] if len(lives_on_parent_results) > 0: if 'GET_SUB' not in self.supported_methods: return self.create_response(error_message=ErrorMessages.GET_SUB_NOT_SUPPORTED.format(self.model.__name__)) linked_object = lives_on_parent_results[0] parent_obj = linked_object['parent_cls'].objects.using(self.read_db).get(pk=parent_pk) obj = getattr(parent_obj, linked_object['sub_resource_name']) else: # check if this method has authorized sub resources if 'GET_LIST_SUB' not in self.supported_methods: return self.create_response(error_message=ErrorMessages.GET_LIST_SUB_NOT_SUPPORTED.format(self.model.__name__)) # find the resource that this request is looking for for linked_object in self.linked_objects: if linked_object['parent_resource'] == parent_resource: if linked_object['parent_cls']: # get the resource parent_obj = linked_object['parent_cls'].objects.using(self.read_db).get(pk=parent_pk) else: parent_obj = parent_pk # setup kwargs for django's orm to query kwargs = { linked_object['parent_name']: parent_obj } # if there is a linking table do that logic if linked_object['linking_cls']: # get the linking table items linked_objs = linked_object['linking_cls'].objects.using(self.read_db).filter(**kwargs) # go through linking table items and get the sub resources from each entry into a list linked_obj_ids = [getattr(linked_obj, linked_object['sub_resource_name']).id for linked_obj in linked_objs] obj = self.model.objects.using(self.read_db).filter(pk__in=linked_obj_ids) # no linking table and the link is on this obj itself else: obj = self.model.objects.using(self.read_db).filter(**kwargs) else: # trying to get ALL items in DB if 'GET_LIST' not in self.supported_methods: return self.create_response(error_message=ErrorMessages.GET_LIST_NOT_SUPPORTED.format(self.model.__name__)) obj = self.model.objects.using(self.read_db).all() # handle includes req_includes = request.query_params.get('include', []) if req_includes: req_includes = req_includes.split(',') if type(req_includes) is not list: req_includes = [req_includes]