o2dms/api/dms_lcm_nfdeploymentdesc.py

   1 # Copyright (C) 2021 Wind River Systems, Inc.
   2 #
   3 #  Licensed under the Apache License, Version 2.0 (the "License");
   4 #  you may not use this file except in compliance with the License.
   5 #  You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #  Unless required by applicable law or agreed to in writing, software
  10 #  distributed under the License is distributed on an "AS IS" BASIS,
  11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #  See the License for the specific language governing permissions and
  13 #  limitations under the License.
  14
  15 import json
  16 from sqlalchemy import select
  17 import uuid
  18 from o2common.service import unit_of_work
  19 from o2dms.adapter.orm import nfDeploymentDesc
  20 from o2dms.api.dms_dto import DmsLcmNfDeploymentDescriptorDTO
  21 from o2dms.domain.dms import NfDeploymentDesc
  22 from o2common.helper import o2logging
  23 logger = o2logging.get_logger(__name__)
  24
  25
  26 def lcm_nfdeploymentdesc_list(deploymentManagerID: str,
  27                               uow: unit_of_work.AbstractUnitOfWork):
  28     with uow:
  29         res = uow.session.execute(select(nfDeploymentDesc).where(
  30             nfDeploymentDesc.c.deploymentManagerId == deploymentManagerID))
  31     return [dict(r) for r in res]
  32
  33
  34 def lcm_nfdeploymentdesc_one(nfdeploymentdescriptorid: str,
  35                              uow: unit_of_work.AbstractUnitOfWork):
  36     with uow:
  37         res = uow.session.execute(select(nfDeploymentDesc).where(
  38             nfDeploymentDesc.c.id == nfdeploymentdescriptorid))
  39         first = res.first()
  40     return None if first is None else dict(first)
  41
  42
  43 def _check_duplication(name: str, uow: unit_of_work.AbstractUnitOfWork):
  44     if uow.nfdeployment_descs.count(name=name) > 0:
  45         raise Exception(
  46             "NfDeploymentDescriptor with name {} exists already".format(name))
  47
  48
  49 def lcm_nfdeploymentdesc_create(
  50         deploymentManagerId: str,
  51         input: DmsLcmNfDeploymentDescriptorDTO.
  52         NfDeploymentDescriptor_create,
  53         uow: unit_of_work.AbstractUnitOfWork):
  54
  55     with uow:
  56         _check_duplication(input['name'], uow)
  57         id = str(uuid.uuid4())
  58         entity = NfDeploymentDesc(
  59             id, input['name'], deploymentManagerId, input['description'],
  60             input['inputParams'], input['outputParams'],
  61             input['artifactRepoUrl'], input['artifactName'])
  62         _nfdeploymentdesc_validate(entity)
  63         uow.nfdeployment_descs.add(entity)
  64         uow.commit()
  65     return id
  66
  67
  68 def _nfdeploymentdesc_validate(desc: NfDeploymentDesc):
  69     try:
  70         if desc.inputParams:
  71             json.loads(desc.inputParams)
  72         if desc.outputParams:
  73             json.loads(desc.outputParams)
  74         if not desc.deploymentManagerId:
  75             raise Exception("Invalid deploymentManager Id")
  76         if not desc.artifactRepoUrl:
  77             raise Exception("Invalid artifactRepoUrl")
  78         if not desc.artifactName:
  79             raise Exception("Invalid artifactName")
  80         return
  81     except json.decoder.JSONDecodeError as e:
  82         logger.debug("NfDeploymentDesc json error with: %s" % (str(e)))
  83         raise e
  84     except Exception as e:
  85         logger.debug("NfDeploymentDesc validate error with: %s" % (str(e)))
  86         raise e
  87
  88
  89 def lcm_nfdeploymentdesc_update(
  90         nfdeploymentdescriptorid: str,
  91         input: DmsLcmNfDeploymentDescriptorDTO.NfDeploymentDescriptor_update,
  92         uow: unit_of_work.AbstractUnitOfWork):
  93
  94     with uow:
  95         entity = uow.nfdeployment_descs.get(nfdeploymentdescriptorid)
  96         entity.name = input['name']
  97         entity.description = input['description']
  98         entity.inputParams = input['inputParams']
  99         entity.outputParams = input['outputParams']
 100         entity.artifactRepoUrl = input['artifactRepoUrl']
 101         entity.artifactName = input['artifactName']
 102         uow.commit()
 103     return True
 104
 105
 106 def lcm_nfdeploymentdesc_delete(
 107         nfdeploymentdescriptorid: str, uow: unit_of_work.AbstractUnitOfWork):
 108
 109     with uow:
 110         # check dependency
 111         _check_dependencies(nfdeploymentdescriptorid, uow)
 112         uow.nfdeployment_descs.delete(nfdeploymentdescriptorid)
 113         uow.commit()
 114     return True
 115
 116
 117 def _check_dependencies(
 118     descriptorId: str, uow: unit_of_work.AbstractUnitOfWork
 119 ):
 120     # check if nfdeployment depends on it
 121     if uow.nfdeployments.count(descriptorId=descriptorId) > 0:
 122         raise Exception(
 123             "NfDeployment with descriptorId {} exists".format(
 124                 descriptorId))