From 65c72c02843c37223d638a1e435c7967990257ed Mon Sep 17 00:00:00 2001 From: Benjamin Einaudi Date: Thu, 8 Jan 2026 18:35:50 +0100 Subject: [PATCH] chore(code) better use of generic in entity manager --- cloudfoundry_client/v3/apps.py | 2 +- cloudfoundry_client/v3/buildpacks.py | 2 +- cloudfoundry_client/v3/domains.py | 2 +- cloudfoundry_client/v3/entities.py | 72 +++++++++++-------- cloudfoundry_client/v3/feature_flags.py | 2 +- cloudfoundry_client/v3/isolation_segments.py | 2 +- cloudfoundry_client/v3/jobs.py | 2 +- cloudfoundry_client/v3/organization_quotas.py | 2 +- cloudfoundry_client/v3/organizations.py | 2 +- cloudfoundry_client/v3/processes.py | 4 +- cloudfoundry_client/v3/roles.py | 4 +- cloudfoundry_client/v3/security_groups.py | 4 +- cloudfoundry_client/v3/service_brokers.py | 2 +- .../v3/service_credential_bindings.py | 2 +- cloudfoundry_client/v3/service_instances.py | 2 +- cloudfoundry_client/v3/service_offerings.py | 2 +- cloudfoundry_client/v3/service_plans.py | 2 +- cloudfoundry_client/v3/spaces.py | 2 +- cloudfoundry_client/v3/tasks.py | 2 +- cloudfoundry_client/v3/users.py | 2 +- 20 files changed, 66 insertions(+), 50 deletions(-) diff --git a/cloudfoundry_client/v3/apps.py b/cloudfoundry_client/v3/apps.py index 01746d4..46b9dce 100644 --- a/cloudfoundry_client/v3/apps.py +++ b/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") diff --git a/cloudfoundry_client/v3/buildpacks.py b/cloudfoundry_client/v3/buildpacks.py index a8b3e68..aa88ab8 100644 --- a/cloudfoundry_client/v3/buildpacks.py +++ b/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") diff --git a/cloudfoundry_client/v3/domains.py b/cloudfoundry_client/v3/domains.py index aa752a0..a26bea9 100644 --- a/cloudfoundry_client/v3/domains.py +++ b/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) diff --git a/cloudfoundry_client/v3/entities.py b/cloudfoundry_client/v3/entities.py index be2b257..8a7ba6f 100644 --- a/cloudfoundry_client/v3/entities.py +++ b/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,23 +212,23 @@ 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): @@ -224,30 +236,34 @@ def __len__(self): 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: diff --git a/cloudfoundry_client/v3/feature_flags.py b/cloudfoundry_client/v3/feature_flags.py index c75517d..55717fb 100644 --- a/cloudfoundry_client/v3/feature_flags.py +++ b/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") diff --git a/cloudfoundry_client/v3/isolation_segments.py b/cloudfoundry_client/v3/isolation_segments.py index 772e4cc..eafd874 100644 --- a/cloudfoundry_client/v3/isolation_segments.py +++ b/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") diff --git a/cloudfoundry_client/v3/jobs.py b/cloudfoundry_client/v3/jobs.py index 80bd510..9988886 100644 --- a/cloudfoundry_client/v3/jobs.py +++ b/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") diff --git a/cloudfoundry_client/v3/organization_quotas.py b/cloudfoundry_client/v3/organization_quotas.py index 51f4b56..9304630 100644 --- a/cloudfoundry_client/v3/organization_quotas.py +++ b/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") diff --git a/cloudfoundry_client/v3/organizations.py b/cloudfoundry_client/v3/organizations.py index 18562ee..7dbacc4 100644 --- a/cloudfoundry_client/v3/organizations.py +++ b/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") diff --git a/cloudfoundry_client/v3/processes.py b/cloudfoundry_client/v3/processes.py index cb632f3..34af872 100644 --- a/cloudfoundry_client/v3/processes.py +++ b/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") diff --git a/cloudfoundry_client/v3/roles.py b/cloudfoundry_client/v3/roles.py index d7f00c2..11e8eba 100644 --- a/cloudfoundry_client/v3/roles.py +++ b/cloudfoundry_client/v3/roles.py @@ -1,12 +1,12 @@ 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 RoleManager(EntityManager): +class RoleManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(RoleManager, self).__init__(target_endpoint, client, "/v3/roles") diff --git a/cloudfoundry_client/v3/security_groups.py b/cloudfoundry_client/v3/security_groups.py index dcbc459..6a21459 100644 --- a/cloudfoundry_client/v3/security_groups.py +++ b/cloudfoundry_client/v3/security_groups.py @@ -35,7 +35,7 @@ class GloballyEnabled: staging: bool | None = None -class SecurityGroupManager(EntityManager): +class SecurityGroupManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(SecurityGroupManager, self).__init__(target_endpoint, client, "/v3/security_groups") @@ -82,7 +82,7 @@ def unbind_staging_security_group_from_space(self, security_group_id: str, space def _bind_spaces(self, security_group_id: str, space_guids: ToManyRelationship, relationship: str) \ -> ToManyRelationship: url = "%s%s/%s/relationships/%s" % (self.target_endpoint, self.entity_uri, security_group_id, relationship) - return ToManyRelationship.from_json_object(super()._post(url, space_guids)) + return ToManyRelationship.from_json_object(super()._post(url, data=space_guids)) def _unbind_space(self, security_group_id: str, space_guid: ToOneRelationship, relationship: str): url = "%s%s/%s/relationships/%s/%s" \ diff --git a/cloudfoundry_client/v3/service_brokers.py b/cloudfoundry_client/v3/service_brokers.py index c507f18..14e15e7 100644 --- a/cloudfoundry_client/v3/service_brokers.py +++ b/cloudfoundry_client/v3/service_brokers.py @@ -6,7 +6,7 @@ from cloudfoundry_client.client import CloudFoundryClient -class ServiceBrokerManager(EntityManager): +class ServiceBrokerManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServiceBrokerManager, self).__init__(target_endpoint, client, "/v3/service_brokers") diff --git a/cloudfoundry_client/v3/service_credential_bindings.py b/cloudfoundry_client/v3/service_credential_bindings.py index ff41c00..b1f53d5 100644 --- a/cloudfoundry_client/v3/service_credential_bindings.py +++ b/cloudfoundry_client/v3/service_credential_bindings.py @@ -6,7 +6,7 @@ from cloudfoundry_client.client import CloudFoundryClient -class ServiceCredentialBindingManager(EntityManager): +class ServiceCredentialBindingManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServiceCredentialBindingManager, self).__init__(target_endpoint, client, "/v3/service_credential_bindings") diff --git a/cloudfoundry_client/v3/service_instances.py b/cloudfoundry_client/v3/service_instances.py index 189870e..0d3d5d4 100644 --- a/cloudfoundry_client/v3/service_instances.py +++ b/cloudfoundry_client/v3/service_instances.py @@ -7,7 +7,7 @@ from cloudfoundry_client.client import CloudFoundryClient -class ServiceInstanceManager(EntityManager): +class ServiceInstanceManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServiceInstanceManager, self).__init__(target_endpoint, client, "/v3/service_instances") diff --git a/cloudfoundry_client/v3/service_offerings.py b/cloudfoundry_client/v3/service_offerings.py index bc067aa..76ebc7a 100644 --- a/cloudfoundry_client/v3/service_offerings.py +++ b/cloudfoundry_client/v3/service_offerings.py @@ -6,7 +6,7 @@ from cloudfoundry_client.client import CloudFoundryClient -class ServiceOfferingsManager(EntityManager): +class ServiceOfferingsManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServiceOfferingsManager, self).__init__(target_endpoint, client, "/v3/service_offerings") diff --git a/cloudfoundry_client/v3/service_plans.py b/cloudfoundry_client/v3/service_plans.py index 0c6f56a..3f4f6fc 100644 --- a/cloudfoundry_client/v3/service_plans.py +++ b/cloudfoundry_client/v3/service_plans.py @@ -6,7 +6,7 @@ from cloudfoundry_client.client import CloudFoundryClient -class ServicePlanManager(EntityManager): +class ServicePlanManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServicePlanManager, self).__init__(target_endpoint, client, "/v3/service_plans") diff --git a/cloudfoundry_client/v3/spaces.py b/cloudfoundry_client/v3/spaces.py index fd78d68..ec5751e 100644 --- a/cloudfoundry_client/v3/spaces.py +++ b/cloudfoundry_client/v3/spaces.py @@ -6,7 +6,7 @@ from cloudfoundry_client.client import CloudFoundryClient -class SpaceManager(EntityManager): +class SpaceManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(SpaceManager, self).__init__(target_endpoint, client, "/v3/spaces") diff --git a/cloudfoundry_client/v3/tasks.py b/cloudfoundry_client/v3/tasks.py index c5b23d5..0cda445 100644 --- a/cloudfoundry_client/v3/tasks.py +++ b/cloudfoundry_client/v3/tasks.py @@ -6,7 +6,7 @@ from cloudfoundry_client.client import CloudFoundryClient -class TaskManager(EntityManager): +class TaskManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(TaskManager, self).__init__(target_endpoint, client, "/v3/tasks") diff --git a/cloudfoundry_client/v3/users.py b/cloudfoundry_client/v3/users.py index 431bd19..cadc48f 100644 --- a/cloudfoundry_client/v3/users.py +++ b/cloudfoundry_client/v3/users.py @@ -6,7 +6,7 @@ from cloudfoundry_client.client import CloudFoundryClient -class UserManager(EntityManager): +class UserManager(EntityManager[Entity]): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(UserManager, self).__init__(target_endpoint, client, "/v3/users")