fixes(api) allow no v2 api

Allow client without no V2 api

Closes #220

Author: antechrestos

.github/workflows/pythonpackage.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       max-parallel: 1
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
 
     steps:
     - uses: actions/checkout@v4

main/cloudfoundry_client/client.py

@@ -52,18 +52,27 @@
 class Info:
     def __init__(
         self,
-        api_v2_version: str,
+        api_v2_url: str,
+        api_v3_url: str,
         authorization_endpoint: str,
         api_endpoint: str,
         doppler_endpoint: Optional[str],
         log_stream_endpoint: Optional[str],
     ):
-        self.api_v2_version = api_v2_version
+        self._api_v2_url = api_v2_url
+        self._api_v3_url = api_v3_url
         self.authorization_endpoint = authorization_endpoint
         self.api_endpoint = api_endpoint
         self.doppler_endpoint = doppler_endpoint
         self.log_stream_endpoint = log_stream_endpoint
 
+    @property
+    def api_v2_url(self) -> Optional[str]:
+        return self._api_v2_url
+
+    @property
+    def api_v3_url(self) -> Optional[str]:
+        return self._api_v3_url
 
 class NetworkingV1External(object):
     def __init__(self, target_endpoint: str, credential_manager: "CloudFoundryClient"):
@@ -83,18 +92,18 @@ def __init__(self, target_endpoint: str, credential_manager: "CloudFoundryClient
         self.service_plans = ServicePlanManagerV2(target_endpoint, credential_manager)
         # Default implementations
         self.event = EventManager(target_endpoint, credential_manager)
-        self.organizations = EntityManagerV2(target_endpoint, credential_manager, "/v2/organizations")
-        self.private_domains = EntityManagerV2(target_endpoint, credential_manager, "/v2/private_domains")
+        self.organizations = EntityManagerV2(target_endpoint, credential_manager, "/organizations")
+        self.private_domains = EntityManagerV2(target_endpoint, credential_manager, "/private_domains")
         self.routes = RouteManager(target_endpoint, credential_manager)
-        self.services = EntityManagerV2(target_endpoint, credential_manager, "/v2/services")
-        self.shared_domains = EntityManagerV2(target_endpoint, credential_manager, "/v2/shared_domains")
+        self.services = EntityManagerV2(target_endpoint, credential_manager, "/services")
+        self.shared_domains = EntityManagerV2(target_endpoint, credential_manager, "/shared_domains")
         self.spaces = SpaceManagerV2(target_endpoint, credential_manager)
-        self.stacks = EntityManagerV2(target_endpoint, credential_manager, "/v2/stacks")
+        self.stacks = EntityManagerV2(target_endpoint, credential_manager, "/stacks")
         self.user_provided_service_instances = EntityManagerV2(
-            target_endpoint, credential_manager, "/v2/user_provided_service_instances"
+            target_endpoint, credential_manager, "/user_provided_service_instances"
         )
-        self.security_groups = EntityManagerV2(target_endpoint, credential_manager, "/v2/security_groups")
-        self.users = EntityManagerV2(target_endpoint, credential_manager, "/v2/users")
+        self.security_groups = EntityManagerV2(target_endpoint, credential_manager, "/security_groups")
+        self.users = EntityManagerV2(target_endpoint, credential_manager, "/users")
         # Resources implementation used by push operation
         self.resources = ResourceManager(target_endpoint, credential_manager)
 
@@ -146,8 +155,6 @@ def __init__(self, target_endpoint: str, client_id: str = "cf", client_secret: s
         self.login_hint = kwargs.get("login_hint")
         target_endpoint_trimmed = target_endpoint.rstrip("/")
         info = self._get_info(target_endpoint_trimmed, proxy, verify=verify)
-        if not info.api_v2_version.startswith("2."):
-            raise AssertionError("Only version 2 is supported for now. Found %s" % info.api_v2_version)
         service_information = ServiceInformation(
             None, "%s/oauth/token" % info.authorization_endpoint, client_id, client_secret, [], verify
         )
@@ -156,8 +163,16 @@ def __init__(self, target_endpoint: str, client_id: str = "cf", client_secret: s
             proxies=proxy,
             user_agent=kwargs.get("user_agent", "cf-python-client")
         )
-        self.v2 = V2(target_endpoint_trimmed, self)
-        self.v3 = V3(target_endpoint_trimmed, self)
+        self._v2 = (
+            V2(info.api_v2_url, self)
+            if info.api_v2_url is not None
+            else None
+        )
+        self._v3 = (
+            V3(info.api_v3_url, self)
+            if info.api_v3_url is not None
+            else None
+        )
         self._doppler = (
             DopplerClient(
                 info.doppler_endpoint,
@@ -181,6 +196,18 @@ def __init__(self, target_endpoint: str, client_id: str = "cf", client_secret: s
         self.networking_v1_external = NetworkingV1External(target_endpoint_trimmed, self)
         self.info = info
 
+    @property
+    def v2(self) -> V2:
+        if self._v2 is None:
+            raise NotImplementedError("No V2 endpoint for this instance")
+        return self._v2
+
+    @property
+    def v3(self) -> V3:
+        if self._v3 is None:
+            raise NotImplementedError("No V3 endpoint for this instance")
+        return self._v3
+
     @property
     def doppler(self) -> DopplerClient:
         if self._doppler is None:
@@ -194,7 +221,6 @@ def rlpgateway(self):
         if self._rlpgateway is None:
             raise NotImplementedError("No RLP gateway endpoint for this instance")
         else:
-
             return self._rlpgateway
 
     def _get_info(self, target_endpoint: str, proxy: Optional[dict] = None, verify: bool = True) -> Info:
@@ -206,8 +232,11 @@ def _get_info(self, target_endpoint: str, proxy: Optional[dict] = None, verify:
         root_links = root_info["links"]
         logging = root_links.get("logging")
         log_stream = root_links.get("log_stream")
+        cloud_controller_v2 = root_links.get("cloud_controller_v2")
+        cloud_controller_v3 = root_links.get("cloud_controller_v3")
         return Info(
-            root_links["cloud_controller_v2"]["meta"]["version"],
+            cloud_controller_v2["href"] if cloud_controller_v2 is not None else None,
+            cloud_controller_v3["href"] if cloud_controller_v3 is not None else None,
             self._resolve_login_endpoint(root_links),
             target_endpoint,
             logging.get("href") if logging is not None else None,

main/cloudfoundry_client/common_objects.py

@@ -1,5 +1,5 @@
 import json
-from typing import Callable, TypeVar, Generic, List, Union
+from typing import Callable, TypeVar, Generic, List, Union, Optional
 
 
 class Request(dict):
@@ -21,7 +21,7 @@ def __init__(self, *args, **kwargs):
 class Pagination(Generic[ENTITY]):
     def __init__(self, first_page: JsonObject,
                  total_result: int,
-                 next_page_loader: Callable[[JsonObject], Union[None, JsonObject]],
+                 next_page_loader: Callable[[JsonObject], Optional[JsonObject]],
                  resources_accessor: Callable[[JsonObject], List[JsonObject]],
                  instance_creator: Callable[[JsonObject], ENTITY]):
         self._first_page = first_page

main/cloudfoundry_client/v2/apps.py

@@ -77,7 +77,7 @@ class AppManager(EntityManager):
 
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(AppManager, self).__init__(
-            target_endpoint, client, "/v2/apps", lambda pairs: Application(target_endpoint, client, pairs)
+            target_endpoint, client, "/apps", lambda pairs: Application(target_endpoint, client, pairs)
         )
 
     def get_stats(self, application_guid: str) -> Dict[str, JsonObject]:

main/cloudfoundry_client/v2/buildpacks.py

@@ -8,7 +8,7 @@
 
 class BuildpackManager(EntityManager):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
-        super(BuildpackManager, self).__init__(target_endpoint, client, "/v2/buildpacks")
+        super(BuildpackManager, self).__init__(target_endpoint, client, "/buildpacks")
 
     def update(self, buildpack_guid: str, parameters: dict) -> Entity:
         return super(BuildpackManager, self)._update(buildpack_guid, parameters)

main/cloudfoundry_client/v2/entities.py

@@ -68,7 +68,7 @@ def _list(self, requested_path: str, entity_builder: Optional[EntityBuilder] = N
                           lambda page: page["resources"],
                           lambda json_object: current_builder(list(json_object.items())))
 
-    def _next_page(self, current_page: JsonObject) -> Union[None, JsonObject]:
+    def _next_page(self, current_page: JsonObject) -> Optional[JsonObject]:
         next_url = current_page.get("next_url")
         if next_url is None:
             return None

main/cloudfoundry_client/v2/events.py

@@ -8,7 +8,7 @@
 
 class EventManager(EntityManager):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
-        super(EventManager, self).__init__(target_endpoint, client, "/v2/events")
+        super(EventManager, self).__init__(target_endpoint, client, "/events")
 
     def list_by_type(self, event_type: str) -> Generator[Entity, None, None]:
         return self._list(self.entity_uri, type=event_type)

main/cloudfoundry_client/v2/jobs.py

@@ -12,4 +12,4 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         self.client = client
 
     def get(self, job_guid: str) -> JsonObject:
-        return self.client.get("%s/v2/jobs/%s" % (self.target_endpoint, job_guid)).json(object_pairs_hook=JsonObject)
+        return self.client.get("%s/jobs/%s" % (self.target_endpoint, job_guid)).json(object_pairs_hook=JsonObject)

main/cloudfoundry_client/v2/resources.py

@@ -12,5 +12,5 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         self.client = client
 
     def match(self, items: List[dict]) -> List[JsonObject]:
-        response = self.client.put("%s/v2/resource_match" % self.client.info.api_endpoint, json=items)
+        response = self.client.put("%s/resource_match" % self.client.info.api_endpoint, json=items)
         return response.json(object_pairs_hook=JsonObject)

main/cloudfoundry_client/v2/routes.py

@@ -8,7 +8,7 @@
 
 class RouteManager(EntityManager):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
-        super(RouteManager, self).__init__(target_endpoint, client, "/v2/routes")
+        super(RouteManager, self).__init__(target_endpoint, client, "/routes")
 
     def create_tcp_route(self, domain_guid: str, space_guid: str, port: Optional[int] = None) -> Entity:
         request = self._request(domain_guid=domain_guid, space_guid=space_guid)