Merge pull request #96 from akx/types

Add some type annotations based on Monkeytype trace

Author: ajmyyra

upcloud_api/cloud_manager/cloud_manager.py

@@ -30,7 +30,7 @@ class CloudManager(
     All other managers are mixed in so code can be organized in corresponding submanager classes.
     """
 
-    def __init__(self, username, password, timeout=60):
+    def __init__(self, username: str, password: str, timeout: int = 60) -> None:
         """
         Initiates CloudManager that handles all HTTP conections with UpCloud's API.
 

upcloud_api/cloud_manager/host_mixin.py

@@ -14,15 +14,15 @@ def get_hosts(self):
         res = self.get_request(url)
         return [Host(**host) for host in res['hosts']['host']]
 
-    def get_host(self, id):
+    def get_host(self, id: str) -> Host:
         """
         Returns detailed information about a specific host.
         """
         url = f'/host/{id}'
         res = self.get_request(url)
         return Host(**res['host'])
 
-    def modify_host(self, host, description):
+    def modify_host(self, host: str, description: str) -> Host:
         """
         Modifies description of a specific host.
         """

upcloud_api/cloud_manager/ip_address_mixin.py

@@ -6,7 +6,7 @@ class IPManager:
     Functions for managing IP-addresses. Intended to be used as a mixin for CloudManager.
     """
 
-    def get_ip(self, address):
+    def get_ip(self, address: str) -> IPAddress:
         """
         Get an IPAddress object with the IP address (string) from the API.
 
@@ -25,7 +25,7 @@ def get_ips(self, ignore_ips_without_server=False):
         )
         return IPs
 
-    def attach_ip(self, server, family='IPv4'):
+    def attach_ip(self, server: str, family: str = 'IPv4') -> IPAddress:
         """
         Attach a new (random) IPAddress to the given server (object or UUID).
         """
@@ -34,7 +34,7 @@ def attach_ip(self, server, family='IPv4'):
         res = self.post_request('/ip_address', body)
         return IPAddress(cloud_manager=self, **res['ip_address'])
 
-    def modify_ip(self, ip_addr, ptr_record):
+    def modify_ip(self, ip_addr: str, ptr_record: str) -> IPAddress:
         """
         Modify an IP address' ptr-record (Reverse DNS).
 

upcloud_api/cloud_manager/network_mixin.py

@@ -21,7 +21,7 @@ def get_networks(self, zone=None):
             network.ip_networks = [IpNetwork(**n) for n in network.ip_networks.get('ip_network')]
         return networks
 
-    def get_network(self, uuid):
+    def get_network(self, uuid: str) -> Network:
         """
         Retrieves the details of a specific network.
         """
@@ -198,15 +198,15 @@ def get_routers(self):
         res = self.get_request(url)
         return [Router(**router) for router in res['routers']['router']]
 
-    def get_router(self, uuid):
+    def get_router(self, uuid: str) -> Router:
         """
         Returns detailed information about a specific router.
         """
         url = f'/router/{uuid}'
         res = self.get_request(url)
         return Router(**res['router'])
 
-    def create_router(self, name):
+    def create_router(self, name: str) -> Router:
         """
         Creates a new router.
         """
@@ -215,7 +215,7 @@ def create_router(self, name):
         res = self.post_request(url, body)
         return Router(**res['router'])
 
-    def modify_router(self, router, name):
+    def modify_router(self, router: str, name: str) -> Router:
         """
         Modify an existing router.
         """

upcloud_api/cloud_manager/object_storage_mixin.py

@@ -21,8 +21,14 @@ def get_object_storages(self):
         return object_storages
 
     def create_object_storage(
-        self, zone, access_key, secret_key, size, name=None, description=None
-    ):
+        self,
+        zone: str,
+        access_key: str,
+        secret_key: str,
+        size: int,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+    ) -> ObjectStorage:
         """
         Used to create a new Object Storage device with a given name, size and location.
         """
@@ -42,7 +48,7 @@ def create_object_storage(
         res = self.post_request(url, body)
         return ObjectStorage(cloud_manager=self, **res['object_storage'])
 
-    def get_object_storage(self, uuid):
+    def get_object_storage(self, uuid: str) -> ObjectStorage:
         """
         A request to get details about a specific Object Storage device by the given uuid.
         """
@@ -51,8 +57,13 @@ def get_object_storage(self, uuid):
         return ObjectStorage(cloud_manager=self, **res['object_storage'])
 
     def modify_object_storage(
-        self, object_storage, access_key=None, secret_key=None, description=None, size=None
-    ):
+        self,
+        object_storage: str,
+        access_key: Optional[str] = None,
+        secret_key: Optional[str] = None,
+        description: Optional[str] = None,
+        size: Optional[int] = None,
+    ) -> ObjectStorage:
         """
         Modify requests can be used to update the details of an Object Storage including description, access_key and secret_key.
         """

upcloud_api/cloud_manager/server_mixin.py

@@ -49,7 +49,7 @@ def get_servers(self, populate=False, tags_has_one=None, tags_has_all=None):
 
         return server_list
 
-    def get_server(self, UUID):
+    def get_server(self, UUID: str) -> Server:
         """
         Return a (populated) Server instance.
         """
@@ -73,7 +73,7 @@ def get_server_by_ip(self, ip_address):
         UUID = data['ip_address']['server']
         return self.get_server(UUID)
 
-    def create_server(self, server):
+    def create_server(self, server: Server) -> Server:
         """
         Create a server and its storages based on a (locally created) Server object.
 
@@ -117,7 +117,7 @@ def create_server(self, server):
         server_to_return._reset(res['server'], cloud_manager=self, populated=True)
         return server_to_return
 
-    def modify_server(self, UUID, **kwargs):
+    def modify_server(self, uuid: str, **kwargs) -> Server:
         """
         modify_server allows updating the server's updateable_fields.
 
@@ -130,7 +130,7 @@ def modify_server(self, UUID, **kwargs):
                 Exception(f'{arg} is not an updateable field')
             body['server'][arg] = kwargs[arg]
 
-        res = self.put_request(f'/server/{UUID}', body)
+        res = self.put_request(f'/server/{uuid}', body)
         server = res['server']
 
         # Populate subobjects
@@ -148,15 +148,15 @@ def modify_server(self, UUID, **kwargs):
             cloud_manager=self,
         )
 
-    def delete_server(self, UUID):
+    def delete_server(self, uuid):
         """
         DELETE '/server/UUID'. Permanently destroys the virtual machine.
 
         DOES NOT remove the storage disks.
 
         Returns an empty object.
         """
-        return self.delete_request(f'/server/{UUID}')
+        return self.delete_request(f'/server/{uuid}')
 
     def get_server_data(self, UUID):
         """

upcloud_api/cloud_manager/storage_mixin.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, Union
 
 from upcloud_api.storage import Storage
 from upcloud_api.storage_import import StorageImport
@@ -29,7 +29,7 @@ def get_templates(self):
             templates.append({item.get('title'): item.get('uuid')})
         return templates
 
-    def get_storage(self, storage):
+    def get_storage(self, storage: str) -> Storage:
         """
         Return a Storage object from the API.
         """
@@ -38,12 +38,12 @@ def get_storage(self, storage):
 
     def create_storage(
         self,
-        size=10,
-        tier='maxiops',
-        title='Storage disk',
-        zone='fi-hel1',
+        size: int = 10,
+        tier: str = 'maxiops',
+        title: str = 'Storage disk',
+        zone: str = 'fi-hel1',
         backup_rule: Optional[dict] = None,
-    ):
+    ) -> Storage:
         """
         Create a Storage object. Returns an object based on the API's response.
         """
@@ -71,7 +71,9 @@ def _modify_storage(self, storage, size, title, backup_rule: Optional[dict] = No
             body['storage']['backup_rule'] = backup_rule
         return self.put_request('/storage/' + str(storage), body)
 
-    def modify_storage(self, storage, size, title, backup_rule: Optional[dict] = None):
+    def modify_storage(
+        self, storage: str, size: int, title: str, backup_rule: Optional[dict] = None
+    ) -> Storage:
         """
         Modify a Storage object. Returns an object based on the API's response.
         """
@@ -84,13 +86,16 @@ def delete_storage(self, UUID):
         """
         return self.delete_request('/storage/' + UUID)
 
-    def clone_storage(self, storage, title, zone, tier=None):
+    def clone_storage(
+        self, storage: Union[Storage, str], title: str, zone: str, tier=None
+    ) -> Storage:
         """
         Clones a Storage object. Returns an object based on the API's response.
         """
         body = {'storage': {'title': title, 'zone': zone}}
         if tier:
             body['storage']['tier'] = tier
+        # TODO: `str(storage)` seems unsafe
         res = self.post_request(f'/storage/{str(storage)}/clone', body)
         return Storage(cloud_manager=self, **res['storage'])
 
@@ -144,7 +149,7 @@ def eject_cd_rom(self, server):
         res = self.post_request(url)
         return Storage._create_storage_objs(res['server']['storage_devices'], cloud_manager=self)
 
-    def create_storage_backup(self, storage, title):
+    def create_storage_backup(self, storage: str, title: str) -> Storage:
         """
         Creates a point-in-time backup of a storage resource.
         """
@@ -160,7 +165,7 @@ def restore_storage_backup(self, storage):
         url = f'/storage/{storage}/restore'
         return self.post_request(url)
 
-    def templatize_storage(self, storage, title):
+    def templatize_storage(self, storage: str, title: str) -> Storage:
         """
         Creates an exact copy of an existing storage resource which can be used as a template for creating new servers.
         """
@@ -169,7 +174,9 @@ def templatize_storage(self, storage, title):
         res = self.post_request(url, body)
         return Storage(cloud_manager=self, **res['storage'])
 
-    def create_storage_import(self, storage, source, source_location=None):
+    def create_storage_import(
+        self, storage: str, source: str, source_location=None
+    ) -> StorageImport:
         """
         Creates an import task to import data into an existing storage.
         Source types: http_import or direct_upload.
@@ -190,15 +197,15 @@ def upload_file_for_storage_import(self, storage_import, file):
         body = {'data': data}
         return self.put_request(url, body, timeout=600, request_to_api=False)
 
-    def get_storage_import_details(self, storage):
+    def get_storage_import_details(self, storage: str) -> StorageImport:
         """
         Returns detailed information of an ongoing or finished import task.
         """
         url = f'/storage/{storage}/import'
         res = self.get_request(url)
         return StorageImport(**res['storage_import'])
 
-    def cancel_storage_import(self, storage):
+    def cancel_storage_import(self, storage: str) -> StorageImport:
         """
         Cancels an ongoing import task.
         """

upcloud_api/cloud_manager/tag_mixin.py

@@ -15,12 +15,14 @@ def get_tags(self):
         res = self.get_request('/tag')
         return [Tag(cloud_manager=self, **tag) for tag in res['tags']['tag']]
 
-    def get_tag(self, name):
+    def get_tag(self, name: str) -> Tag:
         """Return the tag as Tag object."""
         res = self.get_request('/tag/' + name)
         return Tag(cloud_manager=self, **res['tag'])
 
-    def create_tag(self, name, description=None, servers: Optional[list] = None):
+    def create_tag(
+        self, name: str, description: Optional[str] = None, servers: Optional[list] = None
+    ) -> Tag:
         """
         Create a new Tag. Only name is mandatory.
 

upcloud_api/ip_address.py

@@ -28,7 +28,7 @@ class IPAddress(UpCloudResource):
         'server': None,
     }
 
-    def save(self):
+    def save(self) -> None:
         """
         IPAddress can only change its PTR record. Saves the current state, PUT /ip_address/uuid.
         """