chore(code) remove Optional (#272)

Author: antechrestos

.github/workflows/pythonpackage.yml

@@ -34,8 +34,8 @@ jobs:
     - name: Install project
       run: poetry install --no-interaction
 
-    - name: Launch tests
-      run: poetry run pytest
-
-    - name: Launch Linting
+    - name: Linting
       run: poetry run flake8 --show-source --statistics
+
+    - name: Tests
+      run: poetry run pytest

cloudfoundry_client/client.py

@@ -2,7 +2,6 @@
 from pathlib import Path
 import json
 from http import HTTPStatus
-from typing import Optional
 
 import requests
 from oauth2_client.credentials_manager import CredentialManager, ServiceInformation
@@ -56,8 +55,8 @@ def __init__(
         api_v3_url: str,
         authorization_endpoint: str,
         api_endpoint: str,
-        doppler_endpoint: Optional[str],
-        log_stream_endpoint: Optional[str],
+        doppler_endpoint: str | None,
+        log_stream_endpoint: str | None,
     ):
         self._api_v2_url = api_v2_url
         self._api_v3_url = api_v3_url
@@ -67,11 +66,11 @@ def __init__(
         self.log_stream_endpoint = log_stream_endpoint
 
     @property
-    def api_v2_url(self) -> Optional[str]:
+    def api_v2_url(self) -> str | None:
         return self._api_v2_url
 
     @property
-    def api_v3_url(self) -> Optional[str]:
+    def api_v3_url(self) -> str | None:
         return self._api_v3_url
 
 
@@ -226,7 +225,7 @@ def rlpgateway(self):
         else:
             return self._rlpgateway
 
-    def _get_info(self, target_endpoint: str, proxy: Optional[dict] = None, verify: bool = True) -> Info:
+    def _get_info(self, target_endpoint: str, proxy: dict | None = None, verify: bool = True) -> Info:
         root_response = CloudFoundryClient._check_response(
             requests.get("%s/" % target_endpoint, proxies=proxy if proxy is not None else dict(http="", https=""), verify=verify)
         )
@@ -247,7 +246,7 @@ def _get_info(self, target_endpoint: str, proxy: Optional[dict] = None, verify:
         )
 
     @staticmethod
-    def build_from_cf_config(config_path: Optional[str] = None, **kwargs) -> 'CloudFoundryClient':
+    def build_from_cf_config(config_path: str | None = None, **kwargs) -> 'CloudFoundryClient':
         config = Path(config_path) if config_path else Path.home() / '.cf/config.json'
         try:
             with open(config) as f:
@@ -318,7 +317,7 @@ def _grant_client_credentials_request(self) -> dict:
             client_secret=self.service_information.client_secret,
         )
 
-    def get(self, url: str, params: Optional[dict] = None, **kwargs) -> Response:
+    def get(self, url: str, params: dict | None = None, **kwargs) -> Response:
         response = super(CloudFoundryClient, self).get(url, params, **kwargs)
         CloudFoundryClient._log_request("GET", url, response)
         return CloudFoundryClient._check_response(response)

cloudfoundry_client/common_objects.py

@@ -1,5 +1,5 @@
 import json
-from typing import Callable, TypeVar, Generic, List, Optional
+from typing import Callable, TypeVar, Generic, List
 
 
 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], Optional[JsonObject]],
+                 next_page_loader: Callable[[JsonObject], JsonObject | None],
                  resources_accessor: Callable[[JsonObject], List[JsonObject]],
                  instance_creator: Callable[[JsonObject], ENTITY]):
         self._first_page = first_page

cloudfoundry_client/doppler/websocket_envelope_reader.py

@@ -1,5 +1,5 @@
 import ssl
-from typing import Callable, Optional, Tuple
+from typing import Callable, Tuple
 
 import websocket
 
@@ -10,9 +10,9 @@ def __init__(
         url,
         access_token_provider: Callable[[], str],
         verify_ssl: bool = True,
-        proxy_host: Optional[str] = None,
-        proxy_port: Optional[int] = None,
-        proxy_auth: Optional[Tuple[str, str]] = None,
+        proxy_host: str | None = None,
+        proxy_port: int | None = None,
+        proxy_auth: Tuple[str, str] | None = None,
     ):
         if not verify_ssl:
             self._ws = websocket.WebSocket(sslopt=dict(cert_reqs=ssl.CERT_NONE))

cloudfoundry_client/networking/entities.py

@@ -1,6 +1,6 @@
 import logging
 from functools import reduce
-from typing import Callable, List, Tuple, Any, Optional, Generator, TYPE_CHECKING
+from typing import Callable, List, Tuple, Any, Generator, TYPE_CHECKING
 from urllib.parse import quote
 
 from requests import Response
@@ -40,7 +40,7 @@ class EntityManager(object):
     list_multi_parameters = ["order-by"]
 
     def __init__(
-        self, target_endpoint: str, client: "CloudFoundryClient", entity_uri: str, entity_builder: Optional[EntityBuilder] = None
+        self, target_endpoint: str, client: "CloudFoundryClient", entity_uri: str, entity_builder: EntityBuilder | None = None
     ):
         self.target_endpoint = target_endpoint
         self.entity_uri = entity_uri
@@ -50,7 +50,7 @@ def __init__(
         )
 
     def _list(
-        self, requested_path: str, entity_builder: Optional[EntityBuilder] = None, **kwargs
+        self, requested_path: str, entity_builder: EntityBuilder | None = None, **kwargs
     ) -> Generator[Entity, None, None]:
         url_requested = self._get_url_filtered("%s%s" % (self.target_endpoint, requested_path), **kwargs)
         response = self.client.get(url_requested)
@@ -68,7 +68,7 @@ def _remove(self, resource_id: str, **kwargs):
         url = "%s%s/%s" % (self.target_endpoint, self.entity_uri, resource_id)
         self._delete(url, **kwargs)
 
-    def _post(self, url: str, data: Optional[dict] = None, **kwargs):
+    def _post(self, url: str, data: dict | None = None, **kwargs):
         response = self.client.post(url, json=data, **kwargs)
         _logger.debug("POST - %s - %s", url, response.text)
         return self._read_response(response)
@@ -83,13 +83,13 @@ def __iter__(self) -> Generator[Entity, None, None]:
     def list(self, **kwargs) -> Generator[Entity, None, None]:
         return self._list(self.entity_uri, **kwargs)
 
-    def get_first(self, **kwargs) -> Optional[Entity]:
+    def get_first(self, **kwargs) -> Entity | None:
         kwargs.setdefault("results-per-page", 1)
         for entity in self._list(self.entity_uri, **kwargs):
             return entity
         return None
 
-    def _read_response(self, response: Response, other_entity_builder: Optional[EntityBuilder] = None):
+    def _read_response(self, response: Response, other_entity_builder: EntityBuilder | None = None):
         entity_builder = self._get_entity_builder(other_entity_builder)
         result = response.json(object_pairs_hook=JsonObject)
         return entity_builder(list(result.items()))
@@ -98,7 +98,7 @@ def _read_response(self, response: Response, other_entity_builder: Optional[Enti
     def _request(**mandatory_parameters) -> Request:
         return Request(**mandatory_parameters)
 
-    def _get_entity_builder(self, entity_builder: Optional[EntityBuilder]) -> EntityBuilder:
+    def _get_entity_builder(self, entity_builder: EntityBuilder | None) -> EntityBuilder:
         if entity_builder is None:
             return self.entity_builder
         else:

cloudfoundry_client/operations/push/file_helper.py

@@ -2,12 +2,12 @@
 import os
 import stat
 import zipfile
-from typing import Callable, Optional, Generator, Tuple, List
+from typing import Callable, Generator, Tuple, List
 
 
 class FileHelper(object):
     @staticmethod
-    def zip(file_location: str, directory_path: str, accept: Optional[Callable[[str], bool]] = None):
+    def zip(file_location: str, directory_path: str, accept: Callable[[str], bool] | None = None):
         with zipfile.ZipFile(file_location, "w", zipfile.ZIP_DEFLATED) as archive_out:
             for dir_path, file_names in FileHelper.walk(directory_path):
                 dir_full_location = os.path.join(directory_path, dir_path)

cloudfoundry_client/operations/push/push.py

@@ -5,7 +5,7 @@
 import shutil
 import tempfile
 import time
-from typing import Tuple, List, Dict, Optional
+from typing import Tuple, List, Dict
 
 from cloudfoundry_client.client import CloudFoundryClient
 from cloudfoundry_client.operations.push.cf_ignore import CfIgnore
@@ -96,7 +96,7 @@ def _build_request_from_manifest(self, app_manifest: dict) -> dict:
         return request
 
     @staticmethod
-    def _merge_environment(app: Optional[Entity], app_manifest: dict) -> dict:
+    def _merge_environment(app: Entity | None, app_manifest: dict) -> dict:
         environment = dict()
         if app is not None and "environment_json" in app["entity"]:
             environment.update(app["entity"]["environment_json"])

cloudfoundry_client/v2/apps.py

@@ -3,7 +3,7 @@
 import os
 from http import HTTPStatus
 from time import sleep
-from typing import Optional, Dict, TYPE_CHECKING
+from typing import Dict, TYPE_CHECKING
 
 from cloudfoundry_client.doppler.client import EnvelopeStream
 from cloudfoundry_client.errors import InvalidStatusCode
@@ -107,9 +107,9 @@ def list_service_bindings(self, application_guid: str, **kwargs) -> Pagination[E
     def start(
         self,
         application_guid: str,
-        check_time: Optional[float] = 0.5,
-        timeout: Optional[float] = 300.0,
-        asynchronous: Optional[bool] = False,
+        check_time: float | None = 0.5,
+        timeout: float | None = 300.0,
+        asynchronous: bool | None = False,
     ) -> Application:
         result = super(AppManager, self)._update(application_guid, dict(state="STARTED"))
         if asynchronous:
@@ -122,9 +122,9 @@ def start(
     def stop(
         self,
         application_guid: str,
-        check_time: Optional[float] = 0.5,
-        timeout: Optional[float] = 500.0,
-        asynchronous: Optional[bool] = False,
+        check_time: float | None = 0.5,
+        timeout: float | None = 500.0,
+        asynchronous: bool | None = False,
     ) -> Application:
         result = super(AppManager, self)._update(application_guid, dict(state="STOPPED"))
         if asynchronous:
@@ -152,7 +152,7 @@ def update(self, application_guid: str, **kwargs) -> Application:
     def remove(self, application_guid: str):
         super(AppManager, self)._remove(application_guid)
 
-    def upload(self, application_guid: str, resources, application: str, asynchronous: Optional[bool] = False):
+    def upload(self, application_guid: str, resources, application: str, asynchronous: bool | None = False):
         application_size = os.path.getsize(application)
         with open(application, "rb") as binary_file:
             return self.client.put(

cloudfoundry_client/v2/entities.py

@@ -1,5 +1,5 @@
 from functools import partial, reduce
-from typing import Callable, List, Tuple, Any, Optional, TYPE_CHECKING
+from typing import Callable, List, Tuple, Any, TYPE_CHECKING
 from urllib.parse import quote
 from requests import Response
 
@@ -50,7 +50,7 @@ class EntityManager(object):
     timestamp_parameters = ["timestamp"]
 
     def __init__(
-        self, target_endpoint: str, client: "CloudFoundryClient", entity_uri: str, entity_builder: Optional[EntityBuilder] = None
+        self, target_endpoint: str, client: "CloudFoundryClient", entity_uri: str, entity_builder: EntityBuilder | None = None
     ):
         self.target_endpoint = target_endpoint
         self.entity_uri = entity_uri
@@ -59,7 +59,7 @@ def __init__(
             entity_builder if entity_builder is not None else lambda pairs: Entity(target_endpoint, client, pairs)
         )
 
-    def _list(self, requested_path: str, entity_builder: Optional[EntityBuilder] = None, **kwargs) -> Pagination[Entity]:
+    def _list(self, requested_path: str, entity_builder: EntityBuilder | None = None, **kwargs) -> Pagination[Entity]:
         url_requested = self._get_url_filtered("%s%s" % (self.target_endpoint, requested_path), **kwargs)
         current_builder = self._get_entity_builder(entity_builder)
         response_json = self._read_response(self.client.get(url_requested), JsonObject)
@@ -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) -> Optional[JsonObject]:
+    def _next_page(self, current_page: JsonObject) -> JsonObject | None:
         next_url = current_page.get("next_url")
         if next_url is None:
             return None
@@ -87,16 +87,16 @@ def _remove(self, resource_id: str, **kwargs):
         url = "%s%s/%s" % (self.target_endpoint, self.entity_uri, resource_id)
         self._delete(url, **kwargs)
 
-    def _get(self, requested_path: str, entity_builder: Optional[EntityBuilder] = None) -> Entity:
+    def _get(self, requested_path: str, entity_builder: EntityBuilder | None = None) -> Entity:
         url = "%s%s" % (self.target_endpoint, requested_path)
         response = self.client.get(url)
         return self._read_response(response, entity_builder)
 
-    def _post(self, url: str, data: Optional[dict] = None, **kwargs):
+    def _post(self, url: str, data: dict | None = None, **kwargs):
         response = self.client.post(url, json=data, **kwargs)
         return self._read_response(response)
 
-    def _put(self, url: str, data: Optional[dict] = None, **kwargs):
+    def _put(self, url: str, data: dict | None = None, **kwargs):
         response = self.client.put(url, json=data, **kwargs)
         return self._read_response(response)
 
@@ -112,7 +112,7 @@ def __getitem__(self, entity_guid) -> Entity:
     def list(self, **kwargs) -> Pagination[Entity]:
         return self._list(self.entity_uri, **kwargs)
 
-    def get_first(self, **kwargs) -> Optional[Entity]:
+    def get_first(self, **kwargs) -> Entity | None:
         kwargs.setdefault("results-per-page", 1)
         for entity in self._list(self.entity_uri, **kwargs):
             return entity
@@ -125,7 +125,7 @@ def get(self, entity_id: str, *extra_paths) -> Entity:
             requested_path = "%s/%s/%s" % (self.entity_uri, entity_id, "/".join(extra_paths))
         return self._get(requested_path)
 
-    def _read_response(self, response: Response, other_entity_builder: Optional[EntityBuilder] = None):
+    def _read_response(self, response: Response, other_entity_builder: EntityBuilder | None = None):
         entity_builder = self._get_entity_builder(other_entity_builder)
         result = response.json(object_pairs_hook=JsonObject)
         return entity_builder(list(result.items()))
@@ -134,7 +134,7 @@ def _read_response(self, response: Response, other_entity_builder: Optional[Enti
     def _request(**mandatory_parameters) -> Request:
         return Request(**mandatory_parameters)
 
-    def _get_entity_builder(self, entity_builder: Optional[EntityBuilder]) -> EntityBuilder:
+    def _get_entity_builder(self, entity_builder: EntityBuilder | None) -> EntityBuilder:
         if entity_builder is None:
             return self.entity_builder
         else:

cloudfoundry_client/v2/routes.py

@@ -1,4 +1,4 @@
-from typing import Optional, TYPE_CHECKING
+from typing import TYPE_CHECKING
 
 from cloudfoundry_client.v2.entities import EntityManager, Entity
 
@@ -10,14 +10,14 @@ class RouteManager(EntityManager):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(RouteManager, self).__init__(target_endpoint, client, "/v2/routes")
 
-    def create_tcp_route(self, domain_guid: str, space_guid: str, port: Optional[int] = None) -> Entity:
+    def create_tcp_route(self, domain_guid: str, space_guid: str, port: int | None = None) -> Entity:
         request = self._request(domain_guid=domain_guid, space_guid=space_guid)
         if port is None:
             return super(RouteManager, self)._create(request, params=dict(generate_port=True))
         else:
             request["port"] = port
             return super(RouteManager, self)._create(request)
 
-    def create_host_route(self, domain_guid: str, space_guid: str, host: str, path: Optional[str] = "") -> Entity:
+    def create_host_route(self, domain_guid: str, space_guid: str, host: str, path: str | None = "") -> Entity:
         request = dict(domain_guid=domain_guid, space_guid=space_guid, host=host, path=path)
         return super(RouteManager, self)._create(request)