o2ims/views/ocloud_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 \r
  17 from o2ims.adapter.orm import ocloud, resource, \\r
  18     resourcetype, resourcepool, deploymentmanager\r
  19 from o2ims.adapter import unit_of_work\r
  20 # from o2ims.domain.ocloud import Ocloud\r
  21 \r
  22 \r
  23 def oclouds(uow: unit_of_work.SqlAlchemyUnitOfWork):\r
  24     with uow:\r
  25         # res = uow.session.execute(\r
  26         #     """\r
  27         #     SELECT "oCloudId", "name" FROM ocloud\r
  28         #     """,\r
  29         # )\r
  30 \r
  31         res = uow.session.execute(select(ocloud))\r
  32     return [dict(r) for r in res]\r
  33 \r
  34 \r
  35 def ocloud_one(ocloudid: str, uow: unit_of_work.AbstractUnitOfWork):\r
  36     with uow:\r
  37         # res = uow.session.execute(\r
  38         #     """\r
  39         #     SELECT "oCloudId", "name" FROM ocloud\r
  40         #     WHERE "oCloudId" = :oCloudId\r
  41         #     """,\r
  42         #     dict(oCloudId=ocloudid),\r
  43         # )\r
  44         res = uow.session.execute(\r
  45             select(ocloud).where(ocloud.c.oCloudId == ocloudid))\r
  46         first = res.first()\r
  47     return None if first is None else dict(first)\r
  48 \r
  49 \r
  50 def resource_types(uow: unit_of_work.SqlAlchemyUnitOfWork):\r
  51     with uow:\r
  52         # res = uow.session.execute(\r
  53         #     """\r
  54         #     SELECT "resourceTypeId", "oCloudId", "name" FROM resourcetype\r
  55         #     """,\r
  56         # )\r
  57         res = uow.session.execute(select(resourcetype))\r
  58     return [dict(r) for r in res]\r
  59 \r
  60 \r
  61 def resource_type_one(resourceTypeId: str,\r
  62                       uow: unit_of_work.SqlAlchemyUnitOfWork):\r
  63     with uow:\r
  64         # res = uow.session.execute(\r
  65         #     """\r
  66         #     SELECT "resourceTypeId", "oCloudId", "name"\r
  67         #     FROM resourcetype WHERE "resourceTypeId" = :resourceTypeId\r
  68         #     """,\r
  69         #     dict(resourceTypeId=resourceTypeId),\r
  70         # )\r
  71         res = uow.session.execute(select(resourcetype).where(\r
  72             resourcetype.c.resourceTypeId == resourceTypeId))\r
  73         first = res.first()\r
  74     return None if first is None else dict(first)\r
  75 \r
  76 \r
  77 def resource_pools(uow: unit_of_work.SqlAlchemyUnitOfWork):\r
  78     with uow:\r
  79         # res = uow.session.execute(\r
  80         #     """\r
  81         #     SELECT "resourcePoolId", "oCloudId", "location", "name"\r
  82         #     FROM resourcepool\r
  83         #     """,\r
  84         # )\r
  85         res = uow.session.execute(select(resourcepool))\r
  86     return [dict(r) for r in res]\r
  87 \r
  88 \r
  89 def resource_pool_one(resourcePoolId: str,\r
  90                       uow: unit_of_work.SqlAlchemyUnitOfWork):\r
  91     with uow:\r
  92         # res = uow.session.execute(\r
  93         #     """\r
  94         #     SELECT "resourcePoolId", "oCloudId", "location", "name"\r
  95         #     FROM resourcepool\r
  96         #     WHERE "resourcePoolId" = :resourcePoolId\r
  97         #     """,\r
  98         #     dict(resourcePoolId=resourcePoolId),\r
  99         # )\r
 100         res = uow.session.execute(select(resourcepool).where(\r
 101             resourcepool.c.resourcePoolId == resourcePoolId))\r
 102         first = res.first()\r
 103     return None if first is None else dict(first)\r
 104 \r
 105 \r
 106 def resources(resourcePoolId: str, uow: unit_of_work.SqlAlchemyUnitOfWork):\r
 107     with uow:\r
 108         # res = uow.session.execute(\r
 109         #     """\r
 110         #     SELECT "resourceId", "parentId", "resourceTypeId",\r
 111         #           "resourcePoolId", "oCloudId"\r
 112         #     FROM resource\r
 113         #     WHERE "resourcePoolId" = :resourcePoolId\r
 114         #     """,\r
 115         #     dict(resourcePoolId=resourcePoolId),\r
 116         # )\r
 117         res = uow.session.execute(select(resource).where(\r
 118             resource.c.resourcePoolId == resourcePoolId))\r
 119     return [dict(r) for r in res]\r
 120 \r
 121 \r
 122 def resource_one(resourceId: str, uow: unit_of_work.SqlAlchemyUnitOfWork):\r
 123     with uow:\r
 124         # res = uow.session.execute(\r
 125         #     """\r
 126         #     SELECT "resourceId", "parentId", "resourceTypeId",\r
 127         #           "resourcePoolId", "oCloudId"\r
 128         #     FROM resource\r
 129         #     WHERE "resourceId" = :resourceId\r
 130         #     """,\r
 131         #     # AND "resourcePoolId" = :resourcePoolId\r
 132         #     # dict(resourcePoolId=resourcePoolId,\r
 133         #     dict(resourceId=resourceId),\r
 134         # )\r
 135         res = uow.session.execute(select(resource).where(\r
 136             resource.c.resourceId == resourceId))\r
 137         first = res.first()\r
 138     return None if first is None else dict(first)\r
 139 \r
 140 \r
 141 def deployment_managers(uow: unit_of_work.SqlAlchemyUnitOfWork):\r
 142     with uow:\r
 143         # res = uow.session.execute(\r
 144         #     """\r
 145         #     SELECT "deploymentManagerId", "oCloudId",\r
 146         #           "deploymentManagementServiceEndpoint", "name"\r
 147         #     FROM deploymentmanager\r
 148         #     """,\r
 149         # )\r
 150         res = uow.session.execute(select(deploymentmanager))\r
 151     return [dict(r) for r in res]\r
 152 \r
 153 \r
 154 def deployment_manager_one(deploymentManagerId: str,\r
 155                            uow: unit_of_work.SqlAlchemyUnitOfWork):\r
 156     with uow:\r
 157         # res = uow.session.execute(\r
 158         #     """\r
 159         #     SELECT "deploymentManagerId", "oCloudId",\r
 160         #           "deploymentManagementServiceEndpoint", "name"\r
 161         #     FROM deploymentmanager\r
 162         #     WHERE "deploymentManagerId" = :deploymentManagerId\r
 163         #     """,\r
 164         #     dict(deploymentManagerId=deploymentManagerId),\r
 165         # )\r
 166         res = uow.session.execute(select(deploymentmanager).where(\r
 167             deploymentmanager.c.deploymentManagerId == deploymentManagerId))\r
 168         first = res.first()\r
 169     return None if first is None else dict(first)\r