client

def delete_from_eos(self, response, **_) -> None:
    """
    requests response hook function to delete a file from eos.
    """
    try:
        response.raise_for_status()
    except HTTPError:
        log.warning("Something went wrong with deleting the attachment.")
        return response

    data = response.json()

    # do nothing if it's not an EOS-type attachment
    # or if betamax is being used (no need to run the cURL for EOS)
    if (
        data["attachment"]["type"] != "eos"
        or response.connection.__class__.__name__ == "BetamaxAdapter"
    ):
        return None

    if "token" not in data:
        log.warning(
            "It seems there is no token, so we are not deleting this from EOS."
        )
        return None

    log.info(
        "It looks like you're deleting an attachment from ITk PD that is stored on EOS, I will try to delete it from EOS for you."
    )

    response.eos_response = eos.delete(data["token"], data["attachment"]["url"])
    return None

def get(self, url, **kwargs):
    is_cern_url = ".cern.ch" in urlparse(url).netloc
    # is_binary_data = "uu-app-binarystore/getBinaryData" in url
    if is_cern_url and "verify" not in kwargs:
        log.info(
            "Identified a cern.ch request, will attach CERN SSL chain to request by overriding `verify`."
        )
        kwargs["verify"] = itkdb_data / "CERN_chain.pem"

    # getBinaryData does not handle chunked requests
    # if is_cern_url or is_binary_data:
    if is_cern_url:
        log.info(
            "Identified a request that potentially downloads larger amounts of data, will execute chunked requests (stream=True)."
        )
        kwargs["stream"] = True
        headers = kwargs.get("headers", {})
        headers["transfer-encoding"] = "chunked"
        kwargs["headers"] = headers
    return super().get(url, **kwargs)

def post(self, url, data=None, json=None, **kwargs):
    """
    Make a POST request.

    Args:
        url (str): URI to POST to
        allow_duplicate (bool): Indicate if this request should allow creation of duplicate item (see below)
        kwargs: (any): All other keyword arguments supposed by [itkdb.core.Session.post][]

    !!! note "Changed in version 0.6.1"
        `allow_duplicate` keyword argument supported

    The following routes have duplicate check support:
        - `uploadTestRunResults` (added in `v0.6.1`)

    """
    allow_duplicate = kwargs.pop("allow_duplicate", True)
    if not allow_duplicate:
        duplicates = []
        if not json:
            msg = "You asked for me to check for duplicates, but you didn't provide any data? If this message was in error, please get in touch with the itkdb developers: https://itkdb.docs.cern.ch/ ."
            raise ValueError(msg)

        if url.endswith("uploadTestRunResults"):
            duplicates = self._get_duplicate_test_runs(json)
            if duplicates:
                msg = f"Duplicate test runs: {duplicates}"
                raise exceptions.DuplicateTestRuns(msg)
        else:
            msg = f"No logic exists to check for duplicates for url: {url}. Either submit an MR or remove 'allow_duplicate=False'."
            raise ValueError(msg)
    return super().post(url, data=data, json=json, **kwargs)

def prepare_request(self, request):
    request.url = self._normalize_url(request.url)
    self._request_handler(request)
    return super().prepare_request(request)

def request(self, method, url, *args, **kwargs):
    self.limit = kwargs.pop("limit", -1)

    response = super(Session, self).request(method, url, *args, **kwargs)
    return self._response_handler(response)

def retry_with_old_binary(
    self, response, request_url, request_params, **kwargs
) -> None:
    """
    requests response hook function to retry with old binary API.
    """
    try:
        self._check_response(response)
        return response
    except exceptions.BadRequest as he:
        redirect_error = he

    redirected_url = "uu-app-binarystore/getBinaryData"
    log.warning("Error from '%s' retrying as '%s'", request_url, redirected_url)

    bin_code = request_params["code"]
    json = {"code": bin_code}

    try:
        retry_response = self.get(redirected_url, json=json, **kwargs)
        retry_response.raise_for_status()
        return retry_response
    except exceptions.BadRequest as re_error:
        # Raise original error
        raise redirect_error from re_error

def upload_to_eos(self, response, eos_file_details=None, **_) -> None:
    """
    requests response hook function to upload a file to eos.
    """
    log.info("I was able to get a token to upload to EOS. Let me upload.")
    try:
        response.raise_for_status()
    except HTTPError:
        log.warning("Something went wrong with uploading to EOS.")
        return response

    # do nothing if betamax is being used (no need to run the cURL for EOS)
    if response.connection.__class__.__name__ == "BetamaxAdapter":
        return None

    token_request = response.json()

    log.info(token_request)

    response.eos_response = eos.put(
        token_request["token"], token_request["url"], eos_file_details
    )
    return None