o2dms/api/dms_lcm_nfdeployment.py

   1 # Copyright (C) 2021 Wind River Systems, Inc.\r
   2 #\r
   3 #  Licensed under the Apache License, Version 2.0 (the "License");\r
   4 #  you may not use this file except in compliance with the License.\r
   5 #  You may obtain a copy of the License at\r
   6 #\r
   7 #      http://www.apache.org/licenses/LICENSE-2.0\r
   8 #\r
   9 #  Unless required by applicable law or agreed to in writing, software\r
  10 #  distributed under the License is distributed on an "AS IS" BASIS,\r
  11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  12 #  See the License for the specific language governing permissions and\r
  13 #  limitations under the License.\r
  14 \r
  15 from sqlalchemy import select\r
  16 import uuid\r
  17 from o2common.service import messagebus\r
  18 from o2dms.domain import events\r
  19 from o2common.service import unit_of_work\r
  20 from o2dms.adapter.orm import nfDeployment\r
  21 from o2dms.api.dms_dto import DmsLcmNfDeploymentDTO\r
  22 from o2dms.domain.dms import NfDeployment\r
  23 from o2common.helper import o2logging\r
  24 logger = o2logging.get_logger(__name__)\r
  25 \r
  26 \r
  27 def lcm_nfdeployment_list(\r
  28         deploymentManagerID: str, uow: unit_of_work.AbstractUnitOfWork):\r
  29     with uow:\r
  30         res = uow.session.execute(select(nfDeployment).where(\r
  31             nfDeployment.c.deploymentManagerId == deploymentManagerID))\r
  32     return [dict(r) for r in res]\r
  33 \r
  34 \r
  35 def lcm_nfdeployment_one(\r
  36         nfdeploymentid: str, uow: unit_of_work.AbstractUnitOfWork):\r
  37     with uow:\r
  38         res = uow.session.execute(select(nfDeployment).where(\r
  39             nfDeployment.c.id == nfdeploymentid))\r
  40         first = res.first()\r
  41     return None if first is None else dict(first)\r
  42 \r
  43 \r
  44 def lcm_nfdeployment_create(\r
  45         deploymentManagerId: str,\r
  46         input: DmsLcmNfDeploymentDTO.\r
  47         NfDeployment_create,\r
  48         bus: messagebus.MessageBus):\r
  49 \r
  50     uow = bus.uow\r
  51     with uow:\r
  52         _check_duplication(input, uow)\r
  53         _check_dependencies(input, uow)\r
  54         id = str(uuid.uuid4())\r
  55         entity = NfDeployment(\r
  56             id, input['name'], deploymentManagerId, input['description'],\r
  57             input['descriptorId'], input['parentDeploymentId'])\r
  58         uow.nfdeployments.add(entity)\r
  59 \r
  60         # publish event\r
  61         event = events.NfDeploymentCreated(NfDeploymentId=id)\r
  62         uow.commit()\r
  63     bus.handle(event)\r
  64 \r
  65     return id\r
  66 \r
  67 \r
  68 def lcm_nfdeployment_update(\r
  69         nfdeploymentid: str,\r
  70         input: DmsLcmNfDeploymentDTO.NfDeployment_update,\r
  71         uow: unit_of_work.AbstractUnitOfWork):\r
  72 \r
  73     with uow:\r
  74         entity = uow.nfdeployments.get(nfdeploymentid)\r
  75         entity.name = input['name']\r
  76         entity.description = input['description']\r
  77         entity.outputParams = input['parentDeploymentId']\r
  78         uow.commit()\r
  79     return True\r
  80 \r
  81 \r
  82 def lcm_nfdeployment_delete(\r
  83         nfdeploymentid: str, uow: unit_of_work.AbstractUnitOfWork):\r
  84 \r
  85     with uow:\r
  86         uow.nfdeployments.delete(nfdeploymentid)\r
  87         uow.commit()\r
  88     return True\r
  89 \r
  90 \r
  91 def _check_duplication(\r
  92         input: DmsLcmNfDeploymentDTO,\r
  93         uow: unit_of_work.AbstractUnitOfWork):\r
  94     name = input['name']\r
  95     descriptorId = input['descriptorId']\r
  96     if uow.nfdeployments.count(name=name) > 0:\r
  97         raise Exception(\r
  98             "NfDeployment with name {} exists already".format(name))\r
  99     if uow.nfdeployments.count(descriptorId=descriptorId) > 0:\r
 100         raise Exception(\r
 101             "NfDeployment with descriptorId {} exists already".format(\r
 102                 descriptorId))\r
 103 \r
 104 \r
 105 def _check_dependencies(\r
 106         input: DmsLcmNfDeploymentDTO,\r
 107         uow: unit_of_work.AbstractUnitOfWork):\r
 108     descriptorId = input['descriptorId']\r
 109     if uow.nfdeployment_descs.count(id=descriptorId) == 0:\r
 110         raise Exception(\r
 111             "NfDeploymentDescriptor with id {} does not exist".format(\r
 112                 descriptorId))\r