From 6294aa2957217160906d7185426345cfae7701af Mon Sep 17 00:00:00 2001 From: David Michaels Date: Tue, 19 Dec 2023 15:12:15 -0500 Subject: [PATCH] minor portal_utils refactoring --- dcicutils/portal_utils.py | 81 ++++++++++++++++++++------------------- 1 file changed, 42 insertions(+), 39 deletions(-) diff --git a/dcicutils/portal_utils.py b/dcicutils/portal_utils.py index df6625ce7..3f84da3cd 100644 --- a/dcicutils/portal_utils.py +++ b/dcicutils/portal_utils.py @@ -8,7 +8,7 @@ import os import re import requests -from requests.models import Response as RequestResponse +from requests.models import Response from threading import Thread from typing import Callable, Dict, List, Optional, Type, Union from uuid import uuid4 as uuid @@ -20,6 +20,7 @@ from dcicutils.zip_utils import temporary_file Portal = Type["Portal"] # Forward type reference for type hints. +OptionalResponse = Optional[Union[Response, TestResponse]] class Portal: @@ -200,6 +201,29 @@ def app(self) -> Optional[str]: def vapp(self) -> Optional[TestApp]: return self._vapp + def get(self, url: str, follow: bool = True, **kwargs) -> OptionalResponse: + if not self._vapp: + return requests.get(self.url(url), allow_redirects=follow, **self._kwargs(**kwargs)) + response = self._vapp.get(self.url(url), **self._kwargs(**kwargs)) + if response and response.status_code in [301, 302, 303, 307, 308] and follow: + response = response.follow() + return self._response(response) + + def patch(self, url: str, data: Optional[dict] = None, json: Optional[dict] = None, **kwargs) -> OptionalResponse: + if not self._vapp: + return requests.patch(self.url(url), data=data, json=json, **self._kwargs(**kwargs)) + return self._response(self._vapp.patch_json(self.url(url), json or data, **self._kwargs(**kwargs))) + + def post(self, url: str, data: Optional[dict] = None, json: Optional[dict] = None, + files: Optional[dict] = None, **kwargs) -> OptionalResponse: + if not self._vapp: + return requests.post(self.url(url), data=data, json=json, files=files, **self._kwargs(**kwargs)) + if files: + response = self._vapp.post(self.url(url), json or data, upload_files=files, **self._kwargs(**kwargs)) + else: + response = self._vapp.post_json(self.url(url), json or data, upload_files=files, **self._kwargs(**kwargs)) + return self._response(response) + def get_metadata(self, object_id: str) -> Optional[dict]: return get_metadata(obj_id=object_id, vapp=self._vapp, key=self._key) @@ -213,29 +237,14 @@ def post_metadata(self, object_type: str, data: str) -> Optional[dict]: return post_metadata(schema_name=object_type, post_item=data, key=self._key) return self.post(f"/{object_type}", data).json() - def get(self, uri: str, follow: bool = True, **kwargs) -> Optional[Union[RequestResponse, TestResponse]]: - if not self._vapp: - return requests.get(self.url(uri), allow_redirects=follow, **self._kwargs(**kwargs)) - response = self._vapp.get(self.url(uri), **self._kwargs(**kwargs)) - if response and response.status_code in [301, 302, 303, 307, 308] and follow: - response = response.follow() - return self._response(response) + def get_health(self) -> Optional[Union[Response, TestResponse]]: + return self.get("/health") - def patch(self, uri: str, data: Optional[dict] = None, - json: Optional[dict] = None, **kwargs) -> Optional[Union[RequestResponse, TestResponse]]: - if not self._vapp: - return requests.patch(self.url(uri), data=data, json=json, **self._kwargs(**kwargs)) - return self._response(self._vapp.patch_json(self.url(uri), json or data, **self._kwargs(**kwargs))) - - def post(self, uri: str, data: Optional[dict] = None, json: Optional[dict] = None, - files: Optional[dict] = None, **kwargs) -> Optional[Union[RequestResponse, TestResponse]]: - if not self._vapp: - return requests.post(self.url(uri), data=data, json=json, files=files, **self._kwargs(**kwargs)) - if files: - response = self._vapp.post(self.url(uri), json or data, upload_files=files, **self._kwargs(**kwargs)) - else: - response = self._vapp.post_json(self.url(uri), json or data, upload_files=files, **self._kwargs(**kwargs)) - return self._response(response) + def ping(self) -> bool: + try: + return self.get_health().status_code == 200 + except Exception: + return False def get_schema(self, schema_name: str) -> Optional[dict]: return get_schema(self.schema_name(schema_name), portal_vapp=self._vapp, key=self._key) @@ -283,20 +292,14 @@ def breadth_first(super_type_map: dict, super_type_name: str) -> dict: super_type_map_flattened[super_type_name] = breadth_first(super_type_map, super_type_name) return super_type_map_flattened - def ping(self) -> bool: - try: - return self.get("/health").status_code == 200 - except Exception: - return False - - def url(self, uri: str) -> str: - if not isinstance(uri, str) or not uri: + def url(self, url: str) -> str: + if not isinstance(url, str) or not url: return "/" - if (luri := uri.lower()).startswith("http://") or luri.startswith("https://"): - return uri - if not (uri := re.sub(r"/+", "/", uri)).startswith("/"): - uri = "/" - return self._server + uri if self._server else uri + if (lurl := url.lower()).startswith("http://") or lurl.startswith("https://"): + return url + if not (url := re.sub(r"/+", "/", url)).startswith("/"): + url = "/" + return self._server + url if self._server else url def _kwargs(self, **kwargs) -> dict: result_kwargs = {"headers": @@ -317,9 +320,9 @@ def infer_app_from_env(env: str) -> Optional[str]: # noqa if is_valid_app(app) or (app := infer_app_from_env(env)): return os.path.join(Portal.KEYS_FILE_DIRECTORY, f".{app.lower()}-keys.json") - def _response(self, response: TestResponse) -> Optional[RequestResponse]: + def _response(self, response: TestResponse) -> Optional[Response]: if response and isinstance(getattr(response.__class__, "json"), property): - class RequestResponseWrapper: # For consistency change json property to method. + class TestResponseWrapper(TestResponse): # For consistency change json property to method. def __init__(self, response, **kwargs): super().__init__(**kwargs) self._response = response @@ -327,7 +330,7 @@ def __getattr__(self, attr): # noqa return getattr(self._response, attr) def json(self): # noqa return self._response.json - response = RequestResponseWrapper(response) + response = TestResponseWrapper(response) return response @staticmethod