chore(code) better use of generic in entity manager (#283)

Author: antechrestos

cloudfoundry_client/v3/apps.py

@@ -7,7 +7,7 @@
     from cloudfoundry_client.client import CloudFoundryClient
 
 
-class AppManager(EntityManager):
+class AppManager(EntityManager[Entity]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(AppManager, self).__init__(target_endpoint, client, "/v3/apps")
 

cloudfoundry_client/v3/buildpacks.py

@@ -6,7 +6,7 @@
     from cloudfoundry_client.client import CloudFoundryClient
 
 
-class BuildpackManager(EntityManager):
+class BuildpackManager(EntityManager[Entity]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(BuildpackManager, self).__init__(target_endpoint, client, "/v3/buildpacks")
 

cloudfoundry_client/v3/domains.py

@@ -19,7 +19,7 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient", **kwargs)
             )
 
 
-class DomainManager(EntityManager):
+class DomainManager(EntityManager[Domain]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(DomainManager, self).__init__(target_endpoint, client, "/v3/domains", Domain)
 

cloudfoundry_client/v3/entities.py

@@ -1,7 +1,7 @@
 import functools
 from collections.abc import Callable
 from json import JSONDecodeError
-from typing import Any, TypeVar, TYPE_CHECKING, Type
+from typing import Any, TypeVar, TYPE_CHECKING, Type, Generic
 from urllib.parse import quote, urlparse
 
 from requests import Response
@@ -111,30 +111,42 @@ def __init__(self, *guids: str):
         self.guids = list(guids)
 
 
-ENTITY_TYPE = TypeVar("ENTITY_TYPE", bound=Entity)
+ENTITY_TYPE = TypeVar("ENTITY_TYPE", bound=Entity, covariant=True)
 
 
-class EntityManager(object):
-    def __init__(self, target_endpoint: str, client: "CloudFoundryClient", entity_uri: str, entity_type: ENTITY_TYPE = Entity):
+class EntityManager(Generic[ENTITY_TYPE]):
+    def __init__(
+            self,
+            target_endpoint: str,
+            client: "CloudFoundryClient",
+            entity_uri: str,
+            entity_type: type[ENTITY_TYPE] = Entity
+    ):
         self.target_endpoint = target_endpoint
         self.entity_uri = entity_uri
         self.client = client
         self.entity_type = entity_type
 
-    def _post(self, url: str, data: dict | None = None, files: Any = None, entity_type: ENTITY_TYPE = None) -> Entity:
-        response = self.client.post(url, json=data, files=files)
-        return self._read_response(response, entity_type)
-
-    def _get(self, url: str, entity_type: ENTITY_TYPE | None = None, **kwargs) -> Entity:
+    def _get(self, url: str, entity_type: type[ENTITY_TYPE] | None = None, **kwargs) -> ENTITY_TYPE:
         url_requested = EntityManager._get_url_with_encoded_params(url, **kwargs)
         response = self.client.get(url_requested)
         return self._read_response(response, entity_type)
 
-    def _put(self, url: str, data: dict, entity_type: ENTITY_TYPE | None = None) -> Entity:
+    def _post(
+            self,
+            url: str,
+            data: dict | None = None,
+            entity_type: type[ENTITY_TYPE] | None = None,
+            files: Any = None
+    ) -> ENTITY_TYPE:
+        response = self.client.post(url, json=data, files=files)
+        return self._read_response(response, entity_type)
+
+    def _put(self, url: str, data: dict, entity_type: type[ENTITY_TYPE] | None = None) -> ENTITY_TYPE:
         response = self.client.put(url, json=data)
         return self._read_response(response, entity_type)
 
-    def _patch(self, url: str, data: dict, entity_type: ENTITY_TYPE | None = None) -> Entity:
+    def _patch(self, url: str, data: dict, entity_type: type[ENTITY_TYPE] | None = None) -> ENTITY_TYPE:
         response = self.client.patch(url, json=data)
         return self._read_response(response, entity_type)
 
@@ -166,21 +178,21 @@ def _extract_job_guid(job_location):
         job_guid = job_url.path.rsplit("/", 1)[-1]
         return job_guid
 
-    def _list(self, requested_path: str, entity_type: ENTITY_TYPE | None = None, **kwargs) -> Pagination[Entity]:
+    def _list(self, requested_path: str, entity_type: type[ENTITY_TYPE] | None = None, **kwargs) -> Pagination[ENTITY_TYPE]:
         url_requested = EntityManager._get_url_with_encoded_params("%s%s" % (self.target_endpoint, requested_path), **kwargs)
         response_json = self._read_response(self.client.get(url_requested), JsonObject)
         return self._pagination(response_json, entity_type)
 
-    def _attempt_to_paginate(self, url_requested: str, entity_type: ENTITY_TYPE | None = None) \
-            -> Pagination[Entity] | Entity:
+    def _attempt_to_paginate(self, url_requested: str, entity_type: type[ENTITY_TYPE] | None = None) \
+            -> Pagination[ENTITY_TYPE] | ENTITY_TYPE:
         response_json = self._read_response(self.client.get(url_requested), JsonObject)
         if "resources" in response_json:
             return self._pagination(response_json, entity_type)
         else:
             return response_json
 
-    def _pagination(self, page: JsonObject, entity_type: ENTITY_TYPE | None = None) -> Pagination[Entity]:
-        def _entity(json_object: JsonObject) -> Entity:
+    def _pagination(self, page: JsonObject, entity_type: type[ENTITY_TYPE] | None = None) -> Pagination[ENTITY_TYPE]:
+        def _entity(json_object: JsonObject) -> ENTITY_TYPE:
             return self._entity(json_object, entity_type)
 
         return Pagination(page,
@@ -200,54 +212,58 @@ def _next_page(self, current_page: JsonObject) -> JsonObject | None:
             return None
         return self._read_response(self.client.get(current_page["pagination"]["next"]["href"]), JsonObject)
 
-    def _create(self, data: dict) -> Entity:
+    def _create(self, data: dict) -> ENTITY_TYPE:
         url = "%s%s" % (self.target_endpoint, self.entity_uri)
         return self._post(url, data=data)
 
-    def _upload_bits(self, resource_id: str, filename: str) -> Entity:
+    def _upload_bits(self, resource_id: str, filename: str) -> ENTITY_TYPE:
         url = "%s%s/%s/upload" % (self.target_endpoint, self.entity_uri, resource_id)
         files = {"bits": (filename, open(filename, "rb"))}
         return self._post(url, files=files)
 
-    def _update(self, resource_id: str, data: dict) -> Entity:
+    def _update(self, resource_id: str, data: dict) -> ENTITY_TYPE:
         url = "%s%s/%s" % (self.target_endpoint, self.entity_uri, resource_id)
         return self._patch(url, data)
 
-    def __iter__(self) -> Pagination[Entity]:
+    def __iter__(self) -> Pagination[ENTITY_TYPE]:
         return self.list()
 
-    def __getitem__(self, entity_guid) -> Entity:
+    def __getitem__(self, entity_guid) -> ENTITY_TYPE:
         return self.get(entity_guid)
 
     def __len__(self):
         return self.len()
 
     def len(self, **kwargs):
         url_requested = EntityManager._get_url_with_encoded_params("%s%s" % (self.target_endpoint, self.entity_uri), **kwargs)
-        response_json = self._read_response(self.client.get(url_requested, JsonObject))
+        response_json = self._read_response(self.client.get(url_requested), JsonObject)
         pagination = response_json.get("pagination")
         if pagination is not None:
             return pagination.get("total_results", 0)
         else:
             return 0
 
-    def list(self, **kwargs) -> Pagination[Entity]:
+    def list(self, **kwargs) -> Pagination[ENTITY_TYPE]:
         return self._list(self.entity_uri, **kwargs)
 
-    def get_first(self, **kwargs) -> Entity | None:
+    def get_first(self, **kwargs) -> ENTITY_TYPE | None:
         kwargs.setdefault("per_page", 1)
         for entity in self._list(self.entity_uri, **kwargs):
             return entity
         return None
 
-    def get(self, entity_id: str, *extra_paths, **kwargs) -> Entity:
+    def get(self, entity_id: str, *extra_paths, **kwargs) -> ENTITY_TYPE:
         if len(extra_paths) == 0:
             requested_path = "%s%s/%s" % (self.target_endpoint, self.entity_uri, entity_id)
         else:
             requested_path = "%s%s/%s/%s" % (self.target_endpoint, self.entity_uri, entity_id, "/".join(extra_paths))
         return self._get(requested_path, **kwargs)
 
-    def _read_response(self, response: Response, entity_type: ENTITY_TYPE | None) -> JsonObject | Entity:
+    def _read_response(
+            self,
+            response: Response,
+            entity_type: type[ENTITY_TYPE] | type[JsonObject] | None
+    ) -> JsonObject | ENTITY_TYPE:
         try:
             result = response.json(object_pairs_hook=JsonObject)
         except JSONDecodeError:
@@ -289,7 +305,7 @@ def _include_resources(self, resource: JsonObject, result: JsonObject) -> None:
     def _get_entity_type(entity_name: str) -> Type[ENTITY_TYPE]:
         return Entity
 
-    def _entity(self, result: JsonObject, entity_type: ENTITY_TYPE | None) -> JsonObject | Entity:
+    def _entity(self, result: JsonObject, entity_type: type[ENTITY_TYPE] | None) -> JsonObject | ENTITY_TYPE:
         if "guid" in result or ("links" in result and "job" in result["links"]):
             return (entity_type or self.entity_type)(self.target_endpoint, self.client, **result)
         else:
@@ -301,7 +317,7 @@ def _append_encoded_parameter(parameters: list[str], args: tuple[str, Any]) -> l
             parameter_name, parameter_value = args[0], args[1]
             if isinstance(parameter_value, (list, tuple)):
                 parameters.append("%s=%s" % (parameter_name, quote(",".join(parameter_value))))
-            elif isinstance(parameter_value, (dict)) and parameter_name == "fields":
+            elif isinstance(parameter_value, dict) and parameter_name == "fields":
                 for resource, key in parameter_value.items():
                     parameters.append("%s[%s]=%s" % (parameter_name, resource, ",".join(key)))
             else:

cloudfoundry_client/v3/feature_flags.py

@@ -6,7 +6,7 @@
     from cloudfoundry_client.client import CloudFoundryClient
 
 
-class FeatureFlagManager(EntityManager):
+class FeatureFlagManager(EntityManager[Entity]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(FeatureFlagManager, self).__init__(target_endpoint, client, "/v3/feature_flags")
 

cloudfoundry_client/v3/isolation_segments.py

@@ -6,7 +6,7 @@
     from cloudfoundry_client.client import CloudFoundryClient
 
 
-class IsolationSegmentManager(EntityManager):
+class IsolationSegmentManager(EntityManager[Entity]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(IsolationSegmentManager, self).__init__(target_endpoint, client, "/v3/isolation_segments")
 

cloudfoundry_client/v3/jobs.py

@@ -13,7 +13,7 @@ class JobTimeout(Exception):
     pass
 
 
-class JobManager(EntityManager):
+class JobManager(EntityManager[Entity]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(JobManager, self).__init__(target_endpoint, client, "/v3/jobs")
 

cloudfoundry_client/v3/organization_quotas.py

@@ -33,7 +33,7 @@ class DomainsQuota:
     total_domains: int
 
 
-class OrganizationQuotaManager(EntityManager):
+class OrganizationQuotaManager(EntityManager[Entity]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super().__init__(target_endpoint, client, "/v3/organization_quotas")
 

cloudfoundry_client/v3/organizations.py

@@ -6,7 +6,7 @@
     from cloudfoundry_client.client import CloudFoundryClient
 
 
-class OrganizationManager(EntityManager):
+class OrganizationManager(EntityManager[Entity]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(OrganizationManager, self).__init__(target_endpoint, client, "/v3/organizations")
 

cloudfoundry_client/v3/processes.py

@@ -1,11 +1,11 @@
 from typing import TYPE_CHECKING
 
-from cloudfoundry_client.v3.entities import EntityManager
+from cloudfoundry_client.v3.entities import EntityManager, Entity
 
 if TYPE_CHECKING:
     from cloudfoundry_client.client import CloudFoundryClient
 
 
-class ProcessManager(EntityManager):
+class ProcessManager(EntityManager[Entity]):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(ProcessManager, self).__init__(target_endpoint, client, "/v3/processes")