Send an HTTP request. Args: path (str): API endpoint path. method (str): HTTP method (default: "GET"). body (dict, optional): Request body (default: None). Returns: dict | iter: JSON response or stream iterator.
(self, path = "", method = "GET", post_body = None, provider_key = None)
| 25 | } |
| 26 | self.host = f"http{'://' if dev else 's://'}{'localhost:8080' if dev else 'api.bytez.com'}/models/v2/" |
| 27 | def request(self, path = "", method = "GET", post_body = None, provider_key = None): |
| 28 | """ |
| 29 | Send an HTTP request. |
| 30 | |
| 31 | Args: |
| 32 | path (str): API endpoint path. |
| 33 | method (str): HTTP method (default: "GET"). |
| 34 | body (dict, optional): Request body (default: None). |
| 35 | |
| 36 | Returns: |
| 37 | dict | iter: JSON response or stream iterator. |
| 38 | """ |
| 39 | try: |
| 40 | stream = bool(post_body and post_body.get("stream")) |
| 41 | response = requests.request( |
| 42 | method, |
| 43 | self.host + path, |
| 44 | headers = ( { **self.headers, "provider-key": provider_key } if provider_key is not None else self.headers ), |
| 45 | # drop null values from being sent |
| 46 | data = json.dumps({k: v for k, v in post_body.items() if v is not None}) if post_body else None, |
| 47 | stream = stream |
| 48 | ) |
| 49 | |
| 50 | if stream: |
| 51 | response.encoding = "utf-8" |
| 52 | |
| 53 | return (line for line in response.iter_lines(decode_unicode=True) if line) |
| 54 | else: |
| 55 | results = response.json() |
| 56 | |
| 57 | return Response(output=results.get('output'), error=results.get('error'), provider=results.get('provider')) |
| 58 | except Exception as error: |
| 59 | return Response(error=str(error)) |
no test coverage detected