Type the AWS provider under the pragmatic tier

Annotate every callable in cloudbridge/providers/aws/ (provider, resources,
services, helpers, subservices; ~367 callables). AWS is the first provider
typed and validates the per-provider pattern for the rest.

mypy config: providers wrap untyped cloud SDKs, so providers.aws.* gets a
pragmatic tier (disallow_untyped_defs etc. ON, but warn_return_any and
disallow_untyped_calls OFF) so reading attributes off Any-typed boto3 objects
doesn't require a cast on every getter. providers.* stays exempt
(ignore_errors) until each is typed. base/ is finalised under the full strict
baseline (the temporary disallow_untyped_calls relaxation used while annotating
base is removed).

base/resources.py: the ResultList paging helpers (BaseResultList /
ClientPagedResultList) now accept Sequence[T] instead of list[T], so a
provider passing list[AWSVMType] where list[VMType] is expected no longer trips
list invariance. Benefits every provider.

Notable AWS specifics:
- `from __future__ import annotations` in resources.py/services.py so `list[...]`
  annotations don't collide with the `list` methods these classes define.
- AWS-specific provider members (ec2_conn/s3_conn/session) reached via
  cast("AWSCloudProvider", self._provider) through a TYPE_CHECKING import
  (avoids a runtime import cycle).
- Property getter/setter pairs reordered so the @label.setter immediately
  follows the getter (an untyped @tenacity.retry method in between was severing
  mypy's property link) - no behaviour change, and avoids type: ignore there.
- Two pre-existing latent bugs preserved and flagged in-place for a later fix:
  AWSGatewayService.delete logs self.id (a service has no id), and
  AWSRouter.detach_gateway returns a raw boto3 dict where the interface is None.

Verified: mypy strict green (43 files), flake8 clean, and the mock-provider
test suite (mock subclasses AWS) passes 97/5-skipped with no regression.

cloudbridge/base/resources.py

@@ -18,6 +18,7 @@ from concurrent.futures import wait
 from typing import Any
 from typing import IO
 from typing import Iterator
+from typing import Sequence
 from typing import TYPE_CHECKING
 from typing import TypeVar
 from typing import cast
@@ -204,7 +205,7 @@ class BaseResultList(ResultList[T]):
     def __init__(
             self, is_truncated: bool, marker: str | None,
             supports_total: bool, total: int | None = None,
-            data: list[T] | None = None) -> None:
+            data: Sequence[T] | None = None) -> None:
         # call list constructor
         super(BaseResultList, self).__init__(data or [])
         self._marker = marker
@@ -258,9 +259,9 @@ class ClientPagedResultList(BaseResultList[T]):
     of the full result set entirely on the client side.
     """
 
-    def __init__(self, provider: CloudProvider, objects: list[T],
+    def __init__(self, provider: CloudProvider, objects: Sequence[T],
                  limit: int | None = None, marker: str | None = None) -> None:
-        self._objects = objects
+        self._objects = list(objects)
         limit = limit or provider.config.default_result_limit
         total_size = len(objects)
         if marker:

cloudbridge/providers/aws/helpers.py

@@ -1,5 +1,10 @@
 """A set of AWS-specific helper methods used by the framework."""
+from __future__ import annotations
+
 import logging
+from typing import Any
+from typing import TYPE_CHECKING
+from typing import TypeVar
 
 from boto3.resources.params import create_request_parameters
 
@@ -9,12 +14,19 @@ from botocore.utils import merge_dicts
 
 from cloudbridge.base.resources import ClientPagedResultList
 from cloudbridge.base.resources import ServerPagedResultList
+from cloudbridge.interfaces.resources import CloudResource
+from cloudbridge.interfaces.resources import ResultList
+
+if TYPE_CHECKING:
+    from .provider import AWSCloudProvider
 
 
 log = logging.getLogger(__name__)
 
+T = TypeVar("T")
+
 
-def trim_empty_params(params_dict):
+def trim_empty_params(params_dict: dict[str, Any]) -> dict[str, Any]:
     """
     Given a dict containing potentially null values, trims out
     all the null values. This is to please Boto, which throws
@@ -35,7 +47,7 @@ def trim_empty_params(params_dict):
     return {k: v for k, v in params_dict.items() if v is not None}
 
 
-def find_tag_value(tags, key):
+def find_tag_value(tags: list[dict[str, Any]] | None, key: str) -> Any:
     """
     Finds the value associated with a given key from a list of AWS tags.
 
@@ -59,7 +71,9 @@ class BotoGenericService(object):
     resource, collection and paging support to implement
     basic cloudbridge methods.
     """
-    def __init__(self, provider, cb_resource, boto_conn, boto_collection_name):
+    def __init__(self, provider: AWSCloudProvider,
+                 cb_resource: Any, boto_conn: Any,
+                 boto_collection_name: str) -> None:
         """
         :type provider: :class:`AWSCloudProvider`
         :param provider: CloudBridge AWS provider to use
@@ -86,12 +100,12 @@ class BotoGenericService(object):
         self.boto_resource = self._infer_boto_resource(
             boto_conn, self.boto_collection_model)
 
-    def _infer_collection_model(self, conn, collection_name):
+    def _infer_collection_model(self, conn: Any, collection_name: str) -> Any:
         log.debug("Retrieving boto model for collection: %s", collection_name)
         return next(col for col in conn.meta.resource_model.collections
                     if col.name == collection_name)
 
-    def _infer_boto_resource(self, conn, collection_model):
+    def _infer_boto_resource(self, conn: Any, collection_model: Any) -> Any:
         log.debug("Retrieving resource model for collection: %s",
                   collection_model.name)
         resource_model = next(
@@ -99,7 +113,7 @@ class BotoGenericService(object):
             if sr.resource.model.name == collection_model.resource.model.name)
         return getattr(self.boto_conn, resource_model.name)
 
-    def get_raw(self, resource_id):
+    def get_raw(self, resource_id: str) -> Any:
         """
         Returns a single resource.
 
@@ -124,7 +138,7 @@ class BotoGenericService(object):
             else:
                 raise exc
 
-    def get(self, resource_id):
+    def get(self, resource_id: str) -> Any:
         """
         Returns a single resource.
 
@@ -139,7 +153,7 @@ class BotoGenericService(object):
         else:
             return None
 
-    def _get_list_operation(self):
+    def _get_list_operation(self) -> str:
         """
         This function discovers the list operation for a particular resource
         collection. For example, given the resource collection model for
@@ -147,7 +161,8 @@ class BotoGenericService(object):
         """
         return xform_name(self.boto_collection_model.request.operation)
 
-    def _to_boto_resource(self, collection, params, page):
+    def _to_boto_resource(self, collection: Any, params: Any,
+                          page: Any) -> Any:
         """
         This function duplicates some of the logic of the pages() method in
         boto.resources.collection.ResourceCollection. It will convert a raw
@@ -158,7 +173,8 @@ class BotoGenericService(object):
         # pylint:disable=protected-access
         return collection._handler(collection._parent, params, page)
 
-    def _get_paginated_results(self, limit, marker, collection):
+    def _get_paginated_results(self, limit: int | None, marker: str | None,
+                               collection: Any) -> tuple[Any, Any]:
         """
         If a Boto Paginator is available, use it. The results
         are converted back into BotoResources by directly accessing
@@ -177,7 +193,7 @@ class BotoGenericService(object):
         client = self.boto_conn.meta.client
         list_op = self._get_list_operation()
         paginator = client.get_paginator(list_op)
-        PaginationConfig = {}
+        PaginationConfig: dict[str, Any] = {}
         if limit:
             PaginationConfig = {'MaxItems': limit, 'PageSize': limit}
 
@@ -194,7 +210,8 @@ class BotoGenericService(object):
         resume_token = pages.resume_token
         return (resume_token, boto_objs)
 
-    def _make_query(self, collection, limit, marker):
+    def _make_query(self, collection: Any, limit: int | None,
+                    marker: str | None) -> tuple[str, Any, Any]:
         """
         Decide between server or client pagination,
         depending on the availability of a Boto Paginator.
@@ -213,7 +230,9 @@ class BotoGenericService(object):
                       " limit and page results.")
             return 'client', None, collection
 
-    def list(self, limit=None, marker=None, collection=None, **kwargs):
+    def list(self, limit: int | None = None, marker: str | None = None,
+             collection: Any = None,
+             **kwargs: Any) -> ResultList[CloudResource]:
         """
         List a set of resources.
 
@@ -244,8 +263,9 @@ class BotoGenericService(object):
             return ClientPagedResultList(self.provider, results,
                                          limit=limit, marker=marker)
 
-    def find(self, filters, limit=None, marker=None,
-             **kwargs):
+    def find(self, filters: dict[str, Any], limit: int | None = None,
+             marker: str | None = None,
+             **kwargs: Any) -> ResultList[CloudResource]:
         """
         Return a list of resources by filter.
 
@@ -261,7 +281,7 @@ class BotoGenericService(object):
             collection = collection.filter(**kwargs)
         return self.list(limit=limit, marker=marker, collection=collection)
 
-    def create(self, boto_method, **kwargs):
+    def create(self, boto_method: str, **kwargs: Any) -> Any:
         """
         Creates a resource
 
@@ -281,7 +301,7 @@ class BotoGenericService(object):
         else:
             return self.cb_resource(self.provider, result) if result else None
 
-    def delete(self, resource_id):
+    def delete(self, resource_id: str) -> None:
         """
         Deletes a resource by id
 
@@ -298,8 +318,9 @@ class BotoEC2Service(BotoGenericService):
     """
     Boto EC2 service implementation
     """
-    def __init__(self, provider, cb_resource,
-                 boto_collection_name):
+    def __init__(self, provider: AWSCloudProvider,
+                 cb_resource: Any,
+                 boto_collection_name: str) -> None:
         """
         :type provider: :class:`AWSCloudProvider`
         :param provider: CloudBridge AWS provider to use
@@ -320,8 +341,9 @@ class BotoS3Service(BotoGenericService):
     """
     Boto S3 service implementation.
     """
-    def __init__(self, provider, cb_resource,
-                 boto_collection_name):
+    def __init__(self, provider: AWSCloudProvider,
+                 cb_resource: Any,
+                 boto_collection_name: str) -> None:
         """
         :type provider: :class:`AWSCloudProvider`
         :param provider: CloudBridge AWS provider to use

cloudbridge/providers/aws/provider.py

@@ -1,5 +1,6 @@
 """Provider implementation based on boto library for AWS-compatible clouds."""
 import logging
+from typing import Any
 
 import boto3
 
@@ -7,6 +8,11 @@ from botocore.client import Config
 
 from cloudbridge.base import BaseCloudProvider
 from cloudbridge.base.helpers import get_env
+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 AWSComputeService
 from .services import AWSDnsService
@@ -20,9 +26,9 @@ log = logging.getLogger(__name__)
 
 class AWSCloudProvider(BaseCloudProvider):
     '''AWS cloud provider interface'''
-    PROVIDER_ID = 'aws'
+    PROVIDER_ID: str = 'aws'
 
-    def __init__(self, config):
+    def __init__(self, config: dict[str, Any]) -> None:
         super(AWSCloudProvider, self).__init__(config)
 
         # Initialize cloud connection fields
@@ -70,59 +76,59 @@ class AWSCloudProvider(BaseCloudProvider):
         self._dns = AWSDnsService(self)
 
     @property
-    def session(self):
+    def session(self) -> Any:
         '''Get a low-level session object or create one if needed'''
         if not self._session:
             if self.config.debug_mode:
-                boto3.set_stream_logger(level=log.DEBUG)
+                boto3.set_stream_logger(level=logging.DEBUG)
             self._session = boto3.session.Session(
                 region_name=self.region_name, **self.session_cfg)
         return self._session
 
     @property
-    def ec2_conn(self):
+    def ec2_conn(self) -> Any:
         if not self._ec2_conn:
             self._ec2_conn = self._connect_ec2()
         return self._ec2_conn
 
     @property
-    def s3_conn(self):
+    def s3_conn(self) -> Any:
         if not self._s3_conn:
             self._s3_conn = self._connect_s3()
         return self._s3_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_ec2(self):
+    def _connect_ec2(self) -> Any:
         """
         Get a boto ec2 connection object.
         """
         return self._connect_ec2_region(region_name=self.region_name)
 
-    def _connect_ec2_region(self, region_name=None):
+    def _connect_ec2_region(self, region_name: str | None = None) -> Any:
         '''Get an EC2 resource object'''
         return self.session.resource(
             'ec2', region_name=region_name, **self.ec2_cfg)
 
-    def _connect_s3(self):
+    def _connect_s3(self) -> Any:
         '''Get an S3 resource object'''
         return self.session.resource(
             's3', region_name=self.region_name, **self.s3_cfg)

cloudbridge/providers/aws/subservices.py

@@ -6,41 +6,48 @@ 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__)
 
 
 class AWSBucketObjectSubService(BaseBucketObjectSubService):
 
-    def __init__(self, provider, bucket):
+    def __init__(self, provider: CloudProvider, bucket: Bucket) -> None:
         super(AWSBucketObjectSubService, self).__init__(provider, bucket)
 
 
 class AWSGatewaySubService(BaseGatewaySubService):
 
-    def __init__(self, provider, network):
+    def __init__(self, provider: CloudProvider, network: Network) -> None:
         super(AWSGatewaySubService, self).__init__(provider, network)
 
 
 class AWSVMFirewallRuleSubService(BaseVMFirewallRuleSubService):
 
-    def __init__(self, provider, firewall):
+    def __init__(self, provider: CloudProvider,
+                 firewall: VMFirewall) -> None:
         super(AWSVMFirewallRuleSubService, self).__init__(provider, firewall)
 
 
 class AWSFloatingIPSubService(BaseFloatingIPSubService):
 
-    def __init__(self, provider, gateway):
+    def __init__(self, provider: CloudProvider, gateway: Gateway) -> None:
         super(AWSFloatingIPSubService, self).__init__(provider, gateway)
 
 
 class AWSSubnetSubService(BaseSubnetSubService):
 
-    def __init__(self, provider, network):
+    def __init__(self, provider: CloudProvider, network: Network) -> None:
         super(AWSSubnetSubService, self).__init__(provider, network)
 
 
 class AWSDnsRecordSubService(BaseDnsRecordSubService):
 
-    def __init__(self, provider, dns_zone):
+    def __init__(self, provider: CloudProvider, dns_zone: DnsZone) -> None:
         super(AWSDnsRecordSubService, self).__init__(provider, dns_zone)

pyproject.toml

@@ -141,6 +141,18 @@ disallow_untyped_decorators = false
 module = ["cloudbridge.providers.*"]
 ignore_errors = true
 
+# Typed providers (pragmatic tier). Providers wrap untyped cloud SDKs, so we
+# require complete annotations (disallow_untyped_defs etc. from the strict
+# baseline) but turn off warn_return_any and disallow_untyped_calls: every
+# getter reads an attribute off an Any-typed SDK object, and forcing those
+# 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.*"]
+ignore_errors = false
+warn_return_any = false
+disallow_untyped_calls = false
+
 # base/ is held to the FULL strict bar (it orchestrates through the typed
 # interface layer and never touches the cloud SDKs directly), so it falls under
 # the global strict baseline with no override. The providers will get a