HTTP Request

Inputs

Outputs

Source

python

inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]

import json
import time
import requests

METHODS = ("GET", "POST", "PUT", "PATCH", "DELETE")

url = inp.get("url")
if not isinstance(url, str) or not url.strip():
    raise ValueError("url is empty; give the address to call.")
url = url.strip()
if not url.lower().startswith(("http://", "https://")):
    raise ValueError(f"url must start with http:// or https://, got {url[:40]!r}.")
method = str(inp.get("method") or "GET").upper()
if method not in METHODS:
    raise ValueError(f"method must be one of {', '.join(METHODS)}, got {method!r}.")

headers = {str(k): str(v) for k, v in (inp.get("headers") or {}).items()}
params = inp.get("params") or {}
if not isinstance(params, dict):
    raise ValueError("params must be an object of query parameters.")
json_body = inp.get("json")
text_body = inp.get("body")
if json_body is not None and text_body is not None:
    raise ValueError("Give json or body, not both.")

token = inp.get("token")
if token:
    header_name = str(inp.get("auth_header") or "Authorization")
    scheme = inp["auth_scheme"] if "auth_scheme" in inp and inp["auth_scheme"] is not None else "Bearer"
    headers[header_name] = (str(scheme).strip() + " " + str(token)).strip()

kwargs = {}
if json_body is not None:
    kwargs["json"] = json_body
elif text_body is not None:
    kwargs["data"] = str(text_body).encode("utf-8")
    headers.setdefault("Content-Type", str(inp.get("content_type") or "text/plain; charset=utf-8"))

timeout = float(inp.get("timeout") or 30)
started = time.time()
try:
    response = requests.request(method, url, headers=headers, params=params, timeout=timeout, **kwargs)
except requests.exceptions.Timeout:
    raise TimeoutError(f"No answer from {url} within {timeout:g} seconds.")
except requests.exceptions.ConnectionError:
    raise ConnectionError(f"Could not reach {url}.")
elapsed_ms = int((time.time() - started) * 1000)

text = response.text
parsed = None
content_type = response.headers.get("Content-Type", "")
if text and ("json" in content_type.lower() or text.lstrip()[:1] in ("{", "[")):
    try:
        parsed = json.loads(text)
    except ValueError:
        parsed = None

fail_on_error = inp.get("fail_on_error")
if fail_on_error is None:
    fail_on_error = True
if fail_on_error and response.status_code >= 400:
    if response.status_code in (401, 403):
        raise PermissionError(f"{response.status_code} from {url}: the credentials were refused.")
    raise RuntimeError(f"{response.status_code} from {url}: {text[:200]}")

out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]
out["status"] = response.status_code
out["ok"] = response.status_code < 400
out["headers"] = {k.lower(): v for k, v in response.headers.items()}
out["json"] = parsed
out["text"] = text
out["url"] = response.url
out["elapsed_ms"] = elapsed_ms

Tests

Requires: python:3.9, requests

• get-json

  A GET with query parameters comes back with the JSON the server answered.

• post-json

  A POST sends its JSON body with the JSON content type.

• bearer

  A token is sent as a Bearer header.

• custom-auth-header

  An API that reads X-API-Key gets the token there, with no scheme.

• text-body

  A raw body goes out as text with the given content type.

• not-found

  A 404 is an error that names the status and the address.

• not-found-tolerated

  With fail_on_error false the 404 comes back as a status.

• both-bodies

  json and body together are refused before any call.