diff --git a/.github/workflows/pythonpackage.yml b/.github/workflows/pythonpackage.yml index cf06cde..1582906 100644 --- a/.github/workflows/pythonpackage.yml +++ b/.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 diff --git a/cloudfoundry_client/client.py b/cloudfoundry_client/client.py index 6c85d5f..9163f1b 100644 --- a/cloudfoundry_client/client.py +++ b/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) diff --git a/cloudfoundry_client/common_objects.py b/cloudfoundry_client/common_objects.py index 36ef6ea..99e9c49 100644 --- a/cloudfoundry_client/common_objects.py +++ b/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 diff --git a/cloudfoundry_client/doppler/websocket_envelope_reader.py b/cloudfoundry_client/doppler/websocket_envelope_reader.py index 3bb513e..1a2d77d 100644 --- a/cloudfoundry_client/doppler/websocket_envelope_reader.py +++ b/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)) diff --git a/cloudfoundry_client/networking/entities.py b/cloudfoundry_client/networking/entities.py index 07d34e7..02fbe09 100644 --- a/cloudfoundry_client/networking/entities.py +++ b/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: diff --git a/cloudfoundry_client/operations/push/file_helper.py b/cloudfoundry_client/operations/push/file_helper.py index 713fa0c..d4ac4fd 100644 --- a/cloudfoundry_client/operations/push/file_helper.py +++ b/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) diff --git a/cloudfoundry_client/operations/push/push.py b/cloudfoundry_client/operations/push/push.py index c894bd1..b7f0f3b 100644 --- a/cloudfoundry_client/operations/push/push.py +++ b/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"]) diff --git a/cloudfoundry_client/v2/apps.py b/cloudfoundry_client/v2/apps.py index cf68ed8..857a76f 100644 --- a/cloudfoundry_client/v2/apps.py +++ b/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( diff --git a/cloudfoundry_client/v2/entities.py b/cloudfoundry_client/v2/entities.py index e324d9e..0fac7bf 100644 --- a/cloudfoundry_client/v2/entities.py +++ b/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: diff --git a/cloudfoundry_client/v2/routes.py b/cloudfoundry_client/v2/routes.py index 1e4d52c..4fdcf92 100644 --- a/cloudfoundry_client/v2/routes.py +++ b/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,7 +10,7 @@ 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)) @@ -18,6 +18,6 @@ def create_tcp_route(self, domain_guid: str, space_guid: str, port: Optional[int 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) diff --git a/cloudfoundry_client/v2/service_bindings.py b/cloudfoundry_client/v2/service_bindings.py index 00fc625..21f4e80 100644 --- a/cloudfoundry_client/v2/service_bindings.py +++ b/cloudfoundry_client/v2/service_bindings.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,7 +10,7 @@ class ServiceBindingManager(EntityManager): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServiceBindingManager, self).__init__(target_endpoint, client, "/v2/service_bindings") - def create(self, app_guid: str, instance_guid: str, parameters: Optional[dict] = None, name: Optional[str] = None) -> Entity: + def create(self, app_guid: str, instance_guid: str, parameters: dict | None = None, name: str | None = None) -> Entity: request = self._request(app_guid=app_guid, service_instance_guid=instance_guid) request["parameters"] = parameters request["name"] = name diff --git a/cloudfoundry_client/v2/service_brokers.py b/cloudfoundry_client/v2/service_brokers.py index abe44c7..b29e8b5 100644 --- a/cloudfoundry_client/v2/service_brokers.py +++ b/cloudfoundry_client/v2/service_brokers.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.v2.entities import EntityManager, Entity @@ -11,7 +11,7 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServiceBrokerManager, self).__init__(target_endpoint, client, "/v2/service_brokers") def create( - self, broker_url: str, broker_name: str, auth_username: str, auth_password: str, space_guid: Optional[str] = None + self, broker_url: str, broker_name: str, auth_username: str, auth_password: str, space_guid: str | None = None ) -> Entity: request = self._request(broker_url=broker_url, name=broker_name, auth_username=auth_username, auth_password=auth_password) request["space_guid"] = space_guid @@ -20,10 +20,10 @@ def create( def update( self, broker_guid: str, - broker_url: Optional[str] = None, - broker_name: Optional[str] = None, - auth_username: Optional[str] = None, - auth_password: Optional[str] = None, + broker_url: str | None = None, + broker_name: str | None = None, + auth_username: str | None = None, + auth_password: str | None = None, ) -> Entity: request = self._request() request["broker_url"] = broker_url diff --git a/cloudfoundry_client/v2/service_instances.py b/cloudfoundry_client/v2/service_instances.py index f17fa7f..8d6ebbd 100644 --- a/cloudfoundry_client/v2/service_instances.py +++ b/cloudfoundry_client/v2/service_instances.py @@ -1,4 +1,4 @@ -from typing import Optional, List, Dict, TYPE_CHECKING +from typing import List, Dict, TYPE_CHECKING from cloudfoundry_client.v2.entities import EntityManager, Entity @@ -17,9 +17,9 @@ def create( space_guid: str, instance_name: str, plan_guid: str, - parameters: Optional[dict] = None, + parameters: dict | None = None, tags: List[str] = None, - accepts_incomplete: Optional[bool] = False, + accepts_incomplete: bool | None = False, ) -> Entity: request = self._request(name=instance_name, space_guid=space_guid, service_plan_guid=plan_guid) request["parameters"] = parameters @@ -30,11 +30,11 @@ def create( def update( self, instance_guid: str, - instance_name: Optional[str] = None, - plan_guid: Optional[str] = None, - parameters: Optional[dict] = None, + instance_name: str | None = None, + plan_guid: str | None = None, + parameters: dict | None = None, tags: List[str] = None, - accepts_incomplete: Optional[bool] = False, + accepts_incomplete: bool | None = False, ) -> Entity: request = self._request() request["name"] = instance_name @@ -47,7 +47,7 @@ def update( def list_permissions(self, instance_guid: str) -> Dict[str, bool]: return super(ServiceInstanceManager, self)._get("%s/%s/permissions" % (self.entity_uri, instance_guid), dict) - def remove(self, instance_guid: str, accepts_incomplete: Optional[bool] = False, purge: Optional[bool] = False): + def remove(self, instance_guid: str, accepts_incomplete: bool | None = False, purge: bool | None = False): parameters = {} if accepts_incomplete: parameters["accepts_incomplete"] = "true" diff --git a/cloudfoundry_client/v2/service_keys.py b/cloudfoundry_client/v2/service_keys.py index cb87c3e..d807f1a 100644 --- a/cloudfoundry_client/v2/service_keys.py +++ b/cloudfoundry_client/v2/service_keys.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,7 +10,7 @@ class ServiceKeyManager(EntityManager): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServiceKeyManager, self).__init__(target_endpoint, client, "/v2/service_keys") - def create(self, service_instance_guid: str, name: str, parameters: Optional[dict] = None) -> Entity: + def create(self, service_instance_guid: str, name: str, parameters: dict | None = None) -> Entity: request = self._request(service_instance_guid=service_instance_guid, name=name) request["parameters"] = parameters return super(ServiceKeyManager, self)._create(request) diff --git a/cloudfoundry_client/v3/apps.py b/cloudfoundry_client/v3/apps.py index 5ec8486..01746d4 100644 --- a/cloudfoundry_client/v3/apps.py +++ b/cloudfoundry_client/v3/apps.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING from cloudfoundry_client.common_objects import JsonObject, Pagination from cloudfoundry_client.v3.entities import EntityManager, Entity @@ -16,7 +16,7 @@ def restart(self, application_guid: str): self.entity_uri, application_guid)) - def remove(self, application_guid: str, asynchronous: bool = True) -> Optional[str]: + def remove(self, application_guid: str, asynchronous: bool = True) -> str | None: return super(AppManager, self)._remove(application_guid, asynchronous) def get_env(self, application_guid: str) -> JsonObject: diff --git a/cloudfoundry_client/v3/buildpacks.py b/cloudfoundry_client/v3/buildpacks.py index f6a1284..ba14c9a 100644 --- a/cloudfoundry_client/v3/buildpacks.py +++ b/cloudfoundry_client/v3/buildpacks.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, Entity @@ -13,12 +13,12 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): def create( self, name: str, - position: Optional[int] = 0, - enabled: Optional[bool] = True, - locked: Optional[bool] = False, - stack: Optional[str] = None, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, + position: int | None = 0, + enabled: bool | None = True, + locked: bool | None = False, + stack: str | None = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, ) -> Entity: data = { "name": name, @@ -30,19 +30,19 @@ def create( } return super(BuildpackManager, self)._create(data) - def remove(self, buildpack_guid: str, asynchronous: bool = True) -> Optional[str]: + def remove(self, buildpack_guid: str, asynchronous: bool = True) -> str | None: return super(BuildpackManager, self)._remove(buildpack_guid, asynchronous) def update( self, buildpack_guid: str, name: str, - position: Optional[int] = 0, - enabled: Optional[bool] = True, - locked: Optional[bool] = False, - stack: Optional[str] = None, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, + position: int | None = 0, + enabled: bool | None = True, + locked: bool | None = False, + stack: str | None = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, ) -> Entity: data = { "name": name, diff --git a/cloudfoundry_client/v3/domains.py b/cloudfoundry_client/v3/domains.py index d1f5db3..9068cd1 100644 --- a/cloudfoundry_client/v3/domains.py +++ b/cloudfoundry_client/v3/domains.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.common_objects import Pagination from cloudfoundry_client.v3.entities import EntityManager, ToOneRelationship, ToManyRelationship, Entity @@ -26,11 +26,11 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): def create( self, name: str, - internal: Optional[bool] = False, - organization: Optional[ToOneRelationship] = None, - shared_organizations: Optional[ToManyRelationship] = None, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, + internal: bool | None = False, + organization: ToOneRelationship | None = None, + shared_organizations: ToManyRelationship | None = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, ) -> Domain: data = { "name": name, @@ -47,11 +47,11 @@ def list_domains_for_org(self, org_guid: str, **kwargs) -> Pagination[Entity]: uri = "/v3/organizations/{guid}/domains".format(guid=org_guid) return self._list(uri, **kwargs) - def update(self, domain_guid: str, meta_labels: Optional[dict] = None, meta_annotations: Optional[dict] = None) -> Domain: + def update(self, domain_guid: str, meta_labels: dict | None = None, meta_annotations: dict | None = None) -> Domain: data = {"metadata": {"labels": meta_labels, "annotations": meta_annotations}} return super(DomainManager, self)._update(domain_guid, data) - def remove(self, domain_guid: str, asynchronous: bool = True) -> Optional[str]: + def remove(self, domain_guid: str, asynchronous: bool = True) -> str | None: return super(DomainManager, self)._remove(domain_guid, asynchronous) def __create_shared_domain_url(self, domain_guid: str) -> str: diff --git a/cloudfoundry_client/v3/entities.py b/cloudfoundry_client/v3/entities.py index 7fa7d23..de4b8b4 100644 --- a/cloudfoundry_client/v3/entities.py +++ b/cloudfoundry_client/v3/entities.py @@ -1,6 +1,6 @@ import functools from json import JSONDecodeError -from typing import Any, Optional, List, Tuple, Union, TypeVar, TYPE_CHECKING, Callable, Type +from typing import Any, List, Tuple, Union, TypeVar, TYPE_CHECKING, Callable, Type from urllib.parse import quote, urlparse from requests import Response @@ -63,7 +63,7 @@ def default_method(m, u): return default_method @staticmethod - def _manager_method(link_name: str, link_method: str) -> Optional[str]: + def _manager_method(link_name: str, link_method: str) -> str | None: if link_method == "get": if link_name.endswith("s"): return "_attempt_to_paginate" @@ -79,7 +79,7 @@ def _manager_method(link_name: str, link_method: str) -> Optional[str]: class Relationship(JsonObject): - def __init__(self, guid: Optional[str]): + def __init__(self, guid: str | None): super(Relationship, self).__init__(guid=guid) @@ -93,7 +93,7 @@ def from_json_object(to_one_relationship: JsonObject): result.update(to_one_relationship) return result - def __init__(self, guid: Optional[str]): + def __init__(self, guid: str | None): super(ToOneRelationship, self).__init__(data=Relationship(guid)) self.guid = guid @@ -120,24 +120,24 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient", entity_ur self.client = client self.entity_type = entity_type - def _post(self, url: str, data: Optional[dict] = None, files: Any = None, entity_type: ENTITY_TYPE = None) -> Entity: + 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: Optional[ENTITY_TYPE] = None, **kwargs) -> Entity: + def _get(self, url: str, entity_type: ENTITY_TYPE | None = None, **kwargs) -> Entity: 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: Optional[ENTITY_TYPE] = None) -> Entity: + def _put(self, url: str, data: dict, entity_type: ENTITY_TYPE | None = None) -> Entity: response = self.client.put(url, json=data) return self._read_response(response, entity_type) - def _patch(self, url: str, data: dict, entity_type: Optional[ENTITY_TYPE] = None) -> Entity: + def _patch(self, url: str, data: dict, entity_type: ENTITY_TYPE | None = None) -> Entity: response = self.client.patch(url, json=data) return self._read_response(response, entity_type) - def _delete(self, url: str) -> Optional[str]: + def _delete(self, url: str) -> str | None: response = self.client.delete(url) return self._location(response) @@ -148,7 +148,7 @@ def _location(response): except (AttributeError, KeyError): return None - def _remove(self, resource_id: str, asynchronous: bool = True) -> Optional[str]: + def _remove(self, resource_id: str, asynchronous: bool = True) -> str | None: url = "%s%s/%s" % (self.target_endpoint, self.entity_uri, resource_id) job_location = self._delete(url) if job_location is not None: @@ -165,12 +165,12 @@ 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: Optional[ENTITY_TYPE] = None, **kwargs) -> Pagination[Entity]: + def _list(self, requested_path: str, entity_type: ENTITY_TYPE | None = None, **kwargs) -> Pagination[Entity]: 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: Optional[ENTITY_TYPE] = None) \ + def _attempt_to_paginate(self, url_requested: str, entity_type: ENTITY_TYPE | None = None) \ -> Union[Pagination[Entity], Entity]: response_json = self._read_response(self.client.get(url_requested), JsonObject) if "resources" in response_json: @@ -178,7 +178,7 @@ def _attempt_to_paginate(self, url_requested: str, entity_type: Optional[ENTITY_ else: return response_json - def _pagination(self, page: JsonObject, entity_type: Optional[ENTITY_TYPE] = None) -> Pagination[Entity]: + def _pagination(self, page: JsonObject, entity_type: ENTITY_TYPE | None = None) -> Pagination[Entity]: def _entity(json_object: JsonObject) -> Entity: return self._entity(json_object, entity_type) @@ -188,7 +188,7 @@ def _entity(json_object: JsonObject) -> Entity: lambda p: p["resources"], _entity) - def _next_page(self, current_page: JsonObject) -> Optional[JsonObject]: + def _next_page(self, current_page: JsonObject) -> JsonObject | None: pagination = current_page.get("pagination") if ( pagination is None @@ -233,7 +233,7 @@ def len(self, **kwargs): 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("per_page", 1) for entity in self._list(self.entity_uri, **kwargs): return entity @@ -246,7 +246,7 @@ def get(self, entity_id: str, *extra_paths, **kwargs) -> Entity: 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: Optional[ENTITY_TYPE]) -> Union[JsonObject, Entity]: + def _read_response(self, response: Response, entity_type: ENTITY_TYPE | None) -> Union[JsonObject, Entity]: try: result = response.json(object_pairs_hook=JsonObject) except JSONDecodeError: @@ -288,7 +288,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: Optional[ENTITY_TYPE]) -> Union[JsonObject, Entity]: + def _entity(self, result: JsonObject, entity_type: ENTITY_TYPE | None) -> Union[JsonObject, Entity]: 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: diff --git a/cloudfoundry_client/v3/feature_flags.py b/cloudfoundry_client/v3/feature_flags.py index 47c35d8..c75517d 100644 --- a/cloudfoundry_client/v3/feature_flags.py +++ b/cloudfoundry_client/v3/feature_flags.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, Entity @@ -10,6 +10,6 @@ class FeatureFlagManager(EntityManager): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(FeatureFlagManager, self).__init__(target_endpoint, client, "/v3/feature_flags") - def update(self, name: str, enabled: Optional[bool] = True, custom_error_message: Optional[str] = None) -> Entity: + def update(self, name: str, enabled: bool | None = True, custom_error_message: str | None = None) -> Entity: data = {"enabled": enabled, "custom_error_message": custom_error_message} return super(FeatureFlagManager, self)._update(name, data) diff --git a/cloudfoundry_client/v3/isolation_segments.py b/cloudfoundry_client/v3/isolation_segments.py index 67fa114..e11edb9 100644 --- a/cloudfoundry_client/v3/isolation_segments.py +++ b/cloudfoundry_client/v3/isolation_segments.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, Entity, ToManyRelationship @@ -10,12 +10,12 @@ class IsolationSegmentManager(EntityManager): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(IsolationSegmentManager, self).__init__(target_endpoint, client, "/v3/isolation_segments") - def create(self, name: str, meta_labels: Optional[dict] = None, meta_annotations: Optional[dict] = None) -> Entity: + def create(self, name: str, meta_labels: dict | None = None, meta_annotations: dict | None = None) -> Entity: data = {"name": name, "metadata": {"labels": meta_labels, "annotations": meta_annotations}} return super(IsolationSegmentManager, self)._create(data) def update( - self, isolation_segment_guid: str, name: str, meta_labels: Optional[dict] = None, meta_annotations: Optional[dict] = None + self, isolation_segment_guid: str, name: str, meta_labels: dict | None = None, meta_annotations: dict | None = None ) -> Entity: data = {"name": name, "metadata": {"labels": meta_labels, "annotations": meta_annotations}} return super(IsolationSegmentManager, self)._update(isolation_segment_guid, data) diff --git a/cloudfoundry_client/v3/organization_quotas.py b/cloudfoundry_client/v3/organization_quotas.py index abe5015..51f4b56 100644 --- a/cloudfoundry_client/v3/organization_quotas.py +++ b/cloudfoundry_client/v3/organization_quotas.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from dataclasses import dataclass, asdict from cloudfoundry_client.v3.entities import Entity, EntityManager, ToManyRelationship @@ -37,17 +37,17 @@ class OrganizationQuotaManager(EntityManager): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super().__init__(target_endpoint, client, "/v3/organization_quotas") - def remove(self, guid: str, asynchronous: bool = True) -> Optional[str]: + def remove(self, guid: str, asynchronous: bool = True) -> str | None: return super()._remove(guid, asynchronous) def create( self, name: str, - apps_quota: Optional[AppsQuota] = None, - services_quota: Optional[ServicesQuota] = None, - routes_quota: Optional[RoutesQuota] = None, - domains_quota: Optional[DomainsQuota] = None, - assigned_organizations: Optional[ToManyRelationship] = None, + apps_quota: AppsQuota | None = None, + services_quota: ServicesQuota | None = None, + routes_quota: RoutesQuota | None = None, + domains_quota: DomainsQuota | None = None, + assigned_organizations: ToManyRelationship | None = None, ) -> Entity: data = self._asdict(name, apps_quota, services_quota, routes_quota, domains_quota, assigned_organizations) return super()._create(data) @@ -56,10 +56,10 @@ def update( self, guid: str, name: str, - apps_quota: Optional[AppsQuota] = None, - services_quota: Optional[ServicesQuota] = None, - routes_quota: Optional[RoutesQuota] = None, - domains_quota: Optional[DomainsQuota] = None, + apps_quota: AppsQuota | None = None, + services_quota: ServicesQuota | None = None, + routes_quota: RoutesQuota | None = None, + domains_quota: DomainsQuota | None = None, ) -> Entity: data = self._asdict(name, apps_quota, services_quota, routes_quota, domains_quota) return super()._update(guid, data) @@ -74,11 +74,11 @@ def apply_to_organizations(self, guid: str, organizations: ToManyRelationship) - def _asdict( self, name: str, - apps_quota: Optional[AppsQuota] = None, - services_quota: Optional[ServicesQuota] = None, - routes_quota: Optional[RoutesQuota] = None, - domains_quota: Optional[DomainsQuota] = None, - assigned_organizations: Optional[ToManyRelationship] = None, + apps_quota: AppsQuota | None = None, + services_quota: ServicesQuota | None = None, + routes_quota: RoutesQuota | None = None, + domains_quota: DomainsQuota | None = None, + assigned_organizations: ToManyRelationship | None = None, ): data = {"name": name} if apps_quota: diff --git a/cloudfoundry_client/v3/organizations.py b/cloudfoundry_client/v3/organizations.py index 47064ff..252411d 100644 --- a/cloudfoundry_client/v3/organizations.py +++ b/cloudfoundry_client/v3/organizations.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, Entity, ToOneRelationship @@ -11,7 +11,7 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(OrganizationManager, self).__init__(target_endpoint, client, "/v3/organizations") def create( - self, name: str, suspended: bool, meta_labels: Optional[dict] = None, meta_annotations: Optional[dict] = None + self, name: str, suspended: bool, meta_labels: dict | None = None, meta_annotations: dict | None = None ) -> Entity: data = {"name": name, "suspended": suspended, "metadata": {"labels": meta_labels, "annotations": meta_annotations}} return super(OrganizationManager, self)._create(data) @@ -20,9 +20,9 @@ def update( self, guid: str, name: str, - suspended: Optional[bool] = None, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, + suspended: bool | None = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, ) -> Entity: data = {"name": name} if suspended is not None: @@ -36,7 +36,7 @@ def update( data["metadata"] = metadata return super(OrganizationManager, self)._update(guid, data) - def remove(self, guid: str, asynchronous: bool = True) -> Optional[str]: + def remove(self, guid: str, asynchronous: bool = True) -> str | None: return super(OrganizationManager, self)._remove(guid, asynchronous) def assign_default_isolation_segment(self, org_guid: str, iso_seg_guid: str) -> Entity: diff --git a/cloudfoundry_client/v3/roles.py b/cloudfoundry_client/v3/roles.py index 9123d5a..d7f00c2 100644 --- a/cloudfoundry_client/v3/roles.py +++ b/cloudfoundry_client/v3/roles.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager @@ -10,5 +10,5 @@ class RoleManager(EntityManager): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(RoleManager, self).__init__(target_endpoint, client, "/v3/roles") - def remove(self, role_guid: str, asynchronous: bool = True) -> Optional[str]: + def remove(self, role_guid: str, asynchronous: bool = True) -> str | None: return super(RoleManager, self)._remove(role_guid, asynchronous) diff --git a/cloudfoundry_client/v3/security_groups.py b/cloudfoundry_client/v3/security_groups.py index b134e03..43c32bc 100644 --- a/cloudfoundry_client/v3/security_groups.py +++ b/cloudfoundry_client/v3/security_groups.py @@ -1,6 +1,6 @@ from dataclasses import dataclass, asdict from enum import Enum, auto -from typing import TYPE_CHECKING, Optional, List +from typing import TYPE_CHECKING, List from cloudfoundry_client.v3.entities import EntityManager, ToManyRelationship, Entity, ToOneRelationship @@ -22,17 +22,17 @@ def __repr__(self): class Rule: protocol: RuleProtocol destination: str - ports: Optional[str] = None - type: Optional[int] = None - code: Optional[int] = None - description: Optional[str] = None - log: Optional[bool] = None + ports: str | None = None + type: int | None = None + code: int | None = None + description: str | None = None + log: bool | None = None @dataclass class GloballyEnabled: - running: Optional[bool] = None - staging: Optional[bool] = None + running: bool | None = None + staging: bool | None = None class SecurityGroupManager(EntityManager): @@ -41,24 +41,24 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): def create(self, name: str, - rules: Optional[List[Rule]] = None, - globally_enabled: Optional[GloballyEnabled] = None, - staging_spaces: Optional[ToManyRelationship] = None, - running_spaces: Optional[ToManyRelationship] = None) -> Entity: + rules: List[Rule] | None = None, + globally_enabled: GloballyEnabled | None = None, + staging_spaces: ToManyRelationship | None = None, + running_spaces: ToManyRelationship | None = None) -> Entity: payload = self._generate_payload(name, rules, globally_enabled, staging_spaces, running_spaces) return super()._create(payload) def update(self, security_group_id: str, - name: Optional[str] = None, - rules: Optional[List[Rule]] = None, - globally_enabled: Optional[GloballyEnabled] = None, - staging_spaces: Optional[ToManyRelationship] = None, - running_spaces: Optional[ToManyRelationship] = None) -> Entity: + name: str | None = None, + rules: List[Rule] | None = None, + globally_enabled: GloballyEnabled | None = None, + staging_spaces: ToManyRelationship | None = None, + running_spaces: ToManyRelationship | None = None) -> Entity: payload = self._generate_payload(name, rules, globally_enabled, staging_spaces, running_spaces) return super()._update(security_group_id, payload) - def remove(self, security_group_id: str, asynchronous: bool = True) -> Optional[str]: + def remove(self, security_group_id: str, asynchronous: bool = True) -> str | None: return super()._remove(security_group_id, asynchronous) def bind_running_security_group_to_spaces(self, security_group_id: str, space_guids: ToManyRelationship) \ @@ -90,11 +90,11 @@ def _unbind_space(self, security_group_id: str, space_guid: ToOneRelationship, r super()._delete(url) @staticmethod - def _generate_payload(name: Optional[str], - rules: Optional[List[Rule]], - globally_enabled: Optional[GloballyEnabled], - staging_spaces: Optional[ToManyRelationship], - running_spaces: Optional[ToManyRelationship]): + def _generate_payload(name: str | None, + rules: List[Rule] | None, + globally_enabled: GloballyEnabled | None, + staging_spaces: ToManyRelationship | None, + running_spaces: ToManyRelationship | None): payload = {} if name: payload["name"] = name diff --git a/cloudfoundry_client/v3/service_brokers.py b/cloudfoundry_client/v3/service_brokers.py index c39795b..f1febc6 100644 --- a/cloudfoundry_client/v3/service_brokers.py +++ b/cloudfoundry_client/v3/service_brokers.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, Entity, ToOneRelationship @@ -16,9 +16,9 @@ def create( url: str, auth_username: str, auth_password: str, - space_guid: Optional[str] = None, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, + space_guid: str | None = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, ) -> Entity: credentials = {"type": "basic", "credentials": {"username": auth_username, "password": auth_password}} payload = dict(name=name, url=url, authentication=credentials) @@ -36,12 +36,12 @@ def create( def update( self, guid: str, - name: Optional[str] = None, - url: Optional[str] = None, - auth_username: Optional[str] = None, - auth_password: Optional[str] = None, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, + name: str | None = None, + url: str | None = None, + auth_username: str | None = None, + auth_password: str | None = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, ) -> Entity: payload = dict() if name: @@ -59,5 +59,5 @@ def update( payload["metadata"] = metadata return super(ServiceBrokerManager, self)._update(guid, payload) - def remove(self, guid: str, asynchronous: bool = True) -> Optional[str]: + def remove(self, guid: str, asynchronous: bool = True) -> str | None: return super(ServiceBrokerManager, self)._remove(guid, asynchronous) diff --git a/cloudfoundry_client/v3/service_credential_bindings.py b/cloudfoundry_client/v3/service_credential_bindings.py index d587b7a..b380b90 100644 --- a/cloudfoundry_client/v3/service_credential_bindings.py +++ b/cloudfoundry_client/v3/service_credential_bindings.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Optional, Union +from typing import TYPE_CHECKING, Union from cloudfoundry_client.v3.entities import EntityManager, Entity, ToOneRelationship @@ -16,10 +16,10 @@ def create( name: str, service_credential_binding_type: str, service_instance_guid: str, - application_guid: Optional[str], - parameters: Optional[dict], - meta_labels: Optional[dict], - meta_annotations: Optional[dict], + application_guid: str | None, + parameters: dict | None, + meta_labels: dict | None, + meta_annotations: dict | None, asynchronous: bool = True, ) -> Union[str, Entity, None]: data = { diff --git a/cloudfoundry_client/v3/service_instances.py b/cloudfoundry_client/v3/service_instances.py index ad33614..cfa253f 100644 --- a/cloudfoundry_client/v3/service_instances.py +++ b/cloudfoundry_client/v3/service_instances.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING, List +from typing import TYPE_CHECKING, List from cloudfoundry_client.common_objects import JsonObject from cloudfoundry_client.v3.entities import Entity, EntityManager, ToOneRelationship @@ -16,10 +16,10 @@ def create( name: str, space_guid: str, service_plan_guid: str, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, - parameters: Optional[dict] = None, - tags: Optional[List[str]] = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, + parameters: dict | None = None, + tags: List[str] | None = None, ) -> Entity: data = { "name": name, @@ -42,13 +42,13 @@ def create( def update( self, instance_guid: str, - name: Optional[str] = None, - parameters: Optional[dict] = None, - service_plan: Optional[str] = None, - maintenance_info: Optional[str] = None, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, - tags: Optional[List[str]] = None + name: str | None = None, + parameters: dict | None = None, + service_plan: str | None = None, + maintenance_info: str | None = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, + tags: List[str] | None = None ) -> Entity: data = {} if name: diff --git a/cloudfoundry_client/v3/service_offerings.py b/cloudfoundry_client/v3/service_offerings.py index 37f7d0c..c47ca66 100644 --- a/cloudfoundry_client/v3/service_offerings.py +++ b/cloudfoundry_client/v3/service_offerings.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, Entity @@ -10,7 +10,7 @@ class ServiceOfferingsManager(EntityManager): def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): super(ServiceOfferingsManager, self).__init__(target_endpoint, client, "/v3/service_offerings") - def update(self, guid: str, meta_labels: Optional[dict] = None, meta_annotations: Optional[dict] = None) -> Entity: + def update(self, guid: str, meta_labels: dict | None = None, meta_annotations: dict | None = None) -> Entity: payload = dict() if meta_labels or meta_annotations: metadata = dict() diff --git a/cloudfoundry_client/v3/service_plans.py b/cloudfoundry_client/v3/service_plans.py index 257c3bf..0534cec 100644 --- a/cloudfoundry_client/v3/service_plans.py +++ b/cloudfoundry_client/v3/service_plans.py @@ -1,4 +1,4 @@ -from typing import Dict, List, Optional, TYPE_CHECKING +from typing import Dict, List, TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, Entity @@ -13,8 +13,8 @@ def __init__(self, target_endpoint: str, client: "CloudFoundryClient"): def update( self, guid: str, - meta_labels: Optional[dict] = None, - meta_annotations: Optional[dict] = None, + meta_labels: dict | None = None, + meta_annotations: dict | None = None, ) -> Entity: payload = {"metadata": {}} @@ -32,7 +32,7 @@ def get_visibility(self, service_plan_guid: str) -> Dict: # Updates a service plan visibility. It behaves similar to the POST service plan visibility endpoint but # this endpoint will REPLACE the existing list of organizations when the service plan is organization visible. - def update_visibility(self, service_plan_guid: str, type: str, organizations: Optional[List[dict]] = None) -> Dict: + def update_visibility(self, service_plan_guid: str, type: str, organizations: List[dict] | None = None) -> Dict: payload = {"type": type} if organizations: payload["organizations"] = organizations diff --git a/cloudfoundry_client/v3/spaces.py b/cloudfoundry_client/v3/spaces.py index f871a43..fd78d68 100644 --- a/cloudfoundry_client/v3/spaces.py +++ b/cloudfoundry_client/v3/spaces.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, ToOneRelationship, Entity @@ -23,7 +23,7 @@ def get_assigned_isolation_segment(self, space_guid: str) -> ToOneRelationship: ) ) - def assign_isolation_segment(self, space_guid: str, isolation_segment_guid: Optional[str]) -> ToOneRelationship: + def assign_isolation_segment(self, space_guid: str, isolation_segment_guid: str | None) -> ToOneRelationship: return ToOneRelationship.from_json_object( super(SpaceManager, self)._patch( "%s%s/%s/relationships/isolation_segment" % (self.target_endpoint, self.entity_uri, space_guid), diff --git a/cloudfoundry_client/v3/tasks.py b/cloudfoundry_client/v3/tasks.py index a7ce16e..c5b23d5 100644 --- a/cloudfoundry_client/v3/tasks.py +++ b/cloudfoundry_client/v3/tasks.py @@ -1,4 +1,4 @@ -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from cloudfoundry_client.v3.entities import EntityManager, Entity @@ -14,10 +14,10 @@ def create( self, application_guid: str, command: str, - name: Optional[str] = None, - disk_in_mb: Optional[int] = None, - memory_in_mb: Optional[int] = None, - droplet_guid: Optional[str] = None, + name: str | None = None, + disk_in_mb: int | None = None, + memory_in_mb: int | None = None, + droplet_guid: str | None = None, ) -> Entity: request = self._request(command=command) request["name"] = name diff --git a/tests/abstract_test_case.py b/tests/abstract_test_case.py index 975fc2e..b3c737f 100644 --- a/tests/abstract_test_case.py +++ b/tests/abstract_test_case.py @@ -1,7 +1,6 @@ import json import os from http import HTTPStatus -from typing import Optional from unittest.mock import MagicMock, patch from oauth2_client.credentials_manager import CredentialManager @@ -83,7 +82,7 @@ def get_fixtures_path(*paths): return os.path.join(os.path.dirname(__file__), "fixtures", *paths) @staticmethod - def mock_response(uri: str, status_code: HTTPStatus, headers: Optional[dict], *path_parts: str): + def mock_response(uri: str, status_code: HTTPStatus, headers: dict | None, *path_parts: str): if len(path_parts) > 0: file_name = path_parts[len(path_parts) - 1] extension_idx = file_name.rfind(".") diff --git a/tests/v3/test_apps.py b/tests/v3/test_apps.py index 5c9ed9d..3b51108 100644 --- a/tests/v3/test_apps.py +++ b/tests/v3/test_apps.py @@ -1,7 +1,7 @@ import unittest import yaml from http import HTTPStatus -from typing import Optional, List, Union +from typing import List, Union from abstract_test_case import AbstractTestCase from cloudfoundry_client.common_objects import JsonObject, Pagination @@ -143,17 +143,17 @@ def test_get_manifest(self): manifest_response: str = self.client.v3.apps.get_manifest("app_id") self.assertIsInstance(manifest_response, str) manifest: dict = yaml.safe_load(manifest_response) - applications: Optional[list[dict]] = manifest.get("applications") + applications: list[dict] | None = manifest.get("applications") self.assertIsInstance(applications, list) self.assertEqual(len(applications), 1) application: dict = applications[0] self.assertEqual(application.get("name"), "my-app") self.assertEqual(application.get("stack"), "cflinuxfs4") - application_services: Optional[list[str]] = application.get("services") + application_services: list[str] | None = application.get("services") self.assertIsInstance(application_services, list) self.assertEqual(len(application_services), 1) self.assertEqual(application_services[0], "my-service") - application_routes: Optional[List[Union[dict, str]]] = application.get("routes") + application_routes: List[Union[dict, str]] | None = application.get("routes") self.assertIsInstance(application_routes, list) self.assertEqual(len(application_routes), 1) application_route: dict = application_routes[0]