import json
import logging
import os
-import requests
import uuid
-
from collections import Counter
-from rest_framework import status
+
+import requests
from requests.auth import HTTPBasicAuth
+from rest_framework import status
+
from catalog.packages import const
from catalog.pub.database.models import VnfPkgSubscriptionModel
from catalog.pub.exceptions import VnfPkgSubscriptionException, \
VnfPkgDuplicateSubscriptionException, SubscriptionDoesNotExistsException
from catalog.pub.utils.values import ignore_case_get
-from catalog.pub.config.config import MSB_BASE_URL
logger = logging.getLogger(__name__)
class CreateSubscription(object):
+ """
+ Create subscription info
+ """
def __init__(self, data):
self.data = data
self.filter = ignore_case_get(self.data, "filter", {})
ignore_case_get(self.filter, "vnfProductsFromProviders", [])
def check_callbackuri_connection(self):
+ """
+ Check if the callback uri can access
+ :return:
+ """
logger.debug("SubscribeNotification-post::> Sending GET request "
"to %s" % self.callback_uri)
try:
)
def do_biz(self):
+ """
+ Do business
+ :return:
+ """
self.subscription_id = str(uuid.uuid4())
self.check_valid_auth_info()
self.check_callbackuri_connection()
return subscription.toDict()
def check_valid_auth_info(self):
+ """
+ Check if the Auth info is valid
+ :return:
+ """
logger.debug("SubscribeNotification--post::> Validating Auth "
"details if provided")
if self.authentication.get("paramsBasic", {}) and const.BASIC not in self.authentication.get("authType"):
links = json.loads(subscription.links)
logger.error("Subscriptions has already exists with the same callbackUri and filter:%s" % links)
raise VnfPkgDuplicateSubscriptionException(
- "%s/%s" % (MSB_BASE_URL, links["self"]["href"]))
+ "/%s" % (links["self"]["href"]))
return True
def save_db(self):
+ """
+ Save the subscription info to DB
+ :return:
+ """
logger.debug("SubscribeNotification--post::> Saving the subscription "
"%s to the database" % self.subscription_id)
links = {
class QuerySubscription(object):
+ """
+ The class for query subscription
+ """
def query_multi_subscriptions(self, params):
+ """
+ Query subscriptions
+ :param params:
+ :return:
+ """
query_data = {}
logger.debug("QuerySubscription--get--multi--subscriptions--biz::> Check "
"for filter in query params %s" % params)
return [subscription.toDict() for subscription in subscriptions]
def query_single_subscription(self, subscription_id):
+ """
+ Query subscription by id
+ :param subscription_id:
+ :return:
+ """
logger.debug("QuerySingleSubscriptions--get--single--subscription--biz::> "
"ID: %s" % subscription_id)
class TerminateSubscription(object):
+ """
+ The class to terminate the subscription
+ """
def terminate(self, subscription_id):
logger.debug("TerminateSubscriptions--delete--biz::> "
"ID: %s" % subscription_id)