from kamaki.cli.utils import remove_from_items, filter_dicts_by_dict
from kamaki.cli.errors import (
raiseCLIError, CLISyntaxError, CLIBaseUrlError, CLIInvalidArgument)
-from kamaki.clients.cyclades import CycladesClient, ClientError
+from kamaki.clients.cyclades import CycladesClient
from kamaki.cli.argument import (
FlagArgument, ValueArgument, KeyValueArgument, RepeatableArgument,
ProgressBarArgument, DateArgument, IntArgument)
server_cmds = CommandTree('server', 'Cyclades/Compute API server commands')
flavor_cmds = CommandTree('flavor', 'Cyclades/Compute API flavor commands')
-network_cmds = CommandTree('network', 'Cyclades/Compute API network commands')
-ip_cmds = CommandTree('ip', 'Cyclades/Compute API floating ip commands')
-_commands = [server_cmds, flavor_cmds, network_cmds, ip_cmds]
+_commands = [server_cmds, flavor_cmds]
about_authentication = '\nUser Authentication:\
timeout=timeout if current_status not in ('BUILD', ) else 100)
-class _network_wait(_service_wait):
-
- def _wait(self, net_id, current_status, timeout=60):
- super(_network_wait, self)._wait(
- 'Network', net_id, self.client.wait_network, current_status,
- timeout=timeout)
-
-
-class _firewall_wait(_service_wait):
-
- def _wait(self, server_id, current_status, timeout=60):
- super(_firewall_wait, self)._wait(
- 'Firewall of server',
- server_id, self.client.wait_firewall, current_status,
- timeout=timeout)
-
-
class _init_cyclades(_command_init):
@errors.generic.all
@addLogSettings
@command(server_cmds)
class server_info(_init_cyclades, _optional_json):
- """Detailed information on a Virtual Machine
- Contains:
- - name, id, status, create/update dates
- - network interfaces
- - metadata (e.g., os, superuser) and diagnostics
- - hardware flavor and os image ids
- """
+ """Detailed information on a Virtual Machine"""
+
+ arguments = dict(
+ addr=FlagArgument(
+ 'Show only the network interfaces of this virtual server',
+ '--nics'),
+ vnc=FlagArgument(
+ 'Show VNC connection information (valid for a short period)',
+ '--vnc-credentials'),
+ stats=FlagArgument('Get URLs for server statistics', '--stats')
+ )
@errors.generic.all
@errors.cyclades.connection
@errors.cyclades.server_id
def _run(self, server_id):
vm = self.client.get_server_details(server_id)
- uuids = self._uuids2usernames([vm['user_id'], vm['tenant_id']])
- vm['user_id'] += ' (%s)' % uuids[vm['user_id']]
- vm['tenant_id'] += ' (%s)' % uuids[vm['tenant_id']]
- self._print(vm, self.print_dict)
+ if self['addr']:
+ self._print(vm.get('attachments', []))
+ elif self['vnc']:
+ self.error(
+ '(!) For security reasons, the following credentials are '
+ 'invalidated\nafter a short time period, depending on the '
+ 'server settings\n')
+ self._print(
+ self.client.get_server_console(server_id), self.print_dict)
+ elif self['stats']:
+ self._print(
+ self.client.get_server_stats(server_id), self.print_dict)
+ else:
+ uuids = self._uuids2usernames([vm['user_id'], vm['tenant_id']])
+ vm['user_id'] += ' (%s)' % uuids[vm['user_id']]
+ vm['tenant_id'] += ' (%s)' % uuids[vm['tenant_id']]
+ self._print(vm, self.print_dict)
def main(self, server_id):
super(self.__class__, self)._run()
+ choose_one = ('addr', 'vnc', 'stats')
+ count = len([a for a in choose_one if self[a]])
+ if count > 1:
+ raise CLIInvalidArgument('Invalid argument compination', details=[
+ 'Arguments %s cannot be used simultaneously' % ', '.join(
+ [self.arguments[a].lvalue for a in choose_one])])
self._run(server_id=server_id)
'%s' % ve])
+class NetworkIpArgument(RepeatableArgument):
+
+ @property
+ def value(self):
+ return getattr(self, '_value', [])
+
+ @value.setter
+ def value(self, new_value):
+ for v in (new_value or []):
+ net_and_ip = v.split(',')
+ if len(net_and_ip) < 2:
+ raise CLIInvalidArgument(
+ 'Value "%s" is missing parts' % v,
+ details=['Correct format: %s NETWORK_ID,IP' % (
+ self.parsed_name[0])])
+ self._value = getattr(self, '_value', list())
+ self._value.append(
+ dict(uuid=net_and_ip[0], fixed_ip=net_and_ip[1]))
+
+
@command(server_cmds)
class server_create(_init_cyclades, _optional_json, _server_wait):
"""Create a server (aka Virtual Machine)"""
arguments = dict(
server_name=ValueArgument('The name of the new server', '--name'),
flavor_id=IntArgument('The ID of the hardware flavor', '--flavor-id'),
- image_id=IntArgument('The ID of the hardware image', '--image-id'),
+ image_id=ValueArgument('The ID of the hardware image', '--image-id'),
personality=PersonalityArgument(
(80 * ' ').join(howto_personality), ('-p', '--personality')),
wait=FlagArgument('Wait server to build', ('-w', '--wait')),
'Create a cluster of servers of this size. In this case, the name'
'parameter is the prefix of each server in the cluster (e.g.,'
'srv1, srv2, etc.',
- '--cluster-size')
+ '--cluster-size'),
+ max_threads=IntArgument(
+ 'Max threads in cluster mode (default 1)', '--threads'),
+ network_id=RepeatableArgument(
+ 'Connect server to network (can be repeated)', '--network'),
+ network_id_and_ip=NetworkIpArgument(
+ 'Connect server to network w. floating ip ( NETWORK_ID,IP )'
+ '(can be repeated)',
+ '--network-with-ip'),
+ automatic_ip=FlagArgument(
+ 'Automatically assign an IP to the server', '--automatic-ip')
)
required = ('server_name', 'flavor_id', 'image_id')
@errors.cyclades.cluster_size
def _create_cluster(self, prefix, flavor_id, image_id, size):
+ if self['automatic_ip']:
+ networks = []
+ else:
+ networks = [dict(uuid=netid) for netid in (
+ (self['network_id'] or []) + (self['network_id_and_ip'] or [])
+ )] or None
servers = [dict(
name='%s%s' % (prefix, i if size > 1 else ''),
flavor_id=flavor_id,
image_id=image_id,
- personality=self['personality']) for i in range(1, 1 + size)]
+ personality=self['personality'],
+ networks=networks) for i in range(1, 1 + size)]
if size == 1:
return [self.client.create_server(**servers[0])]
+ self.client.MAX_THREADS = int(self['max_threads'] or 1)
try:
r = self.client.async_run(self.client.create_server, servers)
return r
def main(self):
super(self.__class__, self)._run()
+ if self['automatic_ip'] and (
+ self['network_id'] or self['network_id_and_ip']):
+ raise CLIInvalidArgument('Invalid argument combination', details=[
+ 'Argument %s should not be combined with other' % (
+ self.arguments['automatic_ip'].lvalue),
+ 'network-related arguments i.e., %s or %s' % (
+ self.arguments['network_id'].lvalue,
+ self.arguments['network_id_and_ip'].lvalue)])
self._run(
name=self['server_name'],
flavor_id=self['flavor_id'],
'--firewall'),
metadata_to_set=KeyValueArgument(
'Set metadata in key=value form (can be repeated)',
- '--set-metadata'),
+ '--metadata-set'),
metadata_to_delete=RepeatableArgument(
- 'Delete metadata by key (can be repeated)', '--del-metadata')
+ 'Delete metadata by key (can be repeated)', '--metadata-del')
)
required = [
'server_name', 'flavor_id', 'firewall_profile', 'metadata_to_set',
- 'metadata_to_del']
+ 'metadata_to_delete']
@errors.generic.all
@errors.cyclades.connection
if self['metadata_to_set']:
self.client.update_server_metadata(
server_id, **self['metadata_to_set'])
- for key in self['metadata_to_delete']:
+ for key in (self['metadata_to_delete'] or []):
errors.cyclades.metadata(
self.client.delete_server_metadata)(server_id, key=key)
if self['with_output']:
@command(server_cmds)
-class server_console(_init_cyclades, _optional_json):
- """Get a VNC console to access an existing virtual server
- Console connection information provided (at least):
- - host: (url or address) a VNC host
- - port: (int) the gateway to enter virtual server on host
- - password: for VNC authorization
- """
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- def _run(self, server_id):
- self._print(
- self.client.get_server_console(int(server_id)), self.print_dict)
-
- def main(self, server_id):
- super(self.__class__, self)._run()
- self._run(server_id=server_id)
-
-
-@command(server_cmds)
-class server_addr(_init_cyclades, _optional_json):
- """List the addresses of all network interfaces on a virtual server"""
-
- arguments = dict(
- enum=FlagArgument('Enumerate results', '--enumerate')
- )
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- def _run(self, server_id):
- reply = self.client.list_server_nics(int(server_id))
- self._print(reply, with_enumeration=self['enum'] and (reply) > 1)
-
- def main(self, server_id):
- super(self.__class__, self)._run()
- self._run(server_id=server_id)
-
-
-@command(server_cmds)
-class server_metadata_list(_init_cyclades, _optional_json):
- """Get server metadata"""
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- @errors.cyclades.metadata
- def _run(self, server_id, key=''):
- self._print(
- self.client.get_server_metadata(int(server_id), key),
- self.print_dict)
+class server_addr(_init_cyclades):
+ """DEPRECATED, use: [kamaki] server info SERVER_ID --nics"""
- def main(self, server_id, key=''):
- super(self.__class__, self)._run()
- self._run(server_id=server_id, key=key)
+ def main(self, *args):
+ raiseCLIError('DEPRECATED since v0.12', importance=3, details=[
+ 'Replaced by',
+ ' [kamaki] server info <SERVER_ID> --nics'])
@command(server_cmds)
-class server_metadata_set(_init_cyclades, _optional_json):
- """Set / update virtual server metadata
- Metadata should be given in key/value pairs in key=value format
- For example: /server metadata set <server id> key1=value1 key2=value2
- Old, unreferenced metadata will remain intact
- """
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- def _run(self, server_id, keyvals):
- assert keyvals, 'Please, add some metadata ( key=value)'
- metadata = dict()
- for keyval in keyvals:
- k, sep, v = keyval.partition('=')
- if sep and k:
- metadata[k] = v
- else:
- raiseCLIError(
- 'Invalid piece of metadata %s' % keyval,
- importance=2, details=[
- 'Correct metadata format: key=val',
- 'For example:',
- '/server metadata set <server id>'
- 'key1=value1 key2=value2'])
- self._print(
- self.client.update_server_metadata(int(server_id), **metadata),
- self.print_dict)
+class server_console(_init_cyclades, _optional_json):
+ """DEPRECATED, use: [kamaki] server info SERVER_ID --vnc-credentials"""
- def main(self, server_id, *key_equals_val):
- super(self.__class__, self)._run()
- self._run(server_id=server_id, keyvals=key_equals_val)
+ def main(self, *args):
+ raiseCLIError('DEPRECATED since v0.12', importance=3, details=[
+ 'Replaced by',
+ ' [kamaki] server info <SERVER_ID> --vnc-credentials'])
@command(server_cmds)
-class server_metadata_delete(_init_cyclades, _optional_output_cmd):
- """Delete virtual server metadata"""
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- @errors.cyclades.metadata
- def _run(self, server_id, key):
- self._optional_output(
- self.client.delete_server_metadata(int(server_id), key))
+class server_rename(_init_cyclades, _optional_json):
+ """DEPRECATED, use: [kamaki] server modify SERVER_ID --name=NEW_NAME"""
- def main(self, server_id, key):
- super(self.__class__, self)._run()
- self._run(server_id=server_id, key=key)
+ def main(self, *args):
+ raiseCLIError('DEPRECATED since v0.12', importance=3, details=[
+ 'Replaced by',
+ ' [kamaki] server info <SERVER_ID> --stats'])
@command(server_cmds)
class server_stats(_init_cyclades, _optional_json):
- """Get virtual server statistics"""
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- def _run(self, server_id):
- self._print(
- self.client.get_server_stats(int(server_id)), self.print_dict)
+ """DEPRECATED, use: [kamaki] server info SERVER_ID --stats"""
- def main(self, server_id):
- super(self.__class__, self)._run()
- self._run(server_id=server_id)
+ def main(self, *args):
+ raiseCLIError('DEPRECATED since v0.12', importance=3, details=[
+ 'Replaced by',
+ ' [kamaki] server info <SERVER_ID> --stats'])
@command(server_cmds)
net['user_id'] += ' (%s)' % usernames[user_id]
if tenant_id:
net['tenant_id'] += ' (%s)' % usernames[tenant_id]
-
-
-@command(network_cmds)
-class network_info(_init_cyclades, _optional_json):
- """Detailed information on a network
- To get a list of available networks and network ids, try /network list
- """
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.network_id
- def _run(self, network_id):
- network = self.client.get_network_details(int(network_id))
- _add_name(self, network)
- self._print(network, self.print_dict, exclude=('id'))
-
- def main(self, network_id):
- super(self.__class__, self)._run()
- self._run(network_id=network_id)
-
-
-@command(network_cmds)
-class network_list(_init_cyclades, _optional_json, _name_filter, _id_filter):
- """List networks"""
-
- PERMANENTS = ('id', 'name')
-
- arguments = dict(
- detail=FlagArgument('show detailed output', ('-l', '--details')),
- limit=IntArgument('limit # of listed networks', ('-n', '--number')),
- more=FlagArgument(
- 'output results in pages (-n to set items per page, default 10)',
- '--more'),
- enum=FlagArgument('Enumerate results', '--enumerate'),
- status=ValueArgument('filter by status', ('--status')),
- public=FlagArgument('only public networks', ('--public')),
- private=FlagArgument('only private networks', ('--private')),
- dhcp=FlagArgument('show networks with dhcp', ('--with-dhcp')),
- no_dhcp=FlagArgument('show networks without dhcp', ('--without-dhcp')),
- user_id=ValueArgument('filter by user id', ('--user-id')),
- user_name=ValueArgument('filter by user name', ('--user-name')),
- gateway=ValueArgument('filter by gateway (IPv4)', ('--gateway')),
- gateway6=ValueArgument('filter by gateway (IPv6)', ('--gateway6')),
- cidr=ValueArgument('filter by cidr (IPv4)', ('--cidr')),
- cidr6=ValueArgument('filter by cidr (IPv6)', ('--cidr6')),
- type=ValueArgument('filter by type', ('--type')),
- )
-
- def _apply_common_filters(self, networks):
- common_filter = dict()
- if self['public']:
- if self['private']:
- return []
- common_filter['public'] = self['public']
- elif self['private']:
- common_filter['public'] = False
- if self['dhcp']:
- if self['no_dhcp']:
- return []
- common_filter['dhcp'] = True
- elif self['no_dhcp']:
- common_filter['dhcp'] = False
- if self['user_id'] or self['user_name']:
- uuid = self['user_id'] or self._username2uuid(self['user_name'])
- common_filter['user_id'] = uuid
- for term in ('status', 'gateway', 'gateway6', 'cidr', 'cidr6', 'type'):
- if self[term]:
- common_filter[term] = self[term]
- return filter_dicts_by_dict(networks, common_filter)
-
- def _add_name(self, networks, key='user_id'):
- uuids = self._uuids2usernames(
- list(set([net[key] for net in networks])))
- for net in networks:
- v = net.get(key, None)
- if v:
- net[key] += ' (%s)' % uuids[v]
- return networks
-
- @errors.generic.all
- @errors.cyclades.connection
- def _run(self):
- withcommons = False
- for term in (
- 'status', 'public', 'private', 'user_id', 'user_name', 'type',
- 'gateway', 'gateway6', 'cidr', 'cidr6', 'dhcp', 'no_dhcp'):
- if self[term]:
- withcommons = True
- break
- detail = self['detail'] or withcommons
- networks = self.client.list_networks(detail)
- networks = self._filter_by_name(networks)
- networks = self._filter_by_id(networks)
- if withcommons:
- networks = self._apply_common_filters(networks)
- if not (self['detail'] or (
- self['json_output'] or self['output_format'])):
- remove_from_items(networks, 'links')
- if detail and not self['detail']:
- for net in networks:
- for key in set(net).difference(self.PERMANENTS):
- net.pop(key)
- if self['detail'] and not (
- self['json_output'] or self['output_format']):
- self._add_name(networks)
- self._add_name(networks, 'tenant_id')
- kwargs = dict(with_enumeration=self['enum'])
- if self['more']:
- kwargs['out'] = StringIO()
- kwargs['title'] = ()
- if self['limit']:
- networks = networks[:self['limit']]
- self._print(networks, **kwargs)
- if self['more']:
- pager(kwargs['out'].getvalue())
-
- def main(self):
- super(self.__class__, self)._run()
- self._run()
-
-
-@command(network_cmds)
-class network_create(_init_cyclades, _optional_json, _network_wait):
- """Create an (unconnected) network"""
-
- arguments = dict(
- cidr=ValueArgument('explicitly set cidr', '--with-cidr'),
- gateway=ValueArgument('explicitly set gateway', '--with-gateway'),
- dhcp=FlagArgument('Use dhcp (default: off)', '--with-dhcp'),
- type=ValueArgument(
- 'Valid network types are '
- 'CUSTOM, IP_LESS_ROUTED, MAC_FILTERED (default), PHYSICAL_VLAN',
- '--with-type',
- default='MAC_FILTERED'),
- wait=FlagArgument('Wait network to build', ('-w', '--wait'))
- )
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.network_max
- def _run(self, name):
- r = self.client.create_network(
- name,
- cidr=self['cidr'],
- gateway=self['gateway'],
- dhcp=self['dhcp'],
- type=self['type'])
- _add_name(self, r)
- self._print(r, self.print_dict)
- if self['wait'] and r['status'] in ('PENDING', ):
- self._wait(r['id'], 'PENDING')
-
- def main(self, name):
- super(self.__class__, self)._run()
- self._run(name)
-
-
-@command(network_cmds)
-class network_rename(_init_cyclades, _optional_output_cmd):
- """Set the name of a network"""
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.network_id
- def _run(self, network_id, new_name):
- self._optional_output(
- self.client.update_network_name(int(network_id), new_name))
-
- def main(self, network_id, new_name):
- super(self.__class__, self)._run()
- self._run(network_id=network_id, new_name=new_name)
-
-
-@command(network_cmds)
-class network_delete(_init_cyclades, _optional_output_cmd, _network_wait):
- """Delete a network"""
-
- arguments = dict(
- wait=FlagArgument('Wait network to build', ('-w', '--wait'))
- )
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.network_in_use
- @errors.cyclades.network_id
- def _run(self, network_id):
- status = 'DELETED'
- if self['wait']:
- r = self.client.get_network_details(network_id)
- status = r['status']
- if status in ('DELETED', ):
- return
-
- r = self.client.delete_network(int(network_id))
- self._optional_output(r)
-
- if self['wait']:
- self._wait(network_id, status)
-
- def main(self, network_id):
- super(self.__class__, self)._run()
- self._run(network_id=network_id)
-
-
-@command(network_cmds)
-class network_connect(_init_cyclades, _optional_output_cmd):
- """Connect a server to a network"""
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- @errors.cyclades.network_id
- def _run(self, server_id, network_id):
- self._optional_output(
- self.client.connect_server(int(server_id), int(network_id)))
-
- def main(self, server_id, network_id):
- super(self.__class__, self)._run()
- self._run(server_id=server_id, network_id=network_id)
-
-
-@command(network_cmds)
-class network_disconnect(_init_cyclades):
- """Disconnect a nic that connects a server to a network
- Nic ids are listed as "attachments" in detailed network information
- To get detailed network information: /network info <network id>
- """
-
- @errors.cyclades.nic_format
- def _server_id_from_nic(self, nic_id):
- return nic_id.split('-')[1]
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- @errors.cyclades.nic_id
- def _run(self, nic_id, server_id):
- num_of_disconnected = self.client.disconnect_server(server_id, nic_id)
- if not num_of_disconnected:
- raise ClientError(
- 'Network Interface %s not found on server %s' % (
- nic_id, server_id),
- status=404)
- print('Disconnected %s connections' % num_of_disconnected)
-
- def main(self, nic_id):
- super(self.__class__, self)._run()
- server_id = self._server_id_from_nic(nic_id=nic_id)
- self._run(nic_id=nic_id, server_id=server_id)
-
-
-@command(network_cmds)
-class network_wait(_init_cyclades, _network_wait):
- """Wait for server to finish [PENDING, ACTIVE, DELETED]"""
-
- arguments = dict(
- timeout=IntArgument(
- 'Wait limit in seconds (default: 60)', '--timeout', default=60)
- )
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.network_id
- def _run(self, network_id, current_status):
- net = self.client.get_network_details(network_id)
- if net['status'].lower() == current_status.lower():
- self._wait(network_id, current_status, timeout=self['timeout'])
- else:
- self.error(
- 'Network %s: Cannot wait for status %s, '
- 'status is already %s' % (
- network_id, current_status, net['status']))
-
- def main(self, network_id, current_status='PENDING'):
- super(self.__class__, self)._run()
- self._run(network_id=network_id, current_status=current_status)
-
-
-@command(ip_cmds)
-class ip_pools(_init_cyclades, _optional_json):
- """List pools of floating IPs"""
-
- @errors.generic.all
- @errors.cyclades.connection
- def _run(self):
- r = self.client.get_floating_ip_pools()
- self._print(r if self['json_output'] or self['output_format'] else r[
- 'floating_ip_pools'])
-
- def main(self):
- super(self.__class__, self)._run()
- self._run()
-
-
-@command(ip_cmds)
-class ip_list(_init_cyclades, _optional_json):
- """List reserved floating IPs"""
-
- @errors.generic.all
- @errors.cyclades.connection
- def _run(self):
- r = self.client.get_floating_ips()
- self._print(r if self['json_output'] or self['output_format'] else r[
- 'floating_ips'])
-
- def main(self):
- super(self.__class__, self)._run()
- self._run()
-
-
-@command(ip_cmds)
-class ip_info(_init_cyclades, _optional_json):
- """Details for an IP"""
-
- @errors.generic.all
- @errors.cyclades.connection
- def _run(self, ip):
- self._print(self.client.get_floating_ip(ip), self.print_dict)
-
- def main(self, IP):
- super(self.__class__, self)._run()
- self._run(ip=IP)
-
-
-@command(ip_cmds)
-class ip_reserve(_init_cyclades, _optional_json):
- """Reserve a floating IP
- An IP is reserved from an IP pool. The default IP pool is chosen
- automatically, but there is the option if specifying an explicit IP pool.
- """
-
- arguments = dict(pool=ValueArgument('Source IP pool', ('--pool'), None))
-
- @errors.generic.all
- @errors.cyclades.connection
- def _run(self, ip=None):
- self._print([self.client.alloc_floating_ip(self['pool'], ip)])
-
- def main(self, requested_IP=None):
- super(self.__class__, self)._run()
- self._run(ip=requested_IP)
-
-
-@command(ip_cmds)
-class ip_release(_init_cyclades, _optional_output_cmd):
- """Release a floating IP
- The release IP is "returned" to the IP pool it came from.
- """
-
- @errors.generic.all
- @errors.cyclades.connection
- def _run(self, ip):
- self._optional_output(self.client.delete_floating_ip(ip))
-
- def main(self, IP):
- super(self.__class__, self)._run()
- self._run(ip=IP)
-
-
-@command(ip_cmds)
-class ip_attach(_init_cyclades, _optional_output_cmd):
- """Attach a floating IP to a server
- """
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- def _run(self, server_id, ip):
- self._optional_output(self.client.attach_floating_ip(server_id, ip))
-
- def main(self, server_id, IP):
- super(self.__class__, self)._run()
- self._run(server_id=server_id, ip=IP)
-
-
-@command(ip_cmds)
-class ip_detach(_init_cyclades, _optional_output_cmd):
- """Detach a floating IP from a server
- """
-
- @errors.generic.all
- @errors.cyclades.connection
- @errors.cyclades.server_id
- def _run(self, server_id, ip):
- self._optional_output(self.client.detach_floating_ip(server_id, ip))
-
- def main(self, server_id, IP):
- super(self.__class__, self)._run()
- self._run(server_id=server_id, ip=IP)