o2dms/views/dms_lcm_view.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 unit_of_work\r
  18 from o2ims.adapter.orm import deploymentmanager\r
  19 from o2dms.adapter.orm import nfDeploymentDesc, nfDeployment, nfOCloudVResource\r
  20 from o2dms.views.dms_dto import DmsLcmNfDeploymentDescriptorDTO\r
  21 from o2dms.views.dms_dto import DmsLcmNfDeploymentDTO\r
  22 from o2dms.domain.dms import NfDeploymentDesc, NfDeployment\r
  23 \r
  24 \r
  25 def deployment_managers(uow: unit_of_work.AbstractUnitOfWork):\r
  26     with uow:\r
  27         res = uow.session.execute(select(deploymentmanager))\r
  28     return [dict(r) for r in res]\r
  29 \r
  30 \r
  31 def deployment_manager_one(deploymentManagerId: str,\r
  32                            uow: unit_of_work.AbstractUnitOfWork):\r
  33     with uow:\r
  34         res = uow.session.execute(select(deploymentmanager).where(\r
  35             deploymentmanager.c.deploymentManagerId == deploymentManagerId))\r
  36         first = res.first()\r
  37     return None if first is None else dict(first)\r
  38 \r
  39 \r
  40 def lcm_nfdeploymentdesc_list(deploymentManagerID: str,\r
  41                               uow: unit_of_work.AbstractUnitOfWork):\r
  42     with uow:\r
  43         res = uow.session.execute(select(nfDeploymentDesc).where(\r
  44             nfDeploymentDesc.c.deploymentManagerId == deploymentManagerID))\r
  45     return [dict(r) for r in res]\r
  46 \r
  47 \r
  48 def lcm_nfdeploymentdesc_one(nfdeploymentdescriptorid: str,\r
  49                              uow: unit_of_work.AbstractUnitOfWork):\r
  50     with uow:\r
  51         res = uow.session.execute(select(nfDeploymentDesc).where(\r
  52             nfDeploymentDesc.c.id == nfdeploymentdescriptorid))\r
  53         first = res.first()\r
  54     return None if first is None else dict(first)\r
  55 \r
  56 \r
  57 def lcm_nfdeploymentdesc_create(\r
  58         deploymentManagerId: str,\r
  59         input: DmsLcmNfDeploymentDescriptorDTO.\r
  60         NfDeploymentDescriptor_create,\r
  61         uow: unit_of_work.AbstractUnitOfWork):\r
  62 \r
  63     id = str(uuid.uuid4())\r
  64     entity = NfDeploymentDesc(\r
  65         id, input['name'], deploymentManagerId, input['description'],\r
  66         input['inputParams'], input['outputParams'])\r
  67     with uow:\r
  68         uow.nfdeployment_descs.add(entity)\r
  69         uow.commit()\r
  70     return id\r
  71 \r
  72 \r
  73 def lcm_nfdeploymentdesc_update(\r
  74         nfdeploymentdescriptorid: str,\r
  75         input: DmsLcmNfDeploymentDescriptorDTO.NfDeploymentDescriptor_update,\r
  76         uow: unit_of_work.AbstractUnitOfWork):\r
  77 \r
  78     with uow:\r
  79         entity = uow.nfdeployment_descs.get(nfdeploymentdescriptorid)\r
  80         entity.name = input['name']\r
  81         entity.description = input['description']\r
  82         entity.inputParams = input['inputParams']\r
  83         entity.outputParams = input['outputParams']\r
  84         uow.commit()\r
  85     return True\r
  86 \r
  87 \r
  88 def lcm_nfdeploymentdesc_delete(\r
  89         nfdeploymentdescriptorid: str, uow: unit_of_work.AbstractUnitOfWork):\r
  90 \r
  91     with uow:\r
  92         uow.nfdeployment_descs.delete(nfdeploymentdescriptorid)\r
  93         uow.commit()\r
  94     return True\r
  95 \r
  96 \r
  97 def lcm_nfdeployment_list(\r
  98         deploymentManagerID: str, uow: unit_of_work.AbstractUnitOfWork):\r
  99     with uow:\r
 100         res = uow.session.execute(select(nfDeployment).where(\r
 101             nfDeployment.c.deploymentManagerId == deploymentManagerID))\r
 102     return [dict(r) for r in res]\r
 103 \r
 104 \r
 105 def lcm_nfdeployment_one(\r
 106         nfdeploymentid: str, uow: unit_of_work.AbstractUnitOfWork):\r
 107     with uow:\r
 108         res = uow.session.execute(select(nfDeployment).where(\r
 109             nfDeployment.c.id == nfdeploymentid))\r
 110         first = res.first()\r
 111     return None if first is None else dict(first)\r
 112 \r
 113 \r
 114 def lcm_nfdeployment_create(\r
 115         deploymentManagerId: str,\r
 116         input: DmsLcmNfDeploymentDTO.\r
 117         NfDeployment_create,\r
 118         uow: unit_of_work.AbstractUnitOfWork):\r
 119 \r
 120     id = str(uuid.uuid4())\r
 121     entity = NfDeployment(\r
 122         id, input['name'], deploymentManagerId, input['description'],\r
 123         input['descriptorId'], input['parentDeploymentId'])\r
 124     with uow:\r
 125         uow.nfdeployments.add(entity)\r
 126         uow.commit()\r
 127     return id\r
 128 \r
 129 \r
 130 def lcm_nfdeployment_update(\r
 131         nfdeploymentdescriptorid: str,\r
 132         input: DmsLcmNfDeploymentDTO.NfDeployment_update,\r
 133         uow: unit_of_work.AbstractUnitOfWork):\r
 134 \r
 135     with uow:\r
 136         entity = uow.nfdeployments.get(nfdeploymentdescriptorid)\r
 137         entity.name = input['name']\r
 138         entity.description = input['description']\r
 139         entity.inputParams = input['descriptorId']\r
 140         entity.outputParams = input['parentDeploymentId']\r
 141         uow.commit()\r
 142     return True\r
 143 \r
 144 \r
 145 def lcm_nfdeployment_delete(\r
 146         nfdeploymentdescriptorid: str, uow: unit_of_work.AbstractUnitOfWork):\r
 147 \r
 148     with uow:\r
 149         uow.nfdeployments.delete(nfdeploymentdescriptorid)\r
 150         uow.commit()\r
 151     return True\r
 152 \r
 153 \r
 154 def lcm_nfocloudvresource_list(\r
 155         nfDeploymentId: str, uow: unit_of_work.AbstractUnitOfWork):\r
 156     with uow:\r
 157         res = uow.session.execute(select(nfOCloudVResource).where(\r
 158             nfOCloudVResource.c.nfDeploymentId == nfDeploymentId))\r
 159     return [dict(r) for r in res]\r
 160 \r
 161 \r
 162 def lcm_nfocloudvresource_one(\r
 163         ocloudvresourceId: str, uow: unit_of_work.AbstractUnitOfWork):\r
 164     with uow:\r
 165         res = uow.session.execute(select(nfOCloudVResource).where(\r
 166             nfOCloudVResource.c.id == ocloudvresourceId))\r
 167         first = res.first()\r
 168     return None if first is None else dict(first)\r