You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@libcloud.apache.org by to...@apache.org on 2020/08/30 14:50:10 UTC
[libcloud] 01/10: type annotation for MyPy,
changed signer creation and usage
This is an automated email from the ASF dual-hosted git repository.
tomaz pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/libcloud.git
commit 64a0d90273a80cfc386642e5bcf4658f1c4b48ae
Author: Tio Gobin <ti...@outscale.com>
AuthorDate: Fri Aug 14 18:39:57 2020 +0200
type annotation for MyPy, changed signer creation and usage
---
libcloud/common/osc.py | 36 ++--
libcloud/compute/drivers/outscale.py | 408 +++++++++++------------------------
2 files changed, 144 insertions(+), 300 deletions(-)
diff --git a/libcloud/common/osc.py b/libcloud/common/osc.py
index c030288..a00d155 100644
--- a/libcloud/common/osc.py
+++ b/libcloud/common/osc.py
@@ -29,7 +29,10 @@ class OSCRequestSigner(object):
Class which handles signing the outgoing AWS requests.
"""
- def __init__(self, access_key, access_secret, version, connection):
+ def __init__(self, access_key: str,
+ access_secret: str,
+ version: str,
+ connection):
"""
:param access_key: Access key.
:type access_key: ``str``
@@ -51,26 +54,26 @@ class OSCRequestSigner(object):
class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
@staticmethod
- def sign(key, msg):
+ def sign(key: str, msg: str):
return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
@staticmethod
- def _get_signed_headers(headers):
+ def _get_signed_headers(headers: dict):
return ';'.join([k.lower() for k in sorted(headers.keys())])
@staticmethod
- def _get_canonical_headers(headers):
+ def _get_canonical_headers(headers: dict):
return '\n'.join([':'.join([k.lower(), str(v).strip()])
for k, v in sorted(headers.items())]) + '\n'
@staticmethod
- def _get_request_params(params):
+ def _get_request_params(params: dict):
return '&'.join(["%s=%s" %
(urlquote(k, safe=''), urlquote(str(v), safe='~'))
for k, v in sorted(params.items())])
- def get_request_headers(self, service_name, region, action,
- data=None):
+ def get_request_headers(self, service_name: str, region: str, action: str,
+ data: dict = None):
date = datetime.utcnow()
host = "{}.{}.outscale.com".format(service_name, region)
headers = {
@@ -93,8 +96,11 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
headers.update({'Authorization': sig})
return headers
- def _get_authorization_v4_header(self, headers, dt, method='GET',
- path='/', data=None):
+ def _get_authorization_v4_header(self, headers: dict,
+ dt: datetime,
+ method: str = 'GET',
+ path: str = '/',
+ data: dict = None):
credentials_scope = self._get_credential_scope(dt=dt)
signed_headers = self._get_signed_headers(headers=headers)
signature = self._get_signature(headers=headers, dt=dt,
@@ -108,7 +114,8 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
's': signature
}
- def _get_signature(self, headers, dt, method, path, data):
+ def _get_signature(self, headers: dict, dt: datetime,
+ method: str, path: str, data: dict):
string_to_sign = self._get_string_to_sign(headers=headers, dt=dt,
method=method, path=path,
data=data)
@@ -116,14 +123,15 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
return hmac.new(signing_key, string_to_sign.encode('utf-8'),
hashlib.sha256).hexdigest()
- def _get_key_to_sign_with(self, key, dt):
+ def _get_key_to_sign_with(self, key: str, dt: datetime):
dt = dt.strftime('%Y%m%d')
k_date = self.sign(('OSC4' + key).encode('utf-8'), dt)
k_region = self.sign(k_date, self.connection.region_name)
k_service = self.sign(k_region, self.connection.service_name)
return self.sign(k_service, 'osc4_request')
- def _get_string_to_sign(self, headers, dt, method, path, data):
+ def _get_string_to_sign(self, headers: dict, dt: datetime, method: str,
+ path: str, data: dict):
canonical_request = self._get_canonical_request(headers=headers,
method=method,
path=path,
@@ -139,7 +147,7 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
self.connection.service_name,
'osc4_request'])
- def _get_canonical_request(self, headers, method, path, data="{}"):
+ def _get_canonical_request(self, headers, method, path, data = "{}"):
data = data if data else "{}"
return '\n'.join([
method,
@@ -148,4 +156,4 @@ class OSCRequestSignerAlgorithmV4(OSCRequestSigner):
self._get_canonical_headers(headers),
self._get_signed_headers(headers),
hashlib.sha256(data.encode('utf-8')).hexdigest()
- ])
+ ])
\ No newline at end of file
diff --git a/libcloud/compute/drivers/outscale.py b/libcloud/compute/drivers/outscale.py
index 370fbb4..5e8326b 100644
--- a/libcloud/compute/drivers/outscale.py
+++ b/libcloud/compute/drivers/outscale.py
@@ -36,11 +36,11 @@ class OutscaleNodeDriver(NodeDriver):
website = 'http://www.outscale.com'
def __init__(self,
- key=None,
- secret=None,
- region='eu-west-2',
- service='api',
- version='latest'
+ key: str = None,
+ secret: str = None,
+ region: str = 'eu-west-2',
+ service: str = 'api',
+ version: str = 'latest'
):
self.key = key
self.secret = secret
@@ -50,8 +50,12 @@ class OutscaleNodeDriver(NodeDriver):
self.connection.service_name = service
self.service_name = service
self.version = version
+ self.signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
+ access_secret=self.secret,
+ version=self.version,
+ connection=self.connection)
- def list_locations(self, dry_run=False):
+ def list_locations(self, dry_run: bool = False):
"""
Lists available regions details.
@@ -73,7 +77,7 @@ class OutscaleNodeDriver(NodeDriver):
action)
return requests.post(endpoint, data=data, headers=headers)
- def create_public_ip(self, dry_run=False):
+ def create_public_ip(self, dry_run: bool = False):
"""
Create a new public ip.
@@ -86,22 +90,15 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "CreatePublicIp"
data = json.dumps({"DryRun": dry_run})
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def delete_public_ip(self, dry_run=False,
- public_ip=None,
- public_ip_id=None):
+ def delete_public_ip(self, dry_run: bool = False,
+ public_ip: str = None,
+ public_ip_id: str = None):
"""
Delete instances.
@@ -128,20 +125,13 @@ class OutscaleNodeDriver(NodeDriver):
if public_ip_id is not None:
data.update({"PublicIpId": public_ip_id})
data = json.dumps(data)
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def list_public_ips(self, data="{}"):
+ def list_public_ips(self, data: str = "{}"):
"""
List all nodes.
@@ -153,20 +143,13 @@ class OutscaleNodeDriver(NodeDriver):
:rtype: ``dict``
"""
action = "ReadPublicIps"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def list_public_ip_ranges(self, dry_run=False):
+ def list_public_ip_ranges(self, dry_run: bool = False):
"""
Lists available regions details.
@@ -179,26 +162,19 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "ReadPublicIpRanges"
data = json.dumps({"DryRun": dry_run})
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
def attach_public_ip(self,
- allow_relink=None,
- dry_run=False,
- nic_id=None,
- vm_id=None,
- public_ip=None,
- public_ip_id=None,
+ allow_relink: bool = None,
+ dry_run: bool = False,
+ nic_id: str = None,
+ vm_id: str = None,
+ public_ip: str = None,
+ public_ip_id: str = None,
):
"""
Attach a volume.
@@ -246,22 +222,15 @@ class OutscaleNodeDriver(NodeDriver):
if allow_relink is not None:
data.update({"AllowRelink": allow_relink})
data = json.dumps(data)
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def detach_public_ip(self, public_ip=None,
- link_public_ip_id=None,
- dry_run=False):
+ def detach_public_ip(self, public_ip: str = None,
+ link_public_ip_id: str = None,
+ dry_run: bool = False):
"""
Detach a volume.
@@ -288,37 +257,30 @@ class OutscaleNodeDriver(NodeDriver):
if link_public_ip_id is not None:
data.update({"LinkPublicIpId": link_public_ip_id})
data = json.dumps(data)
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
def create_node(self,
- image_id,
- dry_run=False,
- block_device_mapping=None,
- boot_on_creation=True,
- bsu_optimized=True,
- client_token=None,
- deletion_protection=False,
- keypair_name=None,
- max_vms_count=None,
- min_vms_count=None,
- nics=None,
- performance=None,
- placement=None,
- private_ips=None,
- security_group_ids=None,
- security_groups=None,
- subnet_id=None,
+ image_id: str,
+ dry_run: bool = False,
+ block_device_mapping: dict = None,
+ boot_on_creation: bool = True,
+ bsu_optimized: bool = True,
+ client_token: str = None,
+ deletion_protection: bool = False,
+ keypair_name: str = None,
+ max_vms_count: int = None,
+ min_vms_count: int = None,
+ nics: dict = None,
+ performance: str = None,
+ placement: dict = None,
+ private_ips: [str] = None,
+ security_group_ids: [str] = None,
+ security_groups: [str] = None,
+ subnet_id: str = None,
):
"""
Create a new instance.
@@ -428,20 +390,13 @@ class OutscaleNodeDriver(NodeDriver):
data.update({"SubnetId": subnet_id})
action = "CreateVms"
data = json.dumps(data)
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def reboot_node(self, node_ids):
+ def reboot_node(self, node_ids: [str]):
"""
Reboot instances.
@@ -454,20 +409,13 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "RebootVms"
data = json.dumps({"VmIds": node_ids})
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def list_nodes(self, data="{}"):
+ def list_nodes(self, data: str = "{}"):
"""
List all nodes.
@@ -475,20 +423,13 @@ class OutscaleNodeDriver(NodeDriver):
:rtype: ``dict``
"""
action = "ReadVms"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def delete_node(self, node_ids):
+ def delete_node(self, node_ids: [str]):
"""
Delete instances.
@@ -500,14 +441,7 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "DeleteVms"
data = json.dumps({"VmIds": node_ids})
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
@@ -515,16 +449,16 @@ class OutscaleNodeDriver(NodeDriver):
def create_image(
self,
- architecture=None,
- vm_id=None,
- image_name=None,
- description=None,
- block_device_mapping=None,
- no_reboot=False,
- root_device_name=None,
- dry_run=False,
- source_region_name=None,
- file_location=None
+ architecture: str = None,
+ vm_id: str = None,
+ image_name: str = None,
+ description: str = None,
+ block_device_mapping: dict = None,
+ no_reboot: bool = False,
+ root_device_name: str = None,
+ dry_run: bool = False,
+ source_region_name: str = None,
+ file_location: str = None
):
"""
Create a new image.
@@ -592,20 +526,13 @@ class OutscaleNodeDriver(NodeDriver):
data.update({"FileLocation": file_location})
data = json.dumps(data)
action = "CreateImage"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def list_images(self, data="{}"):
+ def list_images(self, data: str = "{}"):
"""
List all images.
@@ -613,20 +540,13 @@ class OutscaleNodeDriver(NodeDriver):
:rtype: ``dict``
"""
action = "ReadImages"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def get_image(self, image_id):
+ def get_image(self, image_id: str):
"""
Get a specific image.
@@ -638,20 +558,13 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "ReadImages"
data = '{"Filters": {"ImageIds": ["' + image_id + '"]}}'
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def delete_image(self, image_id):
+ def delete_image(self, image_id: str):
"""
Delete an image.
@@ -663,20 +576,16 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "DeleteImage"
data = '{"ImageId": "' + image_id + '"}'
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def create_key_pair(self, name, dry_run=False, public_key=None):
+ def create_key_pair(self,
+ name: str,
+ dry_run: bool = False,
+ public_key: str = None):
"""
Create a new key pair.
@@ -702,20 +611,13 @@ class OutscaleNodeDriver(NodeDriver):
data.update({"PublicKey": public_key})
data = json.dumps(data)
action = "CreateKeypair"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def list_key_pairs(self, data="{}"):
+ def list_key_pairs(self, data: str = "{}"):
"""
List all key pairs.
@@ -723,20 +625,13 @@ class OutscaleNodeDriver(NodeDriver):
:rtype: ``dict``
"""
action = "ReadKeypairs"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def get_key_pair(self, name):
+ def get_key_pair(self, name: str):
"""
Get a specific key pair.
@@ -749,20 +644,13 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "ReadKeypairs"
data = '{"Filters": {"KeypairNames" : ["' + name + '"]}}'
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def delete_key_pair(self, name):
+ def delete_key_pair(self, name: str):
"""
Delete an image.
@@ -774,27 +662,20 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "DeleteKeypair"
data = '{"KeypairName": "' + name + '"}'
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
def create_snapshot(self,
- description=None,
- dry_run=False,
- file_location=None,
- snapshot_size=None,
- source_region_name=None,
- source_snapshot_id=None,
- volume_id=None,
+ description: str = None,
+ dry_run: bool = False,
+ file_location: str = None,
+ snapshot_size: int = None,
+ source_region_name: str = None,
+ source_snapshot_id: str = None,
+ volume_id: str = None,
):
"""
Create a new snapshot.
@@ -850,20 +731,13 @@ class OutscaleNodeDriver(NodeDriver):
data.update({"VolumeId": volume_id})
data = json.dumps(data)
action = "CreateSnapshot"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def list_snapshots(self, data="{}"):
+ def list_snapshots(self, data: str = "{}"):
"""
List all snapshots.
@@ -871,20 +745,13 @@ class OutscaleNodeDriver(NodeDriver):
:rtype: ``dict``
"""
action = "ReadSnapshots"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def delete_snapshot(self, snapshot_id):
+ def delete_snapshot(self, snapshot_id: str):
"""
Delete a snapshot.
@@ -897,14 +764,7 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "DeleteSnapshot"
data = '{"SnapshotId": "' + snapshot_id + '"}'
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
@@ -912,12 +772,12 @@ class OutscaleNodeDriver(NodeDriver):
def create_volume(
self,
- subregion_name,
- dry_run=False,
- iops=None,
- size=None,
- snapshot_id=None,
- volume_type=None,
+ subregion_name: str,
+ dry_run: bool = False,
+ iops: int = None,
+ size: int = None,
+ snapshot_id: str = None,
+ volume_type: str = None,
):
"""
Create a new volume.
@@ -965,20 +825,14 @@ class OutscaleNodeDriver(NodeDriver):
data.update({"VolumeType": volume_type})
data = json.dumps(data)
action = "CreateVolume"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
+
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def list_volumes(self, data="{}"):
+ def list_volumes(self, data: str = "{}"):
"""
List all volumes.
@@ -986,20 +840,13 @@ class OutscaleNodeDriver(NodeDriver):
:rtype: ``dict``
"""
action = "ReadVolumes"
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def delete_volume(self, volume_id):
+ def delete_volume(self, volume_id: str):
"""
Delete a volume.
@@ -1012,20 +859,13 @@ class OutscaleNodeDriver(NodeDriver):
"""
action = "DeleteVolume"
data = '{"VolumeId": "' + volume_id + '"}'
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def attach_volume(self, node_id, volume_id, device_name):
+ def attach_volume(self, node_id: str, volume_id: str, device_name: str):
"""
Attach a volume.
@@ -1049,20 +889,16 @@ class OutscaleNodeDriver(NodeDriver):
"VolumeId": volume_id,
"DeviceName": device_name
})
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
- def detach_volume(self, volume_id, dry_run=False, force_unlink=False):
+ def detach_volume(self,
+ volume_id: str,
+ dry_run: bool = False,
+ force_unlink: bool = False):
"""
Detach a volume.
@@ -1087,23 +923,23 @@ class OutscaleNodeDriver(NodeDriver):
if force_unlink is not None:
data.update({"ForceUnlink": force_unlink})
data = json.dumps(data)
- signer = OSCRequestSignerAlgorithmV4(access_key=self.key,
- access_secret=self.secret,
- version=self.version,
- connection=self.connection)
- headers = signer.get_request_headers(action=action,
- data=data,
- service_name=self.service_name,
- region=self.region)
+ headers = self._ex_generate_headers(action, data)
endpoint = self._get_outscale_endpoint(self.region,
self.version,
action)
return requests.post(endpoint, data=data, headers=headers)
@staticmethod
- def _get_outscale_endpoint(region, version, action):
+ def _get_outscale_endpoint(region: str, version: str, action: str):
return "https://api.{}.outscale.com/api/{}/{}".format(
region,
version,
action
)
+
+ def _ex_generate_headers(self, action: str, data: dict):
+ return self.signer.get_request_headers(action=action,
+ data=data,
+ service_name=self.service_name,
+ region=self.region)
+