Add Subscription get and list API; Provide swagger openAPI doc
[pti/o2.git] / 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, resourcetype, \\r
18     resourcepool, deploymentmanager, subscription\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
170 \r
171 \r
172 def subscriptions(uow: unit_of_work.SqlAlchemyUnitOfWork):\r
173     with uow:\r
174         res = uow.session.execute(select(subscription))\r
175     return [dict(r) for r in res]\r
176 \r
177 \r
178 def subscription_one(subscriptionId: str,\r
179                      uow: unit_of_work.SqlAlchemyUnitOfWork):\r
180     with uow:\r
181         res = uow.session.execute(select(subscription).where(\r
182             subscription.c.subscriptionId == subscriptionId))\r
183         first = res.first()\r
184     return None if first is None else dict(first)\r