1 # Copyright (C) 2021 Wind River Systems, Inc.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
15 from typing import List
16 # from o2ims.adapter import orm
17 from o2ims.domain import ocloud
18 from o2ims.domain.ocloud_repo import OcloudRepository, ResourceTypeRepository,\
19 ResourcePoolRepository, ResourceRepository, DeploymentManagerRepository,\
20 SubscriptionRepository
23 class OcloudSqlAlchemyRepository(OcloudRepository):
24 def __init__(self, session):
26 self.session = session
28 def _add(self, ocloud: ocloud.Ocloud):
29 self.session.add(ocloud)
30 # self.session.add_all(ocloud.deploymentManagers)
32 def _get(self, ocloud_id) -> ocloud.Ocloud:
33 return self.session.query(ocloud.Ocloud).filter_by(
34 oCloudId=ocloud_id).first()
36 def _list(self) -> List[ocloud.Ocloud]:
37 return self.session.query(ocloud.Ocloud).order_by(
38 ocloud.Ocloud.name).all()
40 def _update(self, ocloud: ocloud.Ocloud):
41 self.session.add(ocloud)
44 class ResouceTypeSqlAlchemyRepository(ResourceTypeRepository):
45 def __init__(self, session):
47 self.session = session
49 def _add(self, resourceType: ocloud.ResourceType):
50 self.session.add(resourceType)
52 def _get(self, resource_type_id) -> ocloud.ResourceType:
53 return self.session.query(ocloud.ResourceType).filter_by(
54 resourceTypeId=resource_type_id).first()
56 def _list(self) -> List[ocloud.ResourceType]:
57 return self.session.query()
59 def _update(self, resourceType: ocloud.ResourceType):
60 self.session.add(resourceType)
63 class ResourcePoolSqlAlchemyRepository(ResourcePoolRepository):
64 def __init__(self, session):
66 self.session = session
68 def _add(self, resourcePool: ocloud.ResourcePool):
69 self.session.add(resourcePool)
71 def _get(self, resource_pool_id) -> ocloud.ResourcePool:
72 return self.session.query(ocloud.ResourcePool).filter_by(
73 resourcePoolId=resource_pool_id).first()
75 def _list(self) -> List[ocloud.ResourcePool]:
76 return self.session.query()
78 def _update(self, resourcePool: ocloud.ResourcePool):
79 self.session.add(resourcePool)
82 class ResourceSqlAlchemyRepository(ResourceRepository):
83 def __init__(self, session):
85 self.session = session
87 def _add(self, resource: ocloud.Resource):
88 self.session.add(resource)
90 def _get(self, resource_id) -> ocloud.Resource:
91 return self.session.query(ocloud.Resource).filter_by(
92 resourceId=resource_id).first()
94 def _list(self) -> List[ocloud.Resource]:
95 return self.session.query()
97 def _update(self, resource: ocloud.Resource):
98 self.session.add(resource)
101 class DeploymentManagerSqlAlchemyRepository(DeploymentManagerRepository):
102 def __init__(self, session):
104 self.session = session
106 def _add(self, deployment_manager: ocloud.DeploymentManager):
107 self.session.add(deployment_manager)
109 def _get(self, deployment_manager_id) -> ocloud.DeploymentManager:
110 return self.session.query(ocloud.DeploymentManager).filter_by(
111 deploymentManagerId=deployment_manager_id).first()
113 def _list(self) -> List[ocloud.DeploymentManager]:
114 return self.session.query()
116 def _update(self, deployment_manager: ocloud.DeploymentManager):
117 self.session.add(deployment_manager)
120 class SubscriptionSqlAlchemyRepository(SubscriptionRepository):
121 def __init__(self, session):
123 self.session = session
125 def _add(self, subscription: ocloud.Subscription):
126 self.session.add(subscription)
128 def _get(self, subscription_id) -> ocloud.Subscription:
129 return self.session.query(ocloud.Subscription).filter_by(
130 subscriptionId=subscription_id).first()
132 def _list(self) -> List[ocloud.Subscription]:
133 return self.session.query()
135 def _update(self, subscription: ocloud.Subscription):
136 self.session.add(subscription)
138 def _delete(self, subscription_id):
139 self.session.query(ocloud.Subscription).filter_by(
140 subscriptionId=subscription_id).delete()