# See the License for the specific language governing permissions and
# limitations under the License.
-
-import traceback
import logging
from drf_yasg.utils import swagger_auto_schema
from rest_framework.views import APIView
from rest_framework.response import Response
-from genericparser.packages.serializers.vnf_pkg_subscription import PkgmSubscriptionRequestSerializer, \
- PkgmSubscriptionSerializer, PkgmSubscriptionsSerializer
+from genericparser.packages.serializers.vnf_pkg_subscription import PkgmSubscriptionRequestSerializer
+from genericparser.packages.serializers.vnf_pkg_subscription import PkgmSubscriptionSerializer
+from genericparser.packages.serializers.vnf_pkg_subscription import PkgmSubscriptionsSerializer
from genericparser.packages.serializers.response import ProblemDetailsSerializer
-from genericparser.packages.biz.vnf_pkg_subscription import CreateSubscription, QuerySubscription, TerminateSubscription
+from genericparser.packages.biz.vnf_pkg_subscription import CreateSubscription
+from genericparser.packages.biz.vnf_pkg_subscription import QuerySubscription
+from genericparser.packages.biz.vnf_pkg_subscription import TerminateSubscription
from genericparser.packages.views.common import validate_data
-from genericparser.pub.exceptions import VnfPkgDuplicateSubscriptionException, VnfPkgSubscriptionException, \
- SubscriptionDoesNotExistsException
+from genericparser.pub.exceptions import VnfPkgSubscriptionException
+from genericparser.pub.exceptions import BadRequestException
-logger = logging.getLogger(__name__)
-VALID_FILTERS = ["callbackUri", "notificationTypes", "vnfdId", "vnfPkgId", "operationalState", "usageState"]
+from .common import view_safe_call_with_log
+logger = logging.getLogger(__name__)
-def get_problem_details_serializer(status_code, error_message):
- problem_details = {
- "status": status_code,
- "detail": error_message
- }
- problem_details_serializer = ProblemDetailsSerializer(data=problem_details)
- problem_details_serializer.is_valid()
- return problem_details_serializer
+VALID_FILTERS = [
+ "callbackUri",
+ "notificationTypes",
+ "vnfdId",
+ "vnfPkgId",
+ "operationalState",
+ "usageState"
+]
class CreateQuerySubscriptionView(APIView):
status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"
}
)
+ @view_safe_call_with_log(logger=logger)
def post(self, request):
logger.debug("Create VNF package Subscription> %s" % request.data)
- try:
- vnf_pkg_subscription_request = validate_data(request.data, PkgmSubscriptionRequestSerializer)
- data = CreateSubscription(vnf_pkg_subscription_request.data).do_biz()
- subscription_info = validate_data(data, PkgmSubscriptionSerializer)
- return Response(data=subscription_info.data, status=status.HTTP_201_CREATED)
- except VnfPkgDuplicateSubscriptionException as e:
- logger.error(e.message)
- logger.error(traceback.format_exc())
- problem_details_serializer = get_problem_details_serializer(status.HTTP_303_SEE_OTHER,
- traceback.format_exc())
- return Response(data=problem_details_serializer.data, status=status.HTTP_303_SEE_OTHER)
- except Exception as e:
- logger.error(e.message)
- logger.error(traceback.format_exc())
- problem_details_serializer = get_problem_details_serializer(status.HTTP_500_INTERNAL_SERVER_ERROR,
- traceback.format_exc())
- return Response(data=problem_details_serializer.data, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ vnf_pkg_subscription_request = validate_data(request.data, PkgmSubscriptionRequestSerializer)
+ data = CreateSubscription(vnf_pkg_subscription_request.data).do_biz()
+ subscription_info = validate_data(data, PkgmSubscriptionSerializer)
+ return Response(data=subscription_info.data, status=status.HTTP_201_CREATED)
@swagger_auto_schema(
responses={
status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()
}
)
+ @view_safe_call_with_log(logger=logger)
def get(self, request):
logger.debug("SubscribeNotification--get::> %s" % request.query_params)
- try:
- if request.query_params and not set(request.query_params).issubset(set(VALID_FILTERS)):
- problem_details_serializer = get_problem_details_serializer(status.HTTP_400_BAD_REQUEST,
- "Not a valid filter")
- return Response(data=problem_details_serializer.data, status=status.HTTP_400_BAD_REQUEST)
- resp_data = QuerySubscription().query_multi_subscriptions(request.query_params)
- subscriptions_serializer = PkgmSubscriptionsSerializer(data=resp_data)
- if not subscriptions_serializer.is_valid():
- raise VnfPkgSubscriptionException(subscriptions_serializer.errors)
+ if request.query_params and not set(request.query_params).issubset(set(VALID_FILTERS)):
+ raise BadRequestException("Not a valid filter")
+
+ resp_data = QuerySubscription().query_multi_subscriptions(request.query_params)
- return Response(data=subscriptions_serializer.data, status=status.HTTP_200_OK)
+ subscriptions_serializer = PkgmSubscriptionsSerializer(data=resp_data)
+ if not subscriptions_serializer.is_valid():
+ raise VnfPkgSubscriptionException(subscriptions_serializer.errors)
- except Exception as e:
- logger.error(e.message)
- logger.error(traceback.format_exc())
- problem_details_serializer = get_problem_details_serializer(status.HTTP_500_INTERNAL_SERVER_ERROR,
- traceback.format_exc())
- return Response(data=problem_details_serializer.data, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+ return Response(data=subscriptions_serializer.data, status=status.HTTP_200_OK)
class QueryTerminateSubscriptionView(APIView):
status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()
}
)
+ @view_safe_call_with_log(logger=logger)
def get(self, request, subscriptionId):
logger.debug("SubscribeNotification--get::> %s" % subscriptionId)
- try:
-
- resp_data = QuerySubscription().query_single_subscription(subscriptionId)
-
- subscription_serializer = PkgmSubscriptionSerializer(data=resp_data)
- if not subscription_serializer.is_valid():
- raise VnfPkgSubscriptionException(subscription_serializer.errors)
-
- return Response(data=subscription_serializer.data, status=status.HTTP_200_OK)
- except SubscriptionDoesNotExistsException as e:
- logger.error(e.message)
- logger.error(traceback.format_exc())
- problem_details_serializer = get_problem_details_serializer(status.HTTP_404_NOT_FOUND,
- traceback.format_exc())
- return Response(data=problem_details_serializer.data, status=status.HTTP_404_NOT_FOUND)
- except Exception as e:
- logger.error(e.message)
- logger.error(traceback.format_exc())
- problem_details_serializer = get_problem_details_serializer(status.HTTP_500_INTERNAL_SERVER_ERROR,
- traceback.format_exc())
- return Response(data=problem_details_serializer.data, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ resp_data = QuerySubscription().query_single_subscription(subscriptionId)
+
+ subscription_serializer = PkgmSubscriptionSerializer(data=resp_data)
+ if not subscription_serializer.is_valid():
+ raise VnfPkgSubscriptionException(subscription_serializer.errors)
+
+ return Response(data=subscription_serializer.data, status=status.HTTP_200_OK)
@swagger_auto_schema(
responses={
status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()
}
)
+ @view_safe_call_with_log(logger=logger)
def delete(self, request, subscriptionId):
logger.debug("SubscribeNotification--get::> %s" % subscriptionId)
- try:
- TerminateSubscription().terminate(subscriptionId)
- return Response(status=status.HTTP_204_NO_CONTENT)
- except SubscriptionDoesNotExistsException as e:
- logger.error(e.message)
- logger.error(traceback.format_exc())
- problem_details_serializer = get_problem_details_serializer(status.HTTP_404_NOT_FOUND,
- traceback.format_exc())
- return Response(data=problem_details_serializer.data, status=status.HTTP_404_NOT_FOUND)
- except Exception as e:
- logger.error(e.message)
- logger.error(traceback.format_exc())
- problem_details_serializer = get_problem_details_serializer(status.HTTP_500_INTERNAL_SERVER_ERROR,
- traceback.format_exc())
- return Response(data=problem_details_serializer.data, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ TerminateSubscription().terminate(subscriptionId)
+ return Response(status=status.HTTP_204_NO_CONTENT)
Code Review / modeling / etsicatalog.git / blobdiff