Type the OpenStack provider (pragmatic tier) + conform to interface

Annotate cloudbridge/providers/openstack/ (~341 callables) and add
providers.openstack.* to the pragmatic mypy tier. Conform implementations to
the interface per the established decisions:

- Instance.start/stop -> None; Router.attach_subnet/detach_subnet -> None (drop
  bool returns; honor the Subnet | str arg); BucketObject.upload/upload_from_file
  return the BucketObject; BucketObject/DnsRecord/FloatingIPService.delete -> None;
  ImageService.find wraps its result in ClientPagedResultList.
- VMFirewallRule.direction returns the TrafficDirection enum.
- Fatal id getters raise ProviderInternalException when absent (Instance.zone_id
  /subnet_id, Router.id) instead of returning None.
- GatewaySubService.get_or_create raises when no external network is available
  (was returning None against an -> InternetGateway contract).
- Remove the dead inspect.getargspec fallback in _clean_options (getargspec was
  removed in Python 3.11; the project requires 3.13).

OpenStack-specific provider members (nova/neutron/swift/keystone/os_conn) are
reached via cast("OpenStackCloudProvider", self._provider) through a
TYPE_CHECKING import. Verified: mypy strict/pragmatic green (43 files), flake8
clean, modules import under the OpenStack SDK env, and the AWS mock suite still
passes 97/5-skipped.

cloudbridge/providers/openstack/helpers.py

@@ -1,13 +1,19 @@
 """
 Helper functions
 """
+from __future__ import annotations
+
 import itertools
 import logging as log
+from typing import Any
+from typing import Sequence
 
 from cloudbridge.base.resources import ServerPagedResultList
+from cloudbridge.interfaces.provider import CloudProvider
 
 
-def os_result_limit(provider, requested_limit=None):
+def os_result_limit(provider: CloudProvider,
+                    requested_limit: int | None = None) -> int:
     """
     Calculates the limit for OpenStack.
     """
@@ -21,7 +27,8 @@ def os_result_limit(provider, requested_limit=None):
     return limit + 1
 
 
-def to_server_paged_list(provider, objects, limit=None):
+def to_server_paged_list(provider: CloudProvider, objects: Sequence[Any],
+                         limit: int | None = None) -> ServerPagedResultList[Any]:
     """
     A convenience function for wrapping a list of OpenStack native objects in
     a ServerPagedResultList. OpenStack
@@ -32,9 +39,9 @@ def to_server_paged_list(provider, objects, limit=None):
     limit = limit or provider.config.default_result_limit
     is_truncated = len(objects) > limit
     next_token = objects[limit-1].id if is_truncated else None
-    results = ServerPagedResultList(is_truncated,
-                                    next_token,
-                                    False)
+    results: ServerPagedResultList[Any] = ServerPagedResultList(is_truncated,
+                                                                next_token,
+                                                                False)
     for obj in itertools.islice(objects, limit):
         results.append(obj)
     return results

cloudbridge/providers/openstack/provider.py

@@ -1,6 +1,9 @@
 """Provider implementation based on OpenStack Python clients for OpenStack."""
+from __future__ import annotations
 
 import inspect
+from collections.abc import Callable
+from typing import Any
 
 from keystoneauth1 import session
 
@@ -17,8 +20,13 @@ from swiftclient import client as swift_client
 
 from cloudbridge.base import BaseCloudProvider
 from cloudbridge.base.helpers import get_env
-
+from cloudbridge.base.services import BaseCloudService
 from cloudbridge.interfaces.exceptions import ProviderConnectionException
+from cloudbridge.interfaces.services import ComputeService
+from cloudbridge.interfaces.services import DnsService
+from cloudbridge.interfaces.services import NetworkingService
+from cloudbridge.interfaces.services import SecurityService
+from cloudbridge.interfaces.services import StorageService
 
 from .services import OpenStackComputeService
 from .services import OpenStackDnsService
@@ -30,9 +38,9 @@ from .services import OpenStackStorageService
 class OpenStackCloudProvider(BaseCloudProvider):
     """OpenStack provider implementation."""
 
-    PROVIDER_ID = 'openstack'
+    PROVIDER_ID: str = 'openstack'
 
-    def __init__(self, config):
+    def __init__(self, config: dict[str, Any]) -> None:
         super(OpenStackCloudProvider, self).__init__(config)
 
         # Initialize cloud connection fields
@@ -64,14 +72,14 @@ class OpenStackCloudProvider(BaseCloudProvider):
             get_env('OS_USER_DOMAIN_NAME'))
 
         # Service connections, lazily initialized
-        self._nova = None
-        self._keystone = None
-        self._swift = None
-        self._neutron = None
-        self._os_conn = None
+        self._nova: Any = None
+        self._keystone: Any = None
+        self._swift: Any = None
+        self._neutron: Any = None
+        self._os_conn: Any = None
 
         # Additional cached variables
-        self._cached_keystone_session = None
+        self._cached_keystone_session: Any = None
 
         # Initialize provider services
         self._compute = OpenStackComputeService(self)
@@ -81,19 +89,19 @@ class OpenStackCloudProvider(BaseCloudProvider):
         self._dns = OpenStackDnsService(self)
 
     @property
-    def nova(self):
+    def nova(self) -> Any:
         if not self._nova:
             self._nova = self._connect_nova()
         return self._nova
 
     @property
-    def keystone(self):
+    def keystone(self) -> Any:
         if not self._keystone:
             self._keystone = self._connect_keystone()
         return self._keystone
 
     @property
-    def _keystone_version(self):
+    def _keystone_version(self) -> int:
         """
         Return the numeric version of remote Keystone server.
 
@@ -106,7 +114,7 @@ class OpenStackCloudProvider(BaseCloudProvider):
         return 2
 
     @property
-    def _keystone_session(self):
+    def _keystone_session(self) -> Any:
         """
         Connect to Keystone and return a session object.
 
@@ -118,6 +126,7 @@ class OpenStackCloudProvider(BaseCloudProvider):
 
         if self._keystone_version == 3:
             from keystoneauth1.identity import v3
+            auth: Any
             if self.username and self.password:
                 auth = v3.Password(auth_url=self.auth_url,
                                    username=self.username,
@@ -143,7 +152,7 @@ class OpenStackCloudProvider(BaseCloudProvider):
             self._cached_keystone_session = session.Session(auth=auth)
         return self._cached_keystone_session
 
-    def _connect_openstack(self):
+    def _connect_openstack(self) -> Any:
         return connection.Connection(
             region_name=self.region_name,
             user_agent='cloudbridge',
@@ -152,47 +161,47 @@ class OpenStackCloudProvider(BaseCloudProvider):
         )
 
     @property
-    def swift(self):
+    def swift(self) -> Any:
         if not self._swift:
             self._swift = self._connect_swift()
         return self._swift
 
     @property
-    def neutron(self):
+    def neutron(self) -> Any:
         if not self._neutron:
             self._neutron = self._connect_neutron()
         return self._neutron
 
     @property
-    def os_conn(self):
+    def os_conn(self) -> Any:
         if not self._os_conn:
             self._os_conn = self._connect_openstack()
         return self._os_conn
 
     @property
-    def compute(self):
+    def compute(self) -> ComputeService:
         return self._compute
 
     @property
-    def networking(self):
+    def networking(self) -> NetworkingService:
         return self._networking
 
     @property
-    def security(self):
+    def security(self) -> SecurityService:
         return self._security
 
     @property
-    def storage(self):
+    def storage(self) -> StorageService:
         return self._storage
 
     @property
-    def dns(self):
+    def dns(self) -> DnsService:
         return self._dns
 
-    def _connect_nova(self):
+    def _connect_nova(self) -> Any:
         return self._connect_nova_region(self.region_name)
 
-    def _connect_nova_region(self, region_name):
+    def _connect_nova_region(self, region_name: str | None) -> Any:
         """Get an OpenStack Nova (compute) client object."""
         # Force reauthentication with Keystone
         self._cached_keystone_session = None
@@ -215,7 +224,7 @@ class OpenStackCloudProvider(BaseCloudProvider):
                 http_log_debug=True if self.config.debug_mode else False)
         return nova
 
-    def _connect_keystone(self):
+    def _connect_keystone(self) -> Any:
         """Get an OpenStack Keystone (identity) client object."""
         if self._keystone_version == 3:
             return keystone_client.Client(session=self._keystone_session,
@@ -237,7 +246,8 @@ class OpenStackCloudProvider(BaseCloudProvider):
             return keystone
 
     @staticmethod
-    def _clean_options(options, method_to_match):
+    def _clean_options(options: dict[str, Any] | None,
+                       method_to_match: Callable[..., Any]) -> dict[str, Any]:
         """
         Returns a **copy** of the source options with all keys that are not in
         the ``method_to_match`` parameter list removed.
@@ -261,20 +271,17 @@ class OpenStackCloudProvider(BaseCloudProvider):
             then this will be an empty dictionary
         :rtype: ``dict``
         """
-        result = {}
+        result: dict[str, Any] = {}
         if options:
-            try:
-                method_signature = inspect.signature(method_to_match)
-                parameters = set(method_signature.parameters.keys())
-            except AttributeError:
-                parameters = set(inspect.getargspec(method_to_match).args)
+            method_signature = inspect.signature(method_to_match)
+            parameters = set(method_signature.parameters.keys())
             result = {key: val for key, val in options.items() if
                       key in parameters}
             # Don't allow the options to override our authentication
             result.pop('os_options', None)
         return result
 
-    def _connect_swift(self, options=None):
+    def _connect_swift(self, options: dict[str, Any] | None = None) -> Any:
         """
         Get an OpenStack Swift (object store) client connection.
 
@@ -297,42 +304,57 @@ class OpenStackCloudProvider(BaseCloudProvider):
             clean_options['session'] = self._keystone_session
         return swift_client.Connection(**clean_options)
 
-    def _connect_neutron(self):
+    def _connect_neutron(self) -> Any:
         """Get an OpenStack Neutron (networking) client object cloud."""
         return neutron_client.Client(auth_url=self.auth_url,
                                      session=self._keystone_session,
                                      region_name=self.region_name)
 
-    def service_zone_name(self, service):
+    def service_zone_name(self, service: BaseCloudService) -> str | None:
+        # ``service_zone_name`` is an OpenStack-specific attribute set in each
+        # service's __init__; it is not declared on the typed service
+        # interfaces, so reach it through ``Any``.
+        networking: Any = self.networking
+        security: Any = self.security
+        compute: Any = self.compute
+        storage: Any = self.storage
+        # ``zone_name`` is typed ``str | None`` on the interface, but the base
+        # implementation may return a dict at runtime (via ast.literal_eval);
+        # bind it to ``Any`` so the dict branches type-check.
+        zone_name: Any = self.zone_name
         service_name = service._service_event_pattern
         if "networking" in service_name:
-            if self.networking.service_zone_name:
-                return self.networking.service_zone_name
-            elif (isinstance(self.zone_name, dict) and
-                  self.zone_name.get("networking_zone")):
-                return self.zone_name.get("networking_zone")
+            if networking.service_zone_name:
+                return networking.service_zone_name
+            elif (isinstance(zone_name, dict) and
+                  zone_name.get("networking_zone")):
+                return zone_name.get("networking_zone")
         elif "security" in service_name:
-            if self.security.service_zone_name:
-                return self.security.service_zone_name
-            elif (isinstance(self.zone_name, dict) and
-                  self.zone_name.get("security_zone")):
-                return self.zone_name.get("security_zone")
+            if security.service_zone_name:
+                return security.service_zone_name
+            elif (isinstance(zone_name, dict) and
+                  zone_name.get("security_zone")):
+                return zone_name.get("security_zone")
         elif "compute" in service_name:
-            if self.compute.service_zone_name:
-                return self.compute.service_zone_name
-            elif (isinstance(self.zone_name, dict) and
-                  self.zone_name.get("compute_zone")):
-                return self.zone_name.get("compute_zone")
+            if compute.service_zone_name:
+                return compute.service_zone_name
+            elif (isinstance(zone_name, dict) and
+                  zone_name.get("compute_zone")):
+                return zone_name.get("compute_zone")
         elif "storage" in service_name:
-            if self.storage.service_zone_name:
-                return self.storage.service_zone_name
-            elif (isinstance(self.zone_name, dict) and
-                  self.zone_name.get("storage_zone")):
-                return self.zone_name.get("storage_zone")
-        elif (isinstance(self.zone_name, dict) and
-              self.zone_name.get("default_zone")):
-            return self.zone_name.get("default_zone")
-        elif isinstance(self.zone_name, str):
-            return self.zone_name
+            if storage.service_zone_name:
+                return storage.service_zone_name
+            elif (isinstance(zone_name, dict) and
+                  zone_name.get("storage_zone")):
+                return zone_name.get("storage_zone")
+        elif (isinstance(zone_name, dict) and
+              zone_name.get("default_zone")):
+            return zone_name.get("default_zone")
+        elif isinstance(zone_name, str):
+            return zone_name
         else:
             return None
+        # The branches above only return when an inner condition matched;
+        # fall through to ``None`` otherwise (preserving the original
+        # implicit return).
+        return None

cloudbridge/providers/openstack/subservices.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 
 from cloudbridge.base.subservices import BaseBucketObjectSubService
@@ -6,6 +8,12 @@ from cloudbridge.base.subservices import BaseFloatingIPSubService
 from cloudbridge.base.subservices import BaseGatewaySubService
 from cloudbridge.base.subservices import BaseSubnetSubService
 from cloudbridge.base.subservices import BaseVMFirewallRuleSubService
+from cloudbridge.interfaces.provider import CloudProvider
+from cloudbridge.interfaces.resources import Bucket
+from cloudbridge.interfaces.resources import DnsZone
+from cloudbridge.interfaces.resources import Gateway
+from cloudbridge.interfaces.resources import Network
+from cloudbridge.interfaces.resources import VMFirewall
 
 
 log = logging.getLogger(__name__)
@@ -13,36 +21,36 @@ log = logging.getLogger(__name__)
 
 class OpenStackBucketObjectSubService(BaseBucketObjectSubService):
 
-    def __init__(self, provider, bucket):
+    def __init__(self, provider: CloudProvider, bucket: Bucket) -> None:
         super(OpenStackBucketObjectSubService, self).__init__(provider, bucket)
 
 
 class OpenStackGatewaySubService(BaseGatewaySubService):
 
-    def __init__(self, provider, network):
+    def __init__(self, provider: CloudProvider, network: Network) -> None:
         super(OpenStackGatewaySubService, self).__init__(provider, network)
 
 
 class OpenStackFloatingIPSubService(BaseFloatingIPSubService):
 
-    def __init__(self, provider, gateway):
+    def __init__(self, provider: CloudProvider, gateway: Gateway) -> None:
         super(OpenStackFloatingIPSubService, self).__init__(provider, gateway)
 
 
 class OpenStackVMFirewallRuleSubService(BaseVMFirewallRuleSubService):
 
-    def __init__(self, provider, firewall):
+    def __init__(self, provider: CloudProvider, firewall: VMFirewall) -> None:
         super(OpenStackVMFirewallRuleSubService, self).__init__(
             provider, firewall)
 
 
 class OpenStackSubnetSubService(BaseSubnetSubService):
 
-    def __init__(self, provider, network):
+    def __init__(self, provider: CloudProvider, network: Network) -> None:
         super(OpenStackSubnetSubService, self).__init__(provider, network)
 
 
 class OpenStackDnsRecordSubService(BaseDnsRecordSubService):
 
-    def __init__(self, provider, dns_zone):
+    def __init__(self, provider: CloudProvider, dns_zone: DnsZone) -> None:
         super(OpenStackDnsRecordSubService, self).__init__(provider, dns_zone)

pyproject.toml

@@ -148,7 +148,11 @@ ignore_errors = true
 # through cast()/ignore adds noise without value. A more-specific module pattern
 # here wins over the broad providers.* exemption above.
 [[tool.mypy.overrides]]
-module = ["cloudbridge.providers.aws.*", "cloudbridge.providers.mock.*"]
+module = [
+    "cloudbridge.providers.aws.*",
+    "cloudbridge.providers.mock.*",
+    "cloudbridge.providers.openstack.*",
+]
 ignore_errors = false
 warn_return_any = false
 disallow_untyped_calls = false