-
Notifications
You must be signed in to change notification settings - Fork 14
Expand file tree
/
Copy pathpythonclient.py
More file actions
112 lines (91 loc) · 4.3 KB
/
Copy pathpythonclient.py
File metadata and controls
112 lines (91 loc) · 4.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
from eayunstack_tools.credentials import get_nova_credentials_v2, get_cinder_credentials, get_neutron_credentials, get_keystone_credentials
from novaclient.client import Client as nova_Client
from cinderclient.v2.client import Client as cinder_Client
from neutronclient.v2_0.client import Client as neutron_Client
from novaclient.exceptions import NotFound
from glanceclient.v2.client import Client as glance_Client
from keystoneclient.v2_0.client import Client as keystone_Client
class PythonClient():
def __init__(self):
self.nova_credentials = get_nova_credentials_v2()
self.cinder_credentials = get_cinder_credentials()
self.neutron_credentials = get_neutron_credentials()
self.keystone_credentials = get_keystone_credentials()
self.novaclient = nova_Client(**self.nova_credentials)
self.cinderclient = cinder_Client(**self.cinder_credentials)
self.neutronclient = neutron_Client(**self.neutron_credentials)
self.keystoneclient = keystone_Client(**self.keystone_credentials)
glance_endpoint = self.keystone_get_endpoint('image')
self.glanceclient = glance_Client(glance_endpoint, token=self.keystoneclient.auth_token)
def nova_services_list(self):
services_list = self.novaclient.services.list()
return self.obj2dict(services_list)
def cinder_services_list(self):
services_list = self.cinderclient.services.list()
return self.obj2dict(services_list)
def neutron_agents_list(self):
agents_list = self.neutronclient.list_agents()['agents']
return agents_list
def obj2dict(self, services_list):
sl = []
for service in services_list:
s = {}
s['binary'] = service.binary
s['host'] = service.host
s['zone'] = service.zone
s['status'] = service.status
s['state'] = service.state
sl.append(s)
return sl
def cinder_get_volume(self, volume_id):
volume = self.cinderclient.volumes.get(volume_id)
return volume
def cinder_get_snapshots(self, volume_id):
snapshots = self.cinderclient.volume_snapshots.list(search_opts={'volume_id': volume_id,'all_tenants': 1})
return snapshots
def nova_delete_server_volume(self, server_id, volume_id):
self.novaclient.volumes.delete_server_volume(server_id, volume_id)
def nova_server(self, server_id):
server = self.novaclient.servers.get(server_id)
return server
def nova_volume(self, server_id, volume_id):
volume = self.novaclient.volumes.get_server_volume(server_id, volume_id)
return volume
def neutron_delete_port(self, port_id):
self.neutronclient.delete_port(port_id)
def neutron_router_list(self, tenant_id=None):
if tenant_id:
router_list = \
self.neutronclient.list_routers(tenant_id=tenant_id)['routers']
else:
router_list = self.neutronclient.list_routers()['routers']
return router_list
def neutron_router_port_list(self, router_id):
router_port_list = \
self.neutronclient.list_ports(device_id=router_id)['ports']
return router_port_list
def neutron_l3_agent_list_hosting_router(self, router_id):
l3_agent_list_hosting_router = \
self.neutronclient.list_l3_agent_hosting_routers(
router_id)['agents']
return l3_agent_list_hosting_router
def nova_flavor(self, flavor_id):
flavor = self.novaclient.flavors.get(flavor_id)
return flavor
def nova_server_exist(self, server_id):
try:
self.novaclient.servers.get(server_id)
return True
except NotFound:
return False
def cinder_get_tenant_id(self, snapshot_id):
tenant_id = self.cinderclient.volume_snapshots.get(snapshot_id)._info['os-extended-snapshot-attributes:project_id']
return tenant_id
def glance_get_images(self, tenant_id):
images = self.glanceclient.images.list(filters={'owner': tenant_id})
return images
def glance_delete_image(self, image_id):
self.glanceclient.images.delete(image_id)
def keystone_get_endpoint(self, service_type):
endpoint = self.keystoneclient.service_catalog.url_for(service_type=service_type)
return endpoint