(method, url, payload={}, headers={})
| 22 | |
| 23 | |
| 24 | def _request(method, url, payload={}, headers={}): |
| 25 | base_url = os.environ.get("CLARIFAI_GRPC_BASE", "api.clarifai.com") |
| 26 | |
| 27 | opener = build_opener(HTTPHandler) |
| 28 | full_url = f"https://{base_url}/v2{url}" |
| 29 | request = Request(full_url, data=json.dumps(payload).encode()) |
| 30 | for k in headers.keys(): |
| 31 | request.add_header(k, headers[k]) |
| 32 | request.get_method = lambda: method |
| 33 | try: |
| 34 | response = opener.open(request).read().decode() |
| 35 | except HTTPError as e: |
| 36 | error_body = e.read().decode() |
| 37 | try: |
| 38 | error_body = json.dumps(json.loads(error_body), indent=4) |
| 39 | except Exception: |
| 40 | pass |
| 41 | raise Exception( |
| 42 | "ERROR after a HTTP request to: %s %s" % (method, full_url) |
| 43 | + ". Response: %d %s:\n%s" % (e.code, e.reason, error_body) |
| 44 | ) |
| 45 | return json.loads(response) |
| 46 | |
| 47 | |
| 48 | def login(): |
no test coverage detected