diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 26e5f56..eead7f9 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -6,16 +6,19 @@ on: jobs: release: runs-on: ubuntu-latest + environment: + name: pypi + url: https://pypi.org/p/3scale-api + permissions: + id-token: write # IMPORTANT: this permission is mandatory for trusted publishing steps: - - uses: actions/checkout@v2 - - uses: actions/setup-python@v2 + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 with: python-version: '3.11' - name: setup run: pip install wheel - name: build run: python setup.py --release-version ${GITHUB_REF#refs/tags/v} sdist bdist_wheel - - name: release + - name: Publish package distributions to PyPI uses: pypa/gh-action-pypi-publish@release/v1 - with: - password: ${{ secrets.PYPI_API_TOKEN }} diff --git a/Pipfile b/Pipfile index 1fc5ba7..1dae210 100644 --- a/Pipfile +++ b/Pipfile @@ -4,11 +4,14 @@ url = "https://pypi.org/simple" verify_ssl = true [dev-packages] +black = "*" +flake8 = "*" +mypy = "*" pylint = "*" pytest = "*" -responses = "*" python-dotenv = "*" -flake8 = "*" +responses = "*" +types-requests = "*" [packages] requests = "*" diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..60c13e6 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,48 @@ +[metadata] +name = 3scale-api +version = 0.35.2 +description = 3scale API python client +author = Peter Stanko +author_email = stanko@mail.muni.cz +maintainer = Matej Dujava +maintainer_email = mdujava@redhat.com +url = https://github.com/3scale-qe/3scale-api-python +license = Apache-2.0 +license_files = LICENSE +long_description = file: README.md +long_description_content_type = text/markdown +classifiers = + Operating System :: OS Independent + Intended Audience :: Developers + Topic :: Utilities + Programming Language :: Python :: 3 + +[options] +packages = find: +include_package_data = True +install_requires = + requests + backoff + +[options.packages.find] +exclude = tests + +[options.extras_require] +dev = + coverage + flake8 + mypy + pylint + pytest + python-dotenv + responses +docs = sphinx + +[flake8] +max-line-length = 120 +ignore = E203,W503 + +[mypy] +ignore_missing_imports = True +check_untyped_defs = true +local_partial_types = true diff --git a/setup.py b/setup.py index 5d838fb..6068493 100644 --- a/setup.py +++ b/setup.py @@ -1,56 +1,3 @@ -import re -import sys +from setuptools import setup -from setuptools import find_packages, setup - -with open("README.md", "r") as fh: - long_description = fh.read() - -VERSION = "devel" -if sys.argv[1] == "--release-version": - sys.argv.pop(1) - VERSION = sys.argv.pop(1) - assert re.match(r"[0-9]+\.[0-9]+\.[0-9]+", VERSION), "Version definition required as first arg" - -requirements = ['requests', 'backoff'] - -extra_requirements = { - 'dev': [ - 'pytest', - 'coverage', - 'python-dotenv', - 'responses' - - ], - 'docs': ['sphinx'] -} - -setup(name='3scale-api', - version=VERSION, - description='3scale API python client', - author='Peter Stanko', - author_email='stanko@mail.muni.cz', - maintainer='Peter Stanko', - url='https://github.com/3scale-qe/3scale-api-python', - packages=find_packages(exclude=("tests",)), - long_description=long_description, - long_description_content_type='text/markdown', - include_package_data=True, - install_requires=requirements, - extras_require=extra_requirements, - entry_points={}, - classifiers=[ - "Operating System :: OS Independent", - "License :: OSI Approved :: Apache Software License", - 'Intended Audience :: Developers', - 'Topic :: Utilities', - "Programming Language :: Python :: 3", - 'Programming Language :: Python :: 3.7', - 'Programming Language :: Python :: 3.8', - 'Programming Language :: Python :: 3.9', - 'Programming Language :: Python :: 3.10', - 'Programming Language :: Python :: 3.11', - 'Programming Language :: Python :: 3.12', - 'Programming Language :: Python :: 3.13', - ], - ) +setup() diff --git a/tests/integration/auth/test_app_key_authorization.py b/tests/integration/auth/test_app_key_authorization.py index db48a1a..bcdfc20 100644 --- a/tests/integration/auth/test_app_key_authorization.py +++ b/tests/integration/auth/test_app_key_authorization.py @@ -19,7 +19,7 @@ def proxy(service, proxy): def test_app_key_authorization(proxy, application, ssl_verify): creds = application.authobj().credentials encoded = base64.b64encode( - f"{creds['app_id']}:{creds['app_key']}".encode("utf-8")).decode("utf-8") + f"{creds['app_id']}:{creds['app_key']}".encode()).decode("utf-8") response = application.test_request(verify=ssl_verify) diff --git a/tests/integration/test_integration_response_headers.py b/tests/integration/test_integration_response_headers.py index 7bef6ed..0a33ca2 100644 --- a/tests/integration/test_integration_response_headers.py +++ b/tests/integration/test_integration_response_headers.py @@ -1,4 +1,3 @@ - def test_x_served_by(api): response = api._rest.get(url=api.admin_api_url + '/accounts') assert response.status_code == 200 diff --git a/tests/integration/test_integration_service_plans.py b/tests/integration/test_integration_service_plans.py index 7277061..3f43def 100644 --- a/tests/integration/test_integration_service_plans.py +++ b/tests/integration/test_integration_service_plans.py @@ -1,5 +1,3 @@ - - from tests.integration import asserts def test_list_service_plans(service): diff --git a/tests/integration/test_integration_service_subscriptions.py b/tests/integration/test_integration_service_subscriptions.py index 5e07441..6727258 100644 --- a/tests/integration/test_integration_service_subscriptions.py +++ b/tests/integration/test_integration_service_subscriptions.py @@ -1,4 +1,3 @@ - from tests.integration import asserts def test_list_service_subscriptions(account): diff --git a/threescale_api/auth.py b/threescale_api/auth.py index 86425d3..f29b78f 100644 --- a/threescale_api/auth.py +++ b/threescale_api/auth.py @@ -38,9 +38,7 @@ class UserKeyAuth(BaseClientAuth): def __init__(self, app, location=None): super().__init__(app, location) - self.credentials = { - self.app.service.proxy.list()["auth_user_key"]: self.app["user_key"] - } + self.credentials = {self.app.service.proxy.list()["auth_user_key"]: self.app["user_key"]} def __call__(self, request): if self.location == "authorization": @@ -57,7 +55,7 @@ def __init__(self, app, location=None): proxy = self.app.service.proxy.list() self.credentials = { proxy["auth_app_id"]: self.app["application_id"], - proxy["auth_app_key"]: self.app.keys.list()[-1]["value"] + proxy["auth_app_key"]: self.app.keys.list()[-1]["value"], } def __call__(self, request): diff --git a/threescale_api/client.py b/threescale_api/client.py index aa44dc6..6b9e8c4 100644 --- a/threescale_api/client.py +++ b/threescale_api/client.py @@ -11,8 +11,7 @@ class ThreeScaleClient: - def __init__(self, url: str, token: str, - throws: bool = True, ssl_verify: bool = True, wait: int = -1): + def __init__(self, url: str, token: str, throws: bool = True, ssl_verify: bool = True, wait: int = -1): """Creates instance of the 3scale client Args: url: 3scale instance url @@ -25,40 +24,37 @@ def __init__(self, url: str, token: str, self._rest = RestApiClient(url=url, token=token, throws=throws, ssl_verify=ssl_verify) self._services = resources.Services(self, instance_klass=resources.Service) self._accounts = resources.Accounts(self, instance_klass=resources.Account) - self._provider_accounts = \ - resources.ProviderAccounts(self, instance_klass=resources.ProviderAccount) - self._provider_account_users = \ - resources.ProviderAccountUsers(self, instance_klass=resources.ProviderAccountUser) + self._provider_accounts = resources.ProviderAccounts(self, instance_klass=resources.ProviderAccount) + self._provider_account_users = resources.ProviderAccountUsers( + self, instance_klass=resources.ProviderAccountUser + ) self._methods = resources.Methods(self, instance_klass=resources.Method) self._metrics = resources.Metrics(self, instance_klass=resources.Metric) self._analytics = resources.Analytics(self) self._tenants = resources.Tenants(self, instance_klass=resources.Tenant) self._providers = resources.Providers(self, instance_klass=resources.Provider) - self._access_tokens = \ - resources.AccessTokens(self, instance_klass=resources.AccessToken) + self._access_tokens = resources.AccessTokens(self, instance_klass=resources.AccessToken) self._active_docs = resources.ActiveDocs(self, instance_klass=resources.ActiveDoc) - self._application_plans = \ - resources.ApplicationPlans(self, instance_klass=resources.ApplicationPlan) + self._application_plans = resources.ApplicationPlans(self, instance_klass=resources.ApplicationPlan) self._account_plans = resources.AccountPlans(self, instance_klass=resources.AccountPlan) self._settings = resources.SettingsClient(self) self._admin_portal_auth_providers = resources.AdminPortalAuthProviders( - self, instance_klass=resources.AdminPortalAuthProvider) + self, instance_klass=resources.AdminPortalAuthProvider + ) self._dev_portal_auth_providers = resources.DevPortalAuthProviders( - self, instance_klass=resources.DevPortalAuthProvider) - self._policy_registry = resources.PoliciesRegistry(self, - instance_klass=resources.PolicyRegistry) + self, instance_klass=resources.DevPortalAuthProvider + ) + self._policy_registry = resources.PoliciesRegistry(self, instance_klass=resources.PolicyRegistry) self._backends = resources.Backends(self, instance_klass=resources.Backend) self._webhooks = resources.Webhooks(self) self._invoices = resources.Invoices(self, instance_klass=resources.Invoice) - self._fields_definitions =\ - resources.FieldsDefinitions(self, instance_klass=resources.FieldsDefinition) + self._fields_definitions = resources.FieldsDefinitions(self, instance_klass=resources.FieldsDefinition) self._cms_files = resources.CmsFiles(self, instance_klass=resources.CmsFile) self._cms_sections = resources.CmsSections(self, instance_klass=resources.CmsSection) self._cms_pages = resources.CmsPages(self, instance_klass=resources.CmsPage) self._cms_builtin_pages = resources.CmsBuiltinPages(self, instance_klass=resources.CmsPage) self._cms_layouts = resources.CmsLayouts(self, instance_klass=resources.CmsLayout) - self._cms_builtin_partials =\ - resources.CmsBuiltinPartials(self, instance_klass=resources.CmsPartial) + self._cms_builtin_partials = resources.CmsBuiltinPartials(self, instance_klass=resources.CmsPartial) self._cms_partials = resources.CmsPartials(self, instance_klass=resources.CmsPartial) if wait >= 0: @@ -68,8 +64,7 @@ def __init__(self, url: str, token: str, # here to mitigate the problem. This requires proper fix in checks time.sleep(wait) - @backoff.on_predicate( - backoff.constant, lambda ready: not ready, interval=6, max_tries=90, jitter=None) + @backoff.on_predicate(backoff.constant, lambda ready: not ready, interval=6, max_tries=90, jitter=None) def wait_for_tenant(self) -> bool: """ When True is returned, there is some chance the tenant is actually ready. @@ -78,13 +73,15 @@ def wait_for_tenant(self) -> bool: # ultimate readiness check. There might be duplicates though, so # worth to review it one day try: - return self.account_plans.exists(throws=True) \ - and len(self.account_plans.fetch()["plans"]) >= 1 \ - and len(self.account_plans.list()) >= 1 \ - and self.accounts.exists(throws=True) \ - and len(self.accounts.list()) >= 1 \ - and self.services.exists(throws=True) \ + return ( + self.account_plans.exists(throws=True) + and len(self.account_plans.fetch()["plans"]) >= 1 + and len(self.account_plans.list()) >= 1 + and self.accounts.exists(throws=True) + and len(self.accounts.list()) >= 1 + and self.services.exists(throws=True) and len(self.services.list()) >= 1 + ) except errors.ApiClientError as err: if err.code in (404, 409, 503): log.info("wait_for_tenant failed: %s", err) @@ -95,21 +92,21 @@ def wait_for_tenant(self) -> bool: return False @property - def rest(self) -> 'RestApiClient': + def rest(self) -> "RestApiClient": """Get REST api client instance Returns(RestApiClient): Rest api client instance """ return self._rest @property - def parent(self) -> 'ThreeScaleClient': + def parent(self) -> "ThreeScaleClient": """Parent is self - the 3scale client Returns(ThreeScaleClient): """ return self @property - def threescale_client(self) -> 'ThreeScaleClient': + def threescale_client(self) -> "ThreeScaleClient": """3scale client instance Returns(ThreeScaleClient): 3scale client instance """ @@ -124,7 +121,7 @@ def url(self) -> str: @property def url_with_token(self) -> str: - return self.rest.url.replace('//', f"//{self.rest._token}@") + return self.rest.url.replace("//", f"//{self.rest._token}@") @property def token(self) -> str: @@ -257,7 +254,7 @@ def admin_portal_auth_providers(self) -> resources.AdminPortalAuthProviders: return self._admin_portal_auth_providers @property - def policy_registry(self) -> resources.PolicyRegistry: + def policy_registry(self) -> resources.PoliciesRegistry: return self._policy_registry @property @@ -314,15 +311,15 @@ def __init__(self, url: str, token: str, throws: bool = True, ssl_verify: bool = self._token = token self._throws = throws self._ssl_verify = ssl_verify - log.debug("[REST] New instance: %s token=%s throws=%s ssl=%s", url, token, throws, - ssl_verify) + log.debug("[REST] New instance: %s token=%s throws=%s ssl=%s", url, token, throws, ssl_verify) @property def url(self) -> str: return self._url - def request(self, method='GET', url=None, path='', params: dict = None, - headers: dict = None, throws=None, **kwargs): + def request( + self, method="GET", url=None, path="", params: dict = None, headers: dict = None, throws=None, **kwargs + ): """Create new request Args: method(str): method to be used to create an request @@ -336,8 +333,8 @@ def request(self, method='GET', url=None, path='', params: dict = None, Returns: """ - if 'resource' in kwargs: - del kwargs['resource'] + if "resource" in kwargs: + del kwargs["resource"] full_url = url if url else urljoin(self.url, path) full_url = full_url + ".json" headers = headers or {} @@ -345,27 +342,27 @@ def request(self, method='GET', url=None, path='', params: dict = None, if throws is None: throws = self._throws params.update(access_token=self._token) - log.debug("[%s] (%s) params={%s} headers={%s} %s", method, full_url, params, headers, - kwargs if kwargs else '') - response = requests.request(method=method, url=full_url, headers=headers, - params=params, verify=self._ssl_verify, **kwargs) + log.debug("[%s] (%s) params={%s} headers={%s} %s", method, full_url, params, headers, kwargs if kwargs else "") + response = requests.request( + method=method, url=full_url, headers=headers, params=params, verify=self._ssl_verify, **kwargs + ) process_response = self._process_response(response, throws=throws) return process_response def get(self, *args, **kwargs): - return self.request('GET', *args, **kwargs) + return self.request("GET", *args, **kwargs) def post(self, *args, **kwargs): - return self.request('POST', *args, **kwargs) + return self.request("POST", *args, **kwargs) def put(self, *args, **kwargs): - return self.request('PUT', *args, **kwargs) + return self.request("PUT", *args, **kwargs) def delete(self, *args, **kwargs): - return self.request('DELETE', *args, **kwargs) + return self.request("DELETE", *args, **kwargs) def patch(self, *args, **kwargs): - return self.request('PATCH', *args, **kwargs) + return self.request("PATCH", *args, **kwargs) @classmethod def _process_response(cls, response: requests.Response, throws=None) -> requests.Response: diff --git a/threescale_api/defaults.py b/threescale_api/defaults.py index 833316b..01a60eb 100644 --- a/threescale_api/defaults.py +++ b/threescale_api/defaults.py @@ -14,8 +14,7 @@ class DefaultClient(collections.abc.Mapping): - def __init__(self, parent=None, instance_klass=None, - entity_name: str = None, entity_collection: str = None): + def __init__(self, parent=None, instance_klass=None, entity_name: str = None, entity_collection: str = None): """Creates instance of the default client Args: parent: Parent resource or client @@ -27,7 +26,7 @@ def __init__(self, parent=None, instance_klass=None, self._instance_klass = instance_klass self._entity_name = entity_name if entity_collection is None and entity_name is not None: - entity_collection = f'{entity_name}s' + entity_collection = f"{entity_name}s" self._entity_collection = entity_collection @property @@ -38,7 +37,7 @@ def url(self) -> str: return self.threescale_client.admin_api_url @property - def threescale_client(self) -> 'ThreeScaleClient': + def threescale_client(self) -> "ThreeScaleClient": """Gets instance of the 3scale default client Returns(TheeScaleClient): 3scale client @@ -46,14 +45,14 @@ def threescale_client(self) -> 'ThreeScaleClient': return self.parent.threescale_client @property - def rest(self) -> 'RestApiClient': + def rest(self) -> "RestApiClient": """Rest API client for the 3scale instance Returns(RestApiClient): """ return self.threescale_client.rest - def list(self, **kwargs) -> List['DefaultResource']: + def list(self, **kwargs) -> List["DefaultResource"]: """List all entities Args: **kwargs: Optional parameters @@ -63,7 +62,7 @@ def list(self, **kwargs) -> List['DefaultResource']: instance = self._list(**kwargs) return instance - def create(self, params: dict = None, **kwargs) -> 'DefaultResource': + def create(self, params: dict = None, **kwargs) -> "DefaultResource": """Create a new instance Args: params: Parameters required to create new instance @@ -104,7 +103,7 @@ def exists(self, entity_id=None, throws=False, **kwargs) -> bool: response = self.rest.get(url=url, throws=throws, **kwargs) return response.ok - def update(self, entity_id=None, params: dict = None, **kwargs) -> 'DefaultResource': + def update(self, entity_id=None, params: dict = None, **kwargs) -> "DefaultResource": """Update resource Args: entity_id(int): Entity id @@ -113,8 +112,7 @@ def update(self, entity_id=None, params: dict = None, **kwargs) -> 'DefaultResou Returns(DefaultResource): Resource instance """ - log.info(self._log_message("[UPDATE] Update ", body=params, - entity_id=entity_id, args=kwargs)) + log.info(self._log_message("[UPDATE] Update ", body=params, entity_id=entity_id, args=kwargs)) url = self._entity_url(entity_id=entity_id) response = self.rest.put(url=url, json=params, **kwargs) instance = self._create_instance(response=response) @@ -133,7 +131,7 @@ def fetch(self, entity_id: int = None, **kwargs) -> dict: response = self.rest.get(url=url, **kwargs) return utils.extract_response(response=response, entity=self._entity_name) - def __getitem__(self, selector: Union[int, 'str']) -> 'DefaultResource': + def __getitem__(self, selector: Union[int, "str"]) -> "DefaultResource": """Gets the item Args: selector(Union[int, 'str']): Selector whether id or string @@ -146,10 +144,10 @@ def __getitem__(self, selector: Union[int, 'str']) -> 'DefaultResource': def __len__(self) -> int: return len(self._list()) - def __iter__(self) -> Iterator['DefaultResource']: + def __iter__(self) -> Iterator["DefaultResource"]: return next(iter(self._list())) - def read(self, entity_id: int = None) -> 'DefaultResource': + def read(self, entity_id: int = None) -> "DefaultResource": """Read the instance, read will just create empty resource and lazyloads only if needed Args: entity_id(int): Entity id @@ -158,7 +156,7 @@ def read(self, entity_id: int = None) -> 'DefaultResource': log.debug(self._log_message("[READ] Read ", entity_id=entity_id)) return self._instance_klass(client=self, entity_id=entity_id) - def read_by_name(self, name: str, **kwargs) -> 'DefaultResource': + def read_by_name(self, name: str, **kwargs) -> "DefaultResource": """Read resource by name Args: name: Name of the resource (either system name, name, org_name ...) @@ -171,7 +169,7 @@ def read_by_name(self, name: str, **kwargs) -> 'DefaultResource': if item.entity_name and item.entity_name == name: return item - def select(self, predicate, **kwargs) -> List['DefaultResource']: + def select(self, predicate, **kwargs) -> List["DefaultResource"]: """Select resource s based on the predicate Args: predicate: Predicate @@ -180,7 +178,7 @@ def select(self, predicate, **kwargs) -> List['DefaultResource']: """ return [item for item in self._list(**kwargs) if predicate(item)] - def select_by(self, **params) -> List['DefaultResource']: + def select_by(self, **params) -> List["DefaultResource"]: """Select by params - logical and Usage example: select_by(role='admin') Args: @@ -190,14 +188,14 @@ def select_by(self, **params) -> List['DefaultResource']: log.debug("[SELECT] By params: %s", params) def predicate(item): - for (key, val) in params.items(): + for key, val in params.items(): if item[key] != val: return False return True return self.select(predicate=predicate) - def read_by(self, **params) -> 'DefaultResource': + def read_by(self, **params) -> "DefaultResource": """Read by params - it will return just one instance of the resource Args: **params: params used for selection @@ -216,7 +214,7 @@ def _log_message(self, message, entity_id=None, body=None, args=None) -> str: msg += f" args={args}" return msg - def _list(self, **kwargs) -> List['DefaultResource']: + def _list(self, **kwargs) -> List["DefaultResource"]: """Internal list implementation used in list or `select` methods Args: **kwargs: Optional parameters @@ -232,7 +230,7 @@ def _list(self, **kwargs) -> List['DefaultResource']: def _entity_url(self, entity_id=None) -> str: if not entity_id: return self.url - return self.url + '/' + str(entity_id) + return self.url + "/" + str(entity_id) def _create_instance(self, response: requests.Response, klass=None, collection: bool = False): klass = klass or self._instance_klass @@ -244,7 +242,7 @@ def _create_instance(self, response: requests.Response, klass=None, collection: def _extract_resource(self, response, collection) -> Union[List, Dict]: extract_params = dict(response=response, entity=self._entity_name) if collection: - extract_params['collection'] = self._entity_collection + extract_params["collection"] = self._entity_collection extracted = utils.extract_response(**extract_params) return extracted @@ -260,8 +258,9 @@ def __make_instance(self, extracted: dict, klass): class DefaultResource(collections.abc.MutableMapping): - def __init__(self, client: DefaultClient = None, entity_id: int = None, entity_name: str = None, - entity: dict = None): + def __init__( + self, client: DefaultClient = None, entity_id: int = None, entity_name: str = None, entity: dict = None + ): """Create instance of the resource Args: client: Client instance of the resource @@ -269,17 +268,17 @@ def __init__(self, client: DefaultClient = None, entity_id: int = None, entity_n entity_name(str): Entity name field (system_name or name ...) entity(dict): Entity instance """ - self._entity_id = entity_id or entity.get('id') + self._entity_id = entity_id or entity.get("id") self._entity = entity self._client = client self._entity_name = entity_name @property - def threescale_client(self) -> 'ThreeScaleClient': + def threescale_client(self) -> "ThreeScaleClient": return self.client.threescale_client @property - def parent(self) -> 'DefaultResource': + def parent(self) -> "DefaultResource": return self.client.parent @parent.setter @@ -305,7 +304,7 @@ def client(self) -> DefaultClient: @property def entity_id(self) -> int: - return self._entity_id or self._entity.get('id') + return self._entity_id or self._entity.get("id") @entity_id.setter def entity_id(self, value): @@ -334,9 +333,9 @@ def __repr__(self) -> str: def __eq__(self, other) -> bool: return ( - self.__class__ == other.__class__ and - self.entity_name == other.entity_name and - self.entity_id == other.entity_id + self.__class__ == other.__class__ + and self.entity_name == other.entity_name + and self.entity_id == other.entity_id ) def get(self, item): @@ -345,7 +344,7 @@ def get(self, item): def set(self, item: str, value: Any): self.entity[item] = value - def _lazy_load(self, **kwargs) -> 'DefaultResource': + def _lazy_load(self, **kwargs) -> "DefaultResource": if self._entity is None: # Lazy load the entity fetched = self.fetch(**kwargs) @@ -357,7 +356,7 @@ def _lazy_load(self, **kwargs) -> 'DefaultResource': return None return self - def read(self, **kwargs) -> 'DefaultResource': + def read(self, **kwargs) -> "DefaultResource": self._invalidate() self._lazy_load(**kwargs) return self @@ -371,14 +370,11 @@ def exists(self, **kwargs) -> bool: def delete(self, **kwargs): self.client.delete(entity_id=self.entity_id, resource=self, **kwargs) - def update(self, params: dict = None, **kwargs) -> 'DefaultResource': + def update(self, params: dict = None, **kwargs) -> "DefaultResource": new_params = {**self.entity} if params: new_params.update(params) - new_entity = self.client.update(entity_id=self.entity_id, - params=new_params, - resource=self, - **kwargs) + new_entity = self.client.update(entity_id=self.entity_id, params=new_params, resource=self, **kwargs) self._entity = new_entity.entity return self @@ -387,7 +383,7 @@ def _invalidate(self): class DefaultPaginationClient(DefaultClient): - """ Client to handle API endpoints with pagination. + """Client to handle API endpoints with pagination. List of endpoints supporting pagination with per_page size: - accounts 500 limits per app plan 50 - not implemented in client @@ -401,12 +397,13 @@ class DefaultPaginationClient(DefaultClient): - invoice list 20 - all cms 100 """ + def __init__(self, *args, per_page=500, **kwargs): self.per_page = per_page super().__init__(*args, **kwargs) def _list(self, **kwargs): - """ List all objects via paginated API endpoint """ + """List all objects via paginated API endpoint""" kwargs = kwargs.copy() kwargs.setdefault("params", {}) if "page" in kwargs["params"] or self.per_page is None: @@ -432,7 +429,7 @@ def __iter__(self): class DefaultPlanClient(DefaultClient): - def set_default(self, entity_id: int, **kwargs) -> 'DefaultPlanResource': + def set_default(self, entity_id: int, **kwargs) -> "DefaultPlanResource": """Sets default plan for the entity Args: entity_id: Entity id @@ -440,12 +437,12 @@ def set_default(self, entity_id: int, **kwargs) -> 'DefaultPlanResource': Returns(DefaultPlanResource): """ log.info(self._log_message("[PLAN] Set default ", entity_id=entity_id, args=kwargs)) - url = self._entity_url(entity_id) + '/default' + url = self._entity_url(entity_id) + "/default" response = self.rest.put(url=url, **kwargs) instance = self._create_instance(response=response) return instance - def get_default(self, **kwargs) -> Optional['DefaultResource']: + def get_default(self, **kwargs) -> Optional["DefaultResource"]: """Get default plan if set Args: **kwargs: Optional arguments @@ -458,10 +455,10 @@ def get_default(self, **kwargs) -> Optional['DefaultResource']: class DefaultPlanResource(DefaultResource): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) - def set_default(self, **kwargs) -> 'DefaultStateResource': + def set_default(self, **kwargs) -> "DefaultStateResource": """Set the plan default Args: **kwargs: Optional args @@ -471,7 +468,7 @@ def set_default(self, **kwargs) -> 'DefaultStateResource': @property def is_default(self) -> bool: - return self['default'] is True + return self["default"] is True class DefaultStateClient(DefaultPaginationClient): @@ -485,14 +482,14 @@ def set_state(self, entity_id, state: str, **kwargs): Returns(DefaultStateResource): State resource instance """ log.info(self._log_message("[STATE] Set state ", body=f"[{state}]", args=kwargs)) - url = self._entity_url(entity_id) + '/' + state + url = self._entity_url(entity_id) + "/" + state response = self.rest.put(url=url, **kwargs) instance = self._create_instance(response=response) return instance class DefaultStateResource(DefaultResource): - def set_state(self, state: str, **kwargs) -> 'DefaultStateResource': + def set_state(self, state: str, **kwargs) -> "DefaultStateResource": """Sets the state for the resource Args: state(str): Which state @@ -504,19 +501,19 @@ def set_state(self, state: str, **kwargs) -> 'DefaultStateResource': class DefaultUserResource(DefaultStateResource): - def __init__(self, entity_name='username', **kwargs): + def __init__(self, entity_name="username", **kwargs): super().__init__(entity_name=entity_name, **kwargs) - def suspend(self, **kwargs) -> 'DefaultUserResource': + def suspend(self, **kwargs) -> "DefaultUserResource": """Suspends the user Args: **kwargs: Optional arguments Returns(DefaultUserResource): User instance """ - return self.set_state(state='suspend', **kwargs) + return self.set_state(state="suspend", **kwargs) - def un_suspend(self, **kwargs) -> 'DefaultUserResource': + def un_suspend(self, **kwargs) -> "DefaultUserResource": """Un suspends the user Args: **kwargs: @@ -524,7 +521,7 @@ def un_suspend(self, **kwargs) -> 'DefaultUserResource': Returns(DefaultUserResource): User instance """ - return self.set_state(state='unsuspend', **kwargs) + return self.set_state(state="unsuspend", **kwargs) def resume(self, **kwargs): """Resumes the user @@ -533,7 +530,7 @@ def resume(self, **kwargs): Returns(DefaultUserResource): User instance """ - return self.set_state(state='resume', **kwargs) + return self.set_state(state="resume", **kwargs) def activate(self, **kwargs): """Activates the user @@ -542,7 +539,7 @@ def activate(self, **kwargs): Returns(DefaultUserResource): User instance """ - return self.set_state(state='activate', **kwargs) + return self.set_state(state="activate", **kwargs) def set_as_admin(self, **kwargs): """Promotes the user to admin @@ -551,7 +548,7 @@ def set_as_admin(self, **kwargs): Returns(DefaultUserResource): User instance """ - return self.set_state(state='admin', **kwargs) + return self.set_state(state="admin", **kwargs) def set_as_member(self, **kwargs): """Demotes the user to s member @@ -560,4 +557,4 @@ def set_as_member(self, **kwargs): Returns(DefaultUserResource): User instance """ - return self.set_state(state='member', **kwargs) + return self.set_state(state="member", **kwargs) diff --git a/threescale_api/log_config.py b/threescale_api/log_config.py index 1061190..2ec4015 100644 --- a/threescale_api/log_config.py +++ b/threescale_api/log_config.py @@ -11,12 +11,12 @@ def _to_log_level_map(log_map: dict) -> dict: _LOG_LEVEL_MAP_DEFINITION = { - logging.CRITICAL: ['critical', 'c', 'crit'], - logging.ERROR: ['error', 'e', 'err'], - logging.WARNING: ['warning', 'w', 'warn'], - logging.INFO: ['info', 'i', 'inf'], - logging.DEBUG: ['debug', 'd', 'dbg'], - logging.NOTSET: ['notset', 'n', 'nst'], + logging.CRITICAL: ["critical", "c", "crit"], + logging.ERROR: ["error", "e", "err"], + logging.WARNING: ["warning", "w", "warn"], + logging.INFO: ["info", "i", "inf"], + logging.DEBUG: ["debug", "d", "dbg"], + logging.NOTSET: ["notset", "n", "nst"], } LOG_LEVEL_MAP = _to_log_level_map(_LOG_LEVEL_MAP_DEFINITION) @@ -34,28 +34,17 @@ def to_log_level(level: str, default=logging.NOTSET) -> int: def load_config(level: str = "INFO", handler_level=None, api_level=None, tests_level=None): config = dict( version=1, - formatters={ - 'verbose': { - 'format': - '%(asctime)s %(name)-12s %(levelname)-8s %(message)s' - } - }, + formatters={"verbose": {"format": "%(asctime)s %(name)-12s %(levelname)-8s %(message)s"}}, handlers={ - 'console': { - 'class': 'logging.StreamHandler', - 'formatter': 'verbose', - 'level': to_log_level(handler_level, default=level) + "console": { + "class": "logging.StreamHandler", + "formatter": "verbose", + "level": to_log_level(handler_level, default=level), } }, loggers={ - 'threescale_api': { - 'handlers': ['console'], - 'level': to_log_level(api_level, default=level) - }, - 'tests': { - 'handlers': ['console'], - 'level': to_log_level(handler_level, default=level) - } + "threescale_api": {"handlers": ["console"], "level": to_log_level(api_level, default=level)}, + "tests": {"handlers": ["console"], "level": to_log_level(handler_level, default=level)}, }, ) diff --git a/threescale_api/resources.py b/threescale_api/resources.py index 357e9b5..3f93126 100644 --- a/threescale_api/resources.py +++ b/threescale_api/resources.py @@ -6,180 +6,180 @@ from threescale_api import auth from threescale_api import utils from threescale_api import errors -from threescale_api.defaults import DefaultClient, DefaultPlanClient, DefaultPlanResource, \ - DefaultResource, DefaultStateClient, DefaultUserResource, DefaultStateResource, \ - DefaultPaginationClient +from threescale_api.defaults import ( + DefaultClient, + DefaultPlanClient, + DefaultPlanResource, + DefaultResource, + DefaultStateClient, + DefaultUserResource, + DefaultStateResource, + DefaultPaginationClient, +) from threescale_api import client log = logging.getLogger(__name__) class Services(DefaultPaginationClient): - def __init__(self, *args, entity_name='service', entity_collection='services', - per_page=500, **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + def __init__(self, *args, entity_name="service", entity_collection="services", per_page=500, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/services' + return self.threescale_client.admin_api_url + "/services" class MappingRules(DefaultPaginationClient): - def __init__(self, *args, entity_name='mapping_rule', entity_collection='mapping_rules', - per_page=None, **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, - **kwargs) + def __init__(self, *args, entity_name="mapping_rule", entity_collection="mapping_rules", per_page=None, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.parent.url + '/mapping_rules' + return self.parent.url + "/mapping_rules" class Metrics(DefaultPaginationClient): - def __init__(self, *args, entity_name='metric', entity_collection='metrics', per_page=None, - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + def __init__(self, *args, entity_name="metric", entity_collection="metrics", per_page=None, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.parent.url + '/metrics' + return self.parent.url + "/metrics" class Limits(DefaultClient): - def __init__(self, *args, entity_name='limit', entity_collection='limits', metric=None, - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="limit", entity_collection="limits", metric=None, **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) self._metric = metric @property - def metric(self) -> Union['Metric', 'BackendMetric']: + def metric(self) -> Union["Metric", "BackendMetric"]: return self._metric @property - def application_plan(self) -> 'ApplicationPlan': + def application_plan(self) -> "ApplicationPlan": return self.parent - def __call__(self, metric: 'Metric' = None) -> 'Limits': + def __call__(self, metric: "Metric" = None) -> "Limits": self._metric = metric return self @property def url(self) -> str: - return self.application_plan.plans_url + f'/metrics/{self.metric.entity_id}/limits' + return self.application_plan.plans_url + f"/metrics/{self.metric.entity_id}/limits" def list_per_app_plan(self, **kwargs): log.info("[LIST] List limits per app plan: %s", kwargs) - url = self.parent.url + '/limits' + url = self.parent.url + "/limits" response = self.rest.get(url=url, **kwargs) instance = self._create_instance(response=response) return instance class PricingRules(DefaultClient): - def __init__(self, *args, entity_name='pricing_rule', entity_collection='pricing_rules', - metric: 'Metric' = None, **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__( + self, *args, entity_name="pricing_rule", entity_collection="pricing_rules", metric: "Metric" = None, **kwargs + ): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) self._metric = metric @property - def metric(self) -> 'Metric': + def metric(self) -> "Metric": return self._metric @property - def application_plan(self) -> 'ApplicationPlan': + def application_plan(self) -> "ApplicationPlan": return self.parent - def __call__(self, metric: 'Metric' = None) -> 'PricingRules': + def __call__(self, metric: "Metric" = None) -> "PricingRules": self._metric = metric return self @property def url(self) -> str: - return self.application_plan.plans_url + f'/metrics/{self.metric.entity_id}/pricing_rules' + return self.application_plan.plans_url + f"/metrics/{self.metric.entity_id}/pricing_rules" class Methods(DefaultPaginationClient): - def __init__(self, *args, entity_name='method', entity_collection='methods', per_page=None, - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + def __init__(self, *args, entity_name="method", entity_collection="methods", per_page=None, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.parent.url + '/methods' + return self.parent.url + "/methods" class BackendMethods(Methods): - def __init__(self, *args, entity_name='method', entity_collection='methods', per_page=None, - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + def __init__(self, *args, entity_name="method", entity_collection="methods", per_page=None, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) class ApplicationPlans(DefaultPlanClient): - def __init__(self, *args, entity_name='application_plan', entity_collection='plans', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="application_plan", entity_collection="plans", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: if type(self.parent) is Service: - return self.parent.url + '/application_plans' - return self.threescale_client.admin_api_url + '/application_plans' + return self.parent.url + "/application_plans" + return self.threescale_client.admin_api_url + "/application_plans" @property def plans_url(self) -> str: - return self.threescale_client.admin_api_url + '/application_plans' + return self.threescale_client.admin_api_url + "/application_plans" class ApplicationPlanFeatures(DefaultClient): - def __init__(self, *args, entity_name='feature', entity_collection='features', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="feature", entity_collection="features", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.parent.url + '/features' + return self.parent.url + "/features" class AccountUsers(DefaultStateClient): - def __init__(self, *args, entity_name='user', entity_collection='users', per_page=None, - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + def __init__(self, *args, entity_name="user", entity_collection="users", per_page=None, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.parent.url + '/users' + return self.parent.url + "/users" class AccountPlans(DefaultPlanClient): - def __init__(self, *args, entity_name='account_plan', entity_collection='plans', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="account_plan", entity_collection="plans", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/account_plans' + return self.threescale_client.admin_api_url + "/account_plans" class Accounts(DefaultStateClient): - def __init__(self, *args, entity_name='account', entity_collection='accounts', per_page=500, - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, - **kwargs) + def __init__(self, *args, entity_name="account", entity_collection="accounts", per_page=500, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/accounts' + return self.threescale_client.admin_api_url + "/accounts" - def create(self, params: dict = None, **kwargs) -> 'Account': + def create(self, params: dict = None, **kwargs) -> "Account": """Create new account Args: params(dict): Parameters to used to create new instance @@ -188,7 +188,7 @@ def create(self, params: dict = None, **kwargs) -> 'Account': """ return self.signup(params=params, **kwargs) - def find(self, params: dict, **kwargs) -> 'Account': + def find(self, params: dict, **kwargs) -> "Account": """Find an account Args: params(dict): Parameters to used to find account (name, email, etc) @@ -201,7 +201,7 @@ def find(self, params: dict, **kwargs) -> 'Account': instance = self._create_instance(response=response) return instance - def signup(self, params: dict, **kwargs) -> 'Account': + def signup(self, params: dict, **kwargs) -> "Account": """Sign Up for an account Args: params(dict): Parameters to used to create new instance @@ -209,7 +209,7 @@ def signup(self, params: dict, **kwargs) -> 'Account': Returns(Account): Account instance """ log.info("[SIGNUP] Create new Signup: params=%s, kwargs=%s", params, kwargs) - url = self.threescale_client.admin_api_url + '/signup' + url = self.threescale_client.admin_api_url + "/signup" response = self.rest.post(url=url, json=params, **kwargs) instance = self._create_instance(response=response) return instance @@ -224,7 +224,7 @@ def set_plan(self, entity_id: int, plan_id: int, **kwargs): """ log.info("[PLAN] Set plan for an account(%s): %s", entity_id, plan_id) params = dict(plan_id=plan_id) - url = self._entity_url(entity_id=entity_id) + '/change_plan' + url = self._entity_url(entity_id=entity_id) + "/change_plan" response = self.rest.put(url=url, json=params, **kwargs) instance = self._create_instance(response=response) return instance @@ -241,49 +241,47 @@ def send_message(self, entity_id: int, body: str, subject: str = None, **kwargs) params = dict(body=body) if subject: params["subject"] = subject - url = self._entity_url(entity_id=entity_id) + '/messages' + url = self._entity_url(entity_id=entity_id) + "/messages" response = self.rest.post(url=url, json=params, **kwargs) instance = utils.extract_response(response=response) return instance - def approve(self, entity_id: int, **kwargs) -> 'Account': + def approve(self, entity_id: int, **kwargs) -> "Account": """Approve the account Args: entity_id(int): Entity id **kwargs: Optional args Returns(Account): Account resource """ - return self.set_state(entity_id=entity_id, state='approve', **kwargs) + return self.set_state(entity_id=entity_id, state="approve", **kwargs) - def reject(self, entity_id, **kwargs) -> 'Account': + def reject(self, entity_id, **kwargs) -> "Account": """Reject the account Args: entity_id(int): Entity id **kwargs: Optional args Returns(Account): Account resource """ - return self.set_state(entity_id=entity_id, state='reject', **kwargs) + return self.set_state(entity_id=entity_id, state="reject", **kwargs) - def pending(self, entity_id, **kwargs) -> 'Account': + def pending(self, entity_id, **kwargs) -> "Account": """Set the account as pending Args: entity_id(int): Entity id **kwargs: Optional args Returns(Account): Account resource """ - return self.set_state(entity_id=entity_id, state='make_pending', **kwargs) + return self.set_state(entity_id=entity_id, state="make_pending", **kwargs) class ServiceSubscriptions(DefaultClient): - def __init__(self, *args, entity_name='service_subscription', - entity_collection='service_subscriptions', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="service_subscription", entity_collection="service_subscriptions", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.parent.url + '/service_subscriptions' + return self.parent.url + "/service_subscriptions" def approve(self, entity_id: int, **kwargs): url = self.url + f"/{entity_id}/approve" @@ -301,14 +299,12 @@ def change_plan(self, entity_id: int, plan_id: int, **kwargs): class ServicePlans(DefaultClient): - def __init__(self, *args, entity_name='service_plan', - entity_collection='service_plans', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="service_plan", entity_collection="service_plans", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.parent.url + '/service_plans' + return self.parent.url + "/service_plans" def service_plan_set_default(self, entity_id: int, **kwargs): url = self.url + f"/{entity_id}/default" @@ -318,80 +314,76 @@ def service_plan_set_default(self, entity_id: int, **kwargs): class Applications(DefaultStateClient): - def __init__(self, *args, entity_name='application', entity_collection='applications', - per_page=None, **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + def __init__(self, *args, entity_name="application", entity_collection="applications", per_page=None, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.parent.url + '/applications' + return self.parent.url + "/applications" def change_plan(self, entity_id: int, plan_id: int, **kwargs): log.info("[PLAN] Change plan for application (%s) to %s %s", entity_id, plan_id, kwargs) params = dict(plan_id=plan_id) - url = self._entity_url(entity_id=entity_id) + '/change_plan' + url = self._entity_url(entity_id=entity_id) + "/change_plan" response = self.rest.put(url=url, json=params, **kwargs) instance = utils.extract_response(response=response) return instance def customize_plan(self, entity_id: int, **kwargs): log.info("[PLAN] Customize plan for application (%s) %s", entity_id, kwargs) - url = self._entity_url(entity_id=entity_id) + '/customize_plan' + url = self._entity_url(entity_id=entity_id) + "/customize_plan" response = self.rest.put(url=url, **kwargs) instance = utils.extract_response(response=response) return instance def decustomize_plan(self, entity_id: int, **kwargs): log.info("[PLAN] Decustomize plan for application (%s) %s", entity_id, kwargs) - url = self._entity_url(entity_id=entity_id) + '/decustomize_plan' + url = self._entity_url(entity_id=entity_id) + "/decustomize_plan" response = self.rest.put(url=url, **kwargs) instance = utils.extract_response(response=response) return instance def accept(self, entity_id: int, **kwargs): - self.set_state(entity_id=entity_id, state='accept', **kwargs) + self.set_state(entity_id=entity_id, state="accept", **kwargs) def suspend(self, entity_id: int, **kwargs): - self.set_state(entity_id=entity_id, state='suspend', **kwargs) + self.set_state(entity_id=entity_id, state="suspend", **kwargs) def resume(self, entity_id: int, **kwargs): - self.set_state(entity_id=entity_id, state='resume', **kwargs) + self.set_state(entity_id=entity_id, state="resume", **kwargs) class DevPortalAuthProviders(DefaultClient): - def __init__(self, *args, entity_name='authentication_provider', - entity_collection='authentication_providers', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__( + self, *args, entity_name="authentication_provider", entity_collection="authentication_providers", **kwargs + ): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/authentication_providers' + return self.threescale_client.admin_api_url + "/authentication_providers" class ApplicationReferrerFilters(DefaultClient): - def __init__(self, *args, entity_name='application', entity_collection='applications', - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="application", entity_collection="applications", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.parent.url + '/referrer_filters' + return self.parent.url + "/referrer_filters" class ApplicationKeys(DefaultClient): - def __init__(self, *args, entity_name='key', entity_collection='keys', - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="key", entity_collection="keys", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.parent.url + '/keys' + return self.parent.url + "/keys" - def create(self, params: dict = None, **kwargs) -> 'ApplicationKey': + def create(self, params: dict = None, **kwargs) -> "ApplicationKey": """Create a new instance of ApplicationKey. "keys" POST request returns Application instead of newly create key. Returns: Newly created key. @@ -402,7 +394,7 @@ def create(self, params: dict = None, **kwargs) -> 'ApplicationKey': key.entity_id = quote_plus(key["value"]) return key - def list(self, **kwargs) -> List['ApplicationKey']: + def list(self, **kwargs) -> List["ApplicationKey"]: """List all entities of ApplicationKey. There is no id in list response, so it needs to be assigned the value to be able to work with key instance. @@ -417,76 +409,64 @@ def list(self, **kwargs) -> List['ApplicationKey']: class Providers(DefaultClient): - def __init__(self, *args, entity_name='user', entity_collection='users', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="user", entity_collection="users", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/providers' + return self.threescale_client.admin_api_url + "/providers" def create_token(self, entity_id: int, params, **kwargs): - log.info(self._log_message("[TOKEN] Create token", - entity_id=entity_id, body=params, **kwargs)) - url = self._entity_url(entity_id=entity_id) + '/access_tokens' + log.info(self._log_message("[TOKEN] Create token", entity_id=entity_id, body=params, **kwargs)) + url = self._entity_url(entity_id=entity_id) + "/access_tokens" response = self.rest.put(url, json=params) return utils.extract_response(response=response) class AccessTokens(DefaultClient): - def __init__(self, *args, entity_name='access_token', entity_collection='access_tokens', - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="access_token", entity_collection="access_tokens", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/personal/access_tokens' + return self.threescale_client.admin_api_url + "/personal/access_tokens" class ActiveDocs(DefaultClient): - def __init__(self, *args, entity_name='api_doc', entity_collection='api_docs', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="api_doc", entity_collection="api_docs", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/active_docs' + return self.threescale_client.admin_api_url + "/active_docs" class Analytics(DefaultClient): - def _list_by_resource(self, resource_id: int, resource_type, metric_name: str = 'hits', - since=None, period: str = 'year', **kwargs): - log.info("List analytics by %s (%s) for metric (#%s)", resource_type, resource_id, - metric_name) - params = dict( - metric_name=metric_name, - since=since, - period=period, - **kwargs - ) + def _list_by_resource( + self, resource_id: int, resource_type, metric_name: str = "hits", since=None, period: str = "year", **kwargs + ): + log.info("List analytics by %s (%s) for metric (#%s)", resource_type, resource_id, metric_name) + params = dict(metric_name=metric_name, since=since, period=period, **kwargs) url = self.threescale_client.url + f"/stats/{resource_type}/{resource_id}/usage" response = self.rest.get(url, json=params) return utils.extract_response(response=response) - def list_by_application(self, application: Union['Application', int], **kwargs): + def list_by_application(self, application: Union["Application", int], **kwargs): app_id = _extract_entity_id(application) - return self._list_by_resource(resource_id=app_id, resource_type='applications', **kwargs) + return self._list_by_resource(resource_id=app_id, resource_type="applications", **kwargs) - def list_by_service(self, service: Union['Service', int], **kwargs): + def list_by_service(self, service: Union["Service", int], **kwargs): app_id = _extract_entity_id(service) - return self._list_by_resource(resource_id=app_id, resource_type='services', **kwargs) + return self._list_by_resource(resource_id=app_id, resource_type="services", **kwargs) - def list_by_backend(self, backend: Union['Backend', int], **kwargs): + def list_by_backend(self, backend: Union["Backend", int], **kwargs): backend_id = _extract_entity_id(backend) - return self._list_by_resource( - resource_id=backend_id, resource_type='backend_apis', **kwargs) + return self._list_by_resource(resource_id=backend_id, resource_type="backend_apis", **kwargs) class Tenants(DefaultClient): - def __init__(self, *args, entity_name='tenant', entity_collection='tenants', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="tenant", entity_collection="tenants", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) def read(self, entity_id, **kwargs): log.debug(self._log_message("[GET] Read Tenant", args=kwargs)) @@ -497,9 +477,9 @@ def read(self, entity_id, **kwargs): @property def url(self) -> str: - return self.threescale_client.master_api_url + '/providers' + return self.threescale_client.master_api_url + "/providers" - def trigger_billing(self, tenant: Union['Tenant', int], date: str): + def trigger_billing(self, tenant: Union["Tenant", int], date: str): """Trigger billing for whole tenant Args: tenant: Tenant id or tenant resource @@ -513,8 +493,7 @@ def trigger_billing(self, tenant: Union['Tenant', int], date: str): response = self.rest.post(url=url, json=params) return response.ok - def trigger_billing_account(self, tenant: Union['Tenant', int], account: Union['Account', int], - date: str) -> dict: + def trigger_billing_account(self, tenant: Union["Tenant", int], account: Union["Account", int], date: str) -> dict: """Trigger billing for one account in tenant Args: tenant: Tenant id or tenant resource @@ -532,47 +511,45 @@ def trigger_billing_account(self, tenant: Union['Tenant', int], account: Union[' class Proxies(DefaultClient): - def __init__(self, *args, entity_name='proxy', **kwargs): + def __init__(self, *args, entity_name="proxy", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) @property def url(self) -> str: - return self.parent.url + '/proxy' + return self.parent.url + "/proxy" - def deploy(self) -> 'Proxy': + def deploy(self) -> "Proxy": log.info("[DEPLOY] %s to Staging", self._entity_name) - url = f'{self.url}/deploy' + url = f"{self.url}/deploy" response = self.rest.post(url) instance = self._create_instance(response=response) return instance @property - def oidc(self) -> 'OIDCConfigs': + def oidc(self) -> "OIDCConfigs": return OIDCConfigs(self) @property - def mapping_rules(self) -> 'MappingRules': + def mapping_rules(self) -> "MappingRules": return MappingRules(parent=self, instance_klass=MappingRule) class ProxyConfigs(DefaultClient): - def __init__(self, *args, entity_name='proxy_config', entity_collection='configs', - env: str = None, **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="proxy_config", entity_collection="configs", env: str = None, **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) self._env = env @property def url(self) -> str: - base = self.parent.url + '/configs' + base = self.parent.url + "/configs" return base if not self._env else f"{base}/{self._env}" @property - def proxy(self) -> 'Proxy': + def proxy(self) -> "Proxy": return self.parent @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.proxy.service # tests/integration/test_integration_services.py::test_service_list_configs @@ -582,70 +559,67 @@ def service(self) -> 'Service': def list(self, **kwargs): if "env" in kwargs: self._env = kwargs["env"] - del (kwargs["env"]) + del kwargs["env"] return super().list(**kwargs) - def promote(self, version: int = 1, from_env: str = 'sandbox', to_env: str = 'production', - **kwargs) -> 'Proxy': - log.info("[PROMOTE] %s version %s from %s to %s", self.service, version, from_env, - to_env) - url = f'{self.url}/{from_env}/{version}/promote' + def promote(self, version: int = 1, from_env: str = "sandbox", to_env: str = "production", **kwargs) -> "Proxy": + log.info("[PROMOTE] %s version %s from %s to %s", self.service, version, from_env, to_env) + url = f"{self.url}/{from_env}/{version}/promote" params = dict(to=to_env) kwargs.update() response = self.rest.post(url, json=params, **kwargs) instance = self._create_instance(response=response) return instance - def latest(self, env: str = "sandbox") -> 'ProxyConfig': + def latest(self, env: str = "sandbox") -> "ProxyConfig": log.info("[LATEST] Get latest proxy configuration of %s", env) self._env = env - url = self.url + '/latest' + url = self.url + "/latest" response = self.rest.get(url=url) instance = self._create_instance(response=response) return instance - def version(self, version: int = 1, env: str = "sandbox") -> 'ProxyConfig': + def version(self, version: int = 1, env: str = "sandbox") -> "ProxyConfig": log.info("[VERSION] Get proxy configuration of %s of version %s", env, version) self._env = env - url = f'{self.url}/{version}' + url = f"{self.url}/{version}" response = self.rest.get(url=url) instance = self._create_instance(response=response) return instance class SettingsClient(DefaultClient): - def __init__(self, *args, entity_name='settings', **kwargs): + def __init__(self, *args, entity_name="settings", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/settings' + return self.threescale_client.admin_api_url + "/settings" class AdminPortalAuthProviders(DefaultClient): - def __init__(self, *args, entity_name='authentication_provider', - entity_collection='authentication_providers', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__( + self, *args, entity_name="authentication_provider", entity_collection="authentication_providers", **kwargs + ): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/account/authentication_providers' + return self.threescale_client.admin_api_url + "/account/authentication_providers" class UserPermissionsClient(DefaultClient): - def __init__(self, *args, entity_name='permissions', **kwargs): + def __init__(self, *args, entity_name="permissions", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/tenants' + return self.threescale_client.admin_api_url + "/tenants" class Policies(DefaultClient): - def __init__(self, *args, entity_name='policy', entity_collection='policies', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="policy", entity_collection="policies", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: @@ -659,7 +633,7 @@ def append(self, *policies): def insert(self, index: int, *policies): params = self.list().entity - for (i, policy) in enumerate(policies): + for i, policy in enumerate(policies): params["policies_config"].insert(index + i, policy) params["service_id"] = self.parent["service_id"] return self.update(params=params) @@ -668,7 +642,7 @@ def insert(self, index: int, *policies): class OIDCConfigs(DefaultClient): @property def url(self) -> str: - return self.parent.url + '/oidc_configuration' + return self.parent.url + "/oidc_configuration" def update(self, params: dict = None, **kwargs) -> dict: return self.rest.patch(url=self.url, json=params, **kwargs).json() @@ -678,63 +652,59 @@ def read(self, params: dict = None, **kwargs) -> dict: class Backends(DefaultPaginationClient): - def __init__(self, *args, entity_name='backend_api', - entity_collection='backend_apis', per_page=500, **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + def __init__(self, *args, entity_name="backend_api", entity_collection="backend_apis", per_page=500, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/backend_apis' + return self.threescale_client.admin_api_url + "/backend_apis" class BackendMetrics(Metrics): - def __init__(self, *args, entity_name='metric', entity_collection='metrics', per_page=500, - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, - **kwargs) + def __init__(self, *args, entity_name="metric", entity_collection="metrics", per_page=500, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) class BackendMappingRules(MappingRules): - def __init__(self, *args, entity_name='mapping_rule', - entity_collection='mapping_rules', per_page=500, **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, - **kwargs) + def __init__(self, *args, entity_name="mapping_rule", entity_collection="mapping_rules", per_page=500, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) class BackendUsages(Services): - def __init__(self, *args, entity_name='backend_usage', - entity_collection='backend_usages', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="backend_usage", entity_collection="backend_usages", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.parent.url + '/backend_usages' + return self.parent.url + "/backend_usages" class PoliciesRegistry(DefaultClient): - def __init__(self, *args, entity_name='policy', entity_collection='policies', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="policy", entity_collection="policies", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/registry/policies' + return self.threescale_client.admin_api_url + "/registry/policies" class ProviderAccounts(DefaultClient): """ 3scale endpoints implement only GET and UPDATE methods """ - def __init__(self, *args, entity_name='account', **kwargs): + + def __init__(self, *args, entity_name="account", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/provider' + return self.threescale_client.admin_api_url + "/provider" def fetch(self, **kwargs) -> DefaultResource: """ @@ -742,13 +712,12 @@ def fetch(self, **kwargs) -> DefaultResource: Only one Provider Account (currently used Tenant) is reachable via admin_api_url, therefore `entity_id` is not required. """ - log.debug(self._log_message("[FETCH] Fetch Current Provider Account (Tenant) ", - args=kwargs)) + log.debug(self._log_message("[FETCH] Fetch Current Provider Account (Tenant) ", args=kwargs)) response = self.rest.get(url=self.url, **kwargs) instance = self._create_instance(response=response) return instance - def update(self, params: dict = None, **kwargs) -> 'DefaultResource': + def update(self, params: dict = None, **kwargs) -> "DefaultResource": return super().update(params=params) @@ -757,60 +726,60 @@ class ProviderAccountUsers(DefaultStateClient): Client for Provider Accounts. In 3scale, entity under Account Settings > Users """ - def __init__(self, *args, entity_name='user', entity_collection='users', per_page=None, - **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + + def __init__(self, *args, entity_name="user", entity_collection="users", per_page=None, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/users' + return self.threescale_client.admin_api_url + "/users" - def permissions_update(self, entity_id: int, - allowed_services: [] = None, allowed_sections: [] = None, **kwargs): - allowed_services = allowed_services if allowed_services else ['[]'] - allowed_sections = allowed_sections if allowed_sections else ['[]'] + def permissions_update(self, entity_id: int, allowed_services: [] = None, allowed_sections: [] = None, **kwargs): + allowed_services = allowed_services if allowed_services else ["[]"] + allowed_sections = allowed_sections if allowed_sections else ["[]"] log.info(self._log_message("Change of Provider Account (User) permissions")) - url = self._entity_url(entity_id) + '/permissions' + url = self._entity_url(entity_id) + "/permissions" params = { - 'allowed_service_ids[]': allowed_services, - 'allowed_sections[]': allowed_sections, + "allowed_service_ids[]": allowed_services, + "allowed_sections[]": allowed_sections, } response = self.rest.put(url=url, data=params, **kwargs) return response.json() def allow_all_sections(self, entity_id: int, **kwargs): - log.info(self._log_message("Change of Provider Account (User) " - "permissions to all available permissions")) - return self.permissions_update(entity_id=entity_id, allowed_sections=[ - 'portal', 'finance', 'settings', 'partners', 'monitoring', 'plans', 'policy_registry' - ]) + log.info(self._log_message("Change of Provider Account (User) " "permissions to all available permissions")) + return self.permissions_update( + entity_id=entity_id, + allowed_sections=["portal", "finance", "settings", "partners", "monitoring", "plans", "policy_registry"], + ) def permissions_read(self, entity_id: int, **kwargs): - url = self._entity_url(entity_id) + '/permissions' + url = self._entity_url(entity_id) + "/permissions" response = self.rest.get(url=url, **kwargs) return response.json() def set_role_member(self, entity_id: int): log.info("Changes the role of the user of the provider account to member") - return self.set_state(entity_id, state='member') + return self.set_state(entity_id, state="member") def set_role_admin(self, entity_id: int): log.info("Changes the role of the provider account to admin") - return self.set_state(entity_id, state='admin') + return self.set_state(entity_id, state="admin") def suspend(self, entity_id): log.info("Changes the state of the user of the provider account to suspended") - return self.set_state(entity_id, state='suspend') + return self.set_state(entity_id, state="suspend") def unsuspend(self, entity_id: int): log.info("Revokes the suspension of a user of the provider account") - return self.set_state(entity_id, state='unsuspend') + return self.set_state(entity_id, state="unsuspend") def activate(self, entity_id: int): log.info("Changes the state of the user of the provider account to active") - return self.set_state(entity_id, state='activate') + return self.set_state(entity_id, state="activate") class Webhooks(DefaultClient): @@ -818,13 +787,12 @@ class Webhooks(DefaultClient): Default client for webhooks """ - def __init__(self, *args, entity_name='webhook', entity_collection='webhooks', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="webhook", entity_collection="webhooks", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/webhooks' + return self.threescale_client.admin_api_url + "/webhooks" def update(self, params: dict = None, **kwargs): url = self.url @@ -834,37 +802,37 @@ def setup(self, webhook_type, url): """ Configure webhooks for given webhooks type """ - params = {"url": url, - "active": "true", - "provider_actions": "true"} + params = {"url": url, "active": "true", "provider_actions": "true"} if webhook_type == "Keys": - params.update({ - "application_key_created_on": "true", - "application_key_deleted_on": "true", - "application_key_updated_on": "true" - }) + params.update( + { + "application_key_created_on": "true", + "application_key_deleted_on": "true", + "application_key_updated_on": "true", + } + ) elif webhook_type == "Users": - params.update({ - "user_created_on": "true", - "user_updated_on": "true", - "user_deleted_on": "true" - }) + params.update({"user_created_on": "true", "user_updated_on": "true", "user_deleted_on": "true"}) elif webhook_type == "Applications": - params.update({ - "application_created_on": "true", - "application_updated_on": "true", - "application_suspended_on": "true", - "application_plan_changed_on": "true", - "application_user_key_updated_on": "true", - "application_deleted_on": "true" - }) + params.update( + { + "application_created_on": "true", + "application_updated_on": "true", + "application_suspended_on": "true", + "application_plan_changed_on": "true", + "application_user_key_updated_on": "true", + "application_deleted_on": "true", + } + ) elif webhook_type == "Accounts": - params.update({ - "account_created_on": "true", - "account_updated_on": "true", - "account_deleted_on": "true", - "account_plan_changed_on": "true" - }) + params.update( + { + "account_created_on": "true", + "account_updated_on": "true", + "account_deleted_on": "true", + "account_plan_changed_on": "true", + } + ) return self.update(params=params) @@ -872,38 +840,39 @@ def clear(self): """ Configure webhooks to default settings """ - params = {"url": "", - "active": "false", - "provider_actions": "false", - "account_created_on": "false", - "account_updated_on": "false", - "account_deleted_on": "false", - "user_created_on": "false", - "user_updated_on": "false", - "user_deleted_on": "false", - "application_created_on": "false", - "application_updated_on": "false", - "application_deleted_on": "false", - "account_plan_changed_on": "false", - "application_plan_changed_on": "false", - "application_user_key_updated_on": "false", - "application_key_created_on": "false", - "application_key_deleted_on": "false", - "application_suspended_on": "false", - "application_key_updated_on": "false", - } + params = { + "url": "", + "active": "false", + "provider_actions": "false", + "account_created_on": "false", + "account_updated_on": "false", + "account_deleted_on": "false", + "user_created_on": "false", + "user_updated_on": "false", + "user_deleted_on": "false", + "application_created_on": "false", + "application_updated_on": "false", + "application_deleted_on": "false", + "account_plan_changed_on": "false", + "application_plan_changed_on": "false", + "application_user_key_updated_on": "false", + "application_key_created_on": "false", + "application_key_deleted_on": "false", + "application_suspended_on": "false", + "application_key_updated_on": "false", + } return self.update(params=params) class LineItems(DefaultClient): """Default client for LineItems""" - def __init__(self, *args, entity_name='line_item', entity_collection='line_items', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + + def __init__(self, *args, entity_name="line_item", entity_collection="line_items", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.parent.url + '/line_items' + return self.parent.url + "/line_items" class InvoiceState(Enum): @@ -918,27 +887,28 @@ class InvoiceState(Enum): class Invoices(DefaultPaginationClient): """Default client for Invoices""" - def __init__(self, *args, entity_name='invoice', entity_collection='invoices', - per_page=20, **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, per_page=per_page, **kwargs) + + def __init__(self, *args, entity_name="invoice", entity_collection="invoices", per_page=20, **kwargs): + super().__init__( + *args, entity_name=entity_name, entity_collection=entity_collection, per_page=per_page, **kwargs + ) @property def url(self) -> str: - return self.threescale_client.url + '/api/invoices' + return self.threescale_client.url + "/api/invoices" @property def line_items(self) -> LineItems: return LineItems(parent=self, instance_klass=LineItem) - def list_by_account(self, account: Union['Account', int], **kwargs): + def list_by_account(self, account: Union["Account", int], **kwargs): account_id = _extract_entity_id(account) url = self.threescale_client.url + f"/api/accounts/{account_id}/invoices" response = self.rest.get(url, **kwargs) instance = self._create_instance(response=response, collection=True) return instance - def read_by_account(self, entity_id: int, account: Union['Account', int], **kwargs): + def read_by_account(self, entity_id: int, account: Union["Account", int], **kwargs): account_id = _extract_entity_id(account) url = self.threescale_client.url + f"/api/accounts/{account_id}/invoices/{entity_id}" response = self.rest.get(url, **kwargs) @@ -953,7 +923,7 @@ def state_update(self, entity_id: int, state: InvoiceState, **kwargs): """ log.info("[Invoice] state changed for invoice (%s): %s", entity_id, state) params = dict(state=state.value) - url = self._entity_url(entity_id) + '/state' + url = self._entity_url(entity_id) + "/state" response = self.rest.put(url=url, json=params, **kwargs) instance = self._create_instance(response=response) return instance @@ -961,36 +931,33 @@ def state_update(self, entity_id: int, state: InvoiceState, **kwargs): def charge(self, entity_id: int): """Charge an Invoice.""" log.info("[Invoice] charge invoice (%s)", entity_id) - url = self._entity_url(entity_id) + '/charge' + url = self._entity_url(entity_id) + "/charge" response = self.rest.post(url) instance = self._create_instance(response=response) return instance class PaymentTransactions(DefaultClient): - def __init__(self, *args, entity_name='payment_transaction', - entity_collection='payment_transactions', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="payment_transaction", entity_collection="payment_transactions", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.parent.url + '/payment_transactions' + return self.parent.url + "/payment_transactions" class FieldsDefinitions(DefaultClient): - def __init__(self, *args, entity_name='fields_definition', - entity_collection='fields_definitions', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + def __init__(self, *args, entity_name="fields_definition", entity_collection="fields_definitions", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/fields_definitions' + return self.threescale_client.admin_api_url + "/fields_definitions" class CmsClient(DefaultPaginationClient): - """ Client for all cms api endpoints. """ + """Client for all cms api endpoints.""" + def __init__(self, *args, per_page=100, **kwargs): super().__init__(*args, per_page=per_page, **kwargs) @@ -1021,61 +988,63 @@ def select_by(self, **params): filters = {fil: params.pop(fil) for fil in self.FILTERS if fil in params} def predicate(item): - for (key, val) in params.items(): + for key, val in params.items(): if item[key] != val: return False return True + if filters: return self.select(predicate=predicate, params=filters) return self.select(predicate=predicate) class CmsFiles(CmsClient): - FILTERS = ['parent_id'] + FILTERS = ["parent_id"] """ Client for files. """ - def __init__(self, *args, entity_name='file', entity_collection='collection', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + + def __init__(self, *args, entity_name="file", entity_collection="collection", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/cms/files' + return self.threescale_client.admin_api_url + "/cms/files" class CmsSections(CmsClient): - FILTERS = ['section_id'] + FILTERS = ["section_id"] """ Client for sections. """ - def __init__(self, *args, entity_name='section', entity_collection='collection', **kwargs): - super().__init__(*args, entity_name=entity_name, - entity_collection=entity_collection, **kwargs) + + def __init__(self, *args, entity_name="section", entity_collection="collection", **kwargs): + super().__init__(*args, entity_name=entity_name, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/cms/sections' + return self.threescale_client.admin_api_url + "/cms/sections" class CmsTemplates(CmsClient): - FILTERS = ['type'] # , 'section_id'] + FILTERS = ["type"] # , 'section_id'] """ Client for templates. """ - def __init__(self, *args, entity_collection='collection', **kwargs): + + def __init__(self, *args, entity_collection="collection", **kwargs): super().__init__(*args, entity_collection=entity_collection, **kwargs) @property def url(self) -> str: - return self.threescale_client.admin_api_url + '/cms/templates' + return self.threescale_client.admin_api_url + "/cms/templates" def publish(self, entity_id, **kwargs): - """ Publish template with entity_id """ + """Publish template with entity_id""" log.info("[PUBLISH] %s", entity_id) - url = self._entity_url(entity_id) + '/publish' + url = self._entity_url(entity_id) + "/publish" response = self.rest.put(url=url, **kwargs) instance = self._create_instance(response=response) return instance - def list(self, **kwargs) -> List['DefaultResource']: + def list(self, **kwargs) -> List["DefaultResource"]: """List all entities Args: **kwargs: Optional parameters @@ -1088,7 +1057,7 @@ def list(self, **kwargs) -> List['DefaultResource']: instance = self._list(**kwargs) return instance - def select(self, predicate, **kwargs) -> List['DefaultResource']: + def select(self, predicate, **kwargs) -> List["DefaultResource"]: """Select resource s based on the predicate Args: predicate: Predicate @@ -1100,45 +1069,51 @@ def select(self, predicate, **kwargs) -> List['DefaultResource']: kwargs["params"].setdefault("type", self._entity_name) return [item for item in self._list(**kwargs) if predicate(item)] - def create(self, params: dict = None, - *args, **kwargs) -> 'DefaultResource': - params.update({'type': self._entity_name}) + def create(self, params: dict = None, *args, **kwargs) -> "DefaultResource": + params.update({"type": self._entity_name}) return super().create(params=params, *args, **kwargs) class CmsPages(CmsTemplates): - """ Client for pages """ - def __init__(self, *args, entity_name='page', **kwargs): + """Client for pages""" + + def __init__(self, *args, entity_name="page", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) class CmsBuiltinPages(CmsTemplates): - """ Client for builtin pages. """ - def __init__(self, *args, entity_name='builtin_page', **kwargs): + """Client for builtin pages.""" + + def __init__(self, *args, entity_name="builtin_page", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) class CmsLayouts(CmsTemplates): - """ Client for layouts """ - def __init__(self, *args, entity_name='layout', **kwargs): + """Client for layouts""" + + def __init__(self, *args, entity_name="layout", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) class CmsPartials(CmsTemplates): - """ Client for partials """ - def __init__(self, *args, entity_name='partial', **kwargs): + """Client for partials""" + + def __init__(self, *args, entity_name="partial", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) class CmsBuiltinPartials(CmsTemplates): - """ Client for builtin partials """ - def __init__(self, *args, entity_name='builtin_partial', **kwargs): + """Client for builtin partials""" + + def __init__(self, *args, entity_name="builtin_partial", **kwargs): super().__init__(*args, entity_name=entity_name, **kwargs) + + # Resources class ApplicationPlan(DefaultPlanResource): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property @@ -1146,39 +1121,39 @@ def plans_url(self) -> str: return self.threescale_client.admin_api_url + f"/application_plans/{self.entity_id}" @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.parent - def limits(self, metric: 'Metric' = None) -> 'Limits': + def limits(self, metric: "Metric" = None) -> "Limits": return Limits(self, metric=metric, instance_klass=Limit) - def pricing_rules(self, metric: 'Metric' = None) -> 'PricingRules': + def pricing_rules(self, metric: "Metric" = None) -> "PricingRules": return PricingRules(self, metric=metric, instance_klass=PricingRule) class Method(DefaultResource): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property - def metric(self) -> 'Metric': + def metric(self) -> "Metric": return self.parent @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.metric.parent class BackendMethod(Method): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property - def service(self) -> 'Service': + def service(self) -> "Service": raise AttributeError("'BackendMethod' object has no attribute 'service'") @property - def backend(self) -> 'Backend': + def backend(self) -> "Backend": return self.metric.parent @@ -1195,35 +1170,35 @@ def change_plan(self, **kwargs): class Metric(DefaultResource): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.parent @property - def methods(self) -> 'Methods': + def methods(self) -> "Methods": return Methods(parent=self, instance_klass=Method, per_page=self.client.per_page) class MappingRule(DefaultResource): @property - def proxy(self) -> 'Proxy': + def proxy(self) -> "Proxy": return self.parent @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.proxy.service class ProxyConfig(DefaultResource): @property - def proxy(self) -> 'Proxy': + def proxy(self) -> "Proxy": return self.parent @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.proxy.service # ProxyConfig is once instantiated with just proxy config obj (for example @@ -1244,15 +1219,15 @@ def __len__(self): class Policy(DefaultResource): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property - def proxy(self) -> 'Proxy': + def proxy(self) -> "Proxy": return self.parent @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.proxy.service @@ -1262,7 +1237,7 @@ def url(self) -> str: return self.client.url @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.parent @property @@ -1270,21 +1245,21 @@ def mapping_rules(self) -> MappingRules: return MappingRules(parent=self, instance_klass=MappingRule) @property - def configs(self) -> 'ProxyConfigs': + def configs(self) -> "ProxyConfigs": return ProxyConfigs(parent=self, instance_klass=ProxyConfig) @property - def policies(self) -> 'Policies': + def policies(self) -> "Policies": return Policies(parent=self, instance_klass=Policy) - def promote(self, **kwargs) -> 'Proxy': + def promote(self, **kwargs) -> "Proxy": return self.configs.promote(**kwargs) @property def policies_registry(self) -> PoliciesRegistry: return PoliciesRegistry(parent=self, instance_klass=PolicyRegistry) - def deploy(self) -> 'Proxy': + def deploy(self) -> "Proxy": return self.client.deploy() @@ -1293,7 +1268,7 @@ class Service(DefaultResource): AUTH_APP_ID_KEY = "2" AUTH_OIDC = "oidc" - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property @@ -1309,33 +1284,33 @@ def service_plans(self) -> ServicePlans: return ServicePlans(parent=self, instance_klass=ServicePlan) @property - def proxy(self) -> 'Proxies': + def proxy(self) -> "Proxies": return Proxies(parent=self, instance_klass=Proxy) @property - def mapping_rules(self) -> 'MappingRules': + def mapping_rules(self) -> "MappingRules": return self.proxy.mapping_rules @property - def policies_registry(self) -> 'PoliciesRegistry': + def policies_registry(self) -> "PoliciesRegistry": return PoliciesRegistry(parent=self, instance_klass=PoliciesRegistry) def oidc(self): return self.proxy.oidc @property - def backend_usages(self) -> 'BackendUsages': + def backend_usages(self) -> "BackendUsages": return BackendUsages(parent=self, instance_klass=BackendUsage) @property - def active_docs(self) -> 'ActiveDocs': - """ Active docs related to service. """ + def active_docs(self) -> "ActiveDocs": + """Active docs related to service.""" up_self = self class Wrap(ActiveDocs): - def list(self, **kwargs) -> List['DefaultResource']: + def list(self, **kwargs) -> List["DefaultResource"]: """List all ActiveDocs related to this service.""" - kwargs.update({'service_id': up_self['id']}) + kwargs.update({"service_id": up_self["id"]}) instance = self.select_by(**kwargs) return instance @@ -1343,26 +1318,26 @@ def list(self, **kwargs) -> List['DefaultResource']: class ActiveDoc(DefaultResource): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class Provider(DefaultResource): - def __init__(self, entity_name='org_name', **kwargs): + def __init__(self, entity_name="org_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class AccessToken(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class Tenant(DefaultResource): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) self.admin_base_url = self["signup"]["account"]["admin_base_url"] self.admin_token = None - if "access_token" in self['signup']: + if "access_token" in self["signup"]: self.admin_token = self["signup"]["access_token"]["value"] @property @@ -1377,13 +1352,12 @@ def wait_tenant_ready(self) -> bool: # been finished. return self.admin_api(ssl_verify=False).wait_for_tenant() - def admin_api(self, ssl_verify=True, wait=-1) -> 'client.ThreeScaleClient': + def admin_api(self, ssl_verify=True, wait=-1) -> "client.ThreeScaleClient": """ Returns admin api client for tenant. Its strongly recommended to call this with wait=True """ - return client.ThreeScaleClient( - self.admin_base_url, self.admin_token, ssl_verify=ssl_verify, wait=wait) + return client.ThreeScaleClient(self.admin_base_url, self.admin_token, ssl_verify=ssl_verify, wait=wait) def trigger_billing(self, date: str): """Trigger billing for whole tenant @@ -1394,7 +1368,7 @@ def trigger_billing(self, date: str): """ return self.threescale_client.tenants.trigger_billing(self, date) - def trigger_billing_account(self, account: Union['Account', int], date: str) -> dict: + def trigger_billing_account(self, account: Union["Account", int], date: str) -> dict: """Trigger billing for one account in tenant Args: account: Account id or account resource @@ -1411,28 +1385,22 @@ def plan_upgrade(self, plan_id): @property def account(self): """Return account of this tenant""" - return Account( - client=self.threescale_client.accounts, - entity=self.entity["signup"]["account"] - ) + return Account(client=self.threescale_client.accounts, entity=self.entity["signup"]["account"]) class Application(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) - self._auth_objects = { - Service.AUTH_USER_KEY: auth.UserKeyAuth, - Service.AUTH_APP_ID_KEY: auth.AppIdKeyAuth - } + self._auth_objects = {Service.AUTH_USER_KEY: auth.UserKeyAuth, Service.AUTH_APP_ID_KEY: auth.AppIdKeyAuth} self._api_client_verify = None self._client_factory = utils.HttpClient @property - def account(self) -> 'Account': + def account(self) -> "Account": return self.parent @property - def service(self) -> 'Service': + def service(self) -> "Service": "The service to which this application is bound" return self.threescale_client.services[self["service_id"]] @@ -1456,8 +1424,13 @@ def authobj(self, auth_mode=None, location=None): def register_auth(self, auth_mode: str, factory): self._auth_objects[auth_mode] = factory - def api_client(self, endpoint: str = "sandbox_endpoint", verify: bool = None, cert=None, - disable_retry_status_list: Iterable = ()) -> 'utils.HttpClient': + def api_client( + self, + endpoint: str = "sandbox_endpoint", + verify: bool = None, + cert=None, + disable_retry_status_list: Iterable = (), + ) -> "utils.HttpClient": """This is preconfigured client for the application to run api calls. To avoid failures due to delays in infrastructure it retries call in case of certain condition. To modify this behavior customized session @@ -1525,12 +1498,12 @@ def set_default(self, **kwargs): class ApplicationKey(DefaultResource): - def __init__(self, entity_name='', **kwargs): + def __init__(self, entity_name="", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class Account(DefaultResource): - def __init__(self, entity_name='org_name', **kwargs): + def __init__(self, entity_name="org_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property @@ -1567,20 +1540,20 @@ class UserPermissions(DefaultResource): class AccountUser(DefaultUserResource): - def __init__(self, entity_name='username', **kwargs): + def __init__(self, entity_name="username", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property - def account(self) -> 'Account': + def account(self) -> "Account": return self.parent @property - def permissions(self) -> 'UserPermissionsClient': + def permissions(self) -> "UserPermissionsClient": return UserPermissionsClient(parent=self, instance_klass=UserPermissions) class AccountPlan(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @@ -1597,25 +1570,28 @@ def app_plan(self) -> ApplicationPlan: class Backend(DefaultResource): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property - def metrics(self) -> 'BackendMetrics': + def metrics(self) -> "BackendMetrics": return BackendMetrics(parent=self, instance_klass=BackendMetric) @property - def mapping_rules(self) -> 'BackendMappingRules': + def mapping_rules(self) -> "BackendMappingRules": return BackendMappingRules(parent=self, instance_klass=BackendMappingRule) - def usages(self) -> list['BackendUsage']: - """ Returns list of backend usages where the backend is used.""" - return [usage for service in self.threescale_client.services.list() - for usage in service.backend_usages.select_by(backend_id=self['id'])] + def usages(self) -> list["BackendUsage"]: + """Returns list of backend usages where the backend is used.""" + return [ + usage + for service in self.threescale_client.services.list() + for usage in service.backend_usages.select_by(backend_id=self["id"]) + ] class BackendMetric(Metric): - def __init__(self, entity_name='system_name', **kwargs): + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @@ -1625,40 +1601,38 @@ def __init__(self, **kwargs): class BackendUsage(DefaultResource): - def __init__(self, entity_name='', **kwargs): + def __init__(self, entity_name="", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property - def backend(self) -> 'Backend': - return Backend( - client=Backends(parent=self, instance_klass=Backend), - entity_id=self['backend_id']) + def backend(self) -> "Backend": + return Backend(client=Backends(parent=self, instance_klass=Backend), entity_id=self["backend_id"]) -def _extract_entity_id(entity: Union['DefaultResource', int]): +def _extract_entity_id(entity: Union["DefaultResource", int]): if isinstance(entity, DefaultResource): return entity.entity_id return entity class PolicyRegistry(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property - def proxy(self) -> 'Proxy': + def proxy(self) -> "Proxy": return self.parent @property - def service(self) -> 'Service': + def service(self) -> "Service": return self.proxy.service class ProviderAccount(DefaultResource): - def __init__(self, entity_name='org_name', **kwargs): + def __init__(self, entity_name="org_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) - def update(self, params: dict = None, **kwargs) -> 'DefaultResource': + def update(self, params: dict = None, **kwargs) -> "DefaultResource": new_params = {**self.entity} if params: new_params.update(params) @@ -1668,16 +1642,12 @@ def update(self, params: dict = None, **kwargs) -> 'DefaultResource': class ProviderAccountUser(DefaultStateResource): - def __init__(self, entity_name='username', **kwargs): + def __init__(self, entity_name="username", **kwargs): super().__init__(entity_name=entity_name, **kwargs) - def permissions_update( - self, allowed_services: [] = None, allowed_sections: [] = None, **kwargs): + def permissions_update(self, allowed_services: [] = None, allowed_sections: [] = None, **kwargs): return self.client.permissions_update( - entity_id=self.entity_id, - allowed_services=allowed_services, - allowed_sections=allowed_sections, - **kwargs + entity_id=self.entity_id, allowed_services=allowed_services, allowed_sections=allowed_sections, **kwargs ) def allow_all_sections(self, **kwargs): @@ -1688,32 +1658,32 @@ def permissions_read(self, **kwargs): def set_role_member(self): log.info("Changes the role of the user of the provider account to member") - return self.set_state(state='member') + return self.set_state(state="member") def set_role_admin(self): log.info("Changes the role of the provider account to admin") - return self.set_state(state='admin') + return self.set_state(state="admin") def suspend(self): log.info("Changes the state of the user of the provider account to suspended") - return self.set_state(state='suspend') + return self.set_state(state="suspend") def unsuspend(self): log.info("Revokes the suspension of a user of the provider account") - return self.set_state(state='unsuspend') + return self.set_state(state="unsuspend") def activate(self): log.info("Changes the state of the user of the provider account to active") - return self.set_state(state='activate') + return self.set_state(state="activate") class LineItem(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class Invoice(DefaultResource): - def __init__(self, entity_name='friendly_id', **kwargs): + def __init__(self, entity_name="friendly_id", **kwargs): super().__init__(entity_name=entity_name, **kwargs) @property @@ -1727,65 +1697,71 @@ def charge(self): return self.client.charge(entity_id=self.entity_id) @property - def payment_transactions(self) -> 'PaymentTransactions': + def payment_transactions(self) -> "PaymentTransactions": return PaymentTransactions(parent=self, instance_klass=PaymentTransaction) class PaymentTransaction(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class FieldsDefinition(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class AdminPortalAuthProvider(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class DevPortalAuthProvider(DefaultResource): - def __init__(self, entity_name='name', **kwargs): + def __init__(self, entity_name="name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class CmsFile(DefaultResource): - """ Resource for file """ - def __init__(self, entity_name='path', **kwargs): + """Resource for file""" + + def __init__(self, entity_name="path", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class CmsSection(DefaultResource): - """ Resource for section. """ - def __init__(self, entity_name='id', **kwargs): + """Resource for section.""" + + def __init__(self, entity_name="id", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class CmsTemplate(DefaultResource): - """ Resource for templates """ + """Resource for templates""" + def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) def publish(self, **kwargs): - """ Publish template resource """ + """Publish template resource""" return self.client.publish(entity_id=self.entity_id, **kwargs) class CmsPage(CmsTemplate): - """ Resource for page """ - def __init__(self, entity_name='system_name', **kwargs): + """Resource for page""" + + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class CmsLayout(CmsTemplate): - """ Resource for layout """ - def __init__(self, entity_name='system_name', **kwargs): + """Resource for layout""" + + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) class CmsPartial(CmsTemplate): - """ Resource for partials """ - def __init__(self, entity_name='system_name', **kwargs): + """Resource for partials""" + + def __init__(self, entity_name="system_name", **kwargs): super().__init__(entity_name=entity_name, **kwargs) diff --git a/threescale_api/utils.py b/threescale_api/utils.py index dba6648..12f769f 100644 --- a/threescale_api/utils.py +++ b/threescale_api/utils.py @@ -10,8 +10,7 @@ logger = logging.getLogger(__name__) -def extract_response(response: requests.Response, entity: str = None, - collection: str = None) -> Union[dict, list]: +def extract_response(response: requests.Response, entity: str = None, collection: str = None) -> Union[dict, list]: """Extract the response from the response Args: response(requests.Response): Response @@ -47,8 +46,14 @@ class HttpClient: Iterable collection of status code that should not be retried by requests """ - def __init__(self, app, endpoint: str = "sandbox_endpoint", - verify: bool = None, cert=None, disable_retry_status_list: Iterable = ()): + def __init__( + self, + app, + endpoint: str = "sandbox_endpoint", + verify: bool = None, + cert=None, + disable_retry_status_list: Iterable = (), + ): self._app = app self._endpoint = endpoint self.verify = verify if verify is not None else app.api_client_verify @@ -70,7 +75,7 @@ def retry_for_session(session: requests.Session, status_forcelist: Iterable, tot backoff_factor=1, status_forcelist=status_forcelist, raise_on_status=False, - respect_retry_after_header=False + respect_retry_after_header=False, ) adapter = HTTPAdapter(max_retries=retry) session.mount("https://", adapter) @@ -92,10 +97,23 @@ def extend_connection_pool(self, maxsize: int): self.session.adapters["https://"].poolmanager.connection_pool_kw["maxsize"] = maxsize self.session.adapters["https://"].poolmanager.clear() - def request(self, method, path, - params=None, data=None, headers=None, cookies=None, files=None, - auth=None, timeout=None, allow_redirects=True, proxies=None, - hooks=None, stream=None, json=None) -> requests.Response: + def request( + self, + method, + path, + params=None, + data=None, + headers=None, + cookies=None, + files=None, + auth=None, + timeout=None, + allow_redirects=True, + proxies=None, + hooks=None, + stream=None, + json=None, + ) -> requests.Response: """mimics requests interface""" url = urljoin(self._base_url, path) session = self.session @@ -117,15 +135,11 @@ def request(self, method, path, logger.info("[CLIENT]: %s", request2curl(prep)) - send_kwargs = { - "timeout": timeout, - "allow_redirects": allow_redirects - } + send_kwargs = {"timeout": timeout, "allow_redirects": allow_redirects} proxies = proxies or {} - send_kwargs.update( - session.merge_environment_settings(prep.url, proxies, stream, self.verify, self.cert)) + send_kwargs.update(session.merge_environment_settings(prep.url, proxies, stream, self.verify, self.cert)) response = session.send(prep, **send_kwargs) @@ -135,23 +149,23 @@ def request(self, method, path, def get(self, *args, **kwargs) -> requests.Response: """mimics requests interface""" - return self.request('GET', *args, **kwargs) + return self.request("GET", *args, **kwargs) def post(self, *args, **kwargs) -> requests.Response: """mimics requests interface""" - return self.request('POST', *args, **kwargs) + return self.request("POST", *args, **kwargs) def patch(self, *args, **kwargs) -> requests.Response: """mimics requests interface""" - return self.request('PATCH', *args, **kwargs) + return self.request("PATCH", *args, **kwargs) def put(self, *args, **kwargs) -> requests.Response: """mimics requests interface""" - return self.request('PUT', *args, **kwargs) + return self.request("PUT", *args, **kwargs) def delete(self, *args, **kwargs) -> requests.Response: """mimics requests interface""" - return self.request('DELETE', *args, **kwargs) + return self.request("DELETE", *args, **kwargs) def request2curl(request: requests.PreparedRequest) -> str: @@ -161,9 +175,7 @@ def request2curl(request: requests.PreparedRequest) -> str: cmd = ["curl", "-X %s" % shlex.quote(request.method)] if request.headers: # pylint: disable=consider-using-f-string - cmd.extend([ - "-H %s" % shlex.quote(f"{key}: {value}") - for key, value in request.headers.items()]) + cmd.extend(["-H %s" % shlex.quote(f"{key}: {value}") for key, value in request.headers.items()]) if request.body: body = request.body if isinstance(body, bytes):