mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 09:13:14 +00:00
70 lines
1.9 KiB
Python
70 lines
1.9 KiB
Python
from typing import Any
|
|
|
|
import requests
|
|
|
|
|
|
def http_request(
|
|
url: str,
|
|
method: str = "GET",
|
|
headers: dict[str, str] | None = None,
|
|
data: str | dict | None = None,
|
|
params: dict[str, str] | None = None,
|
|
timeout: int = 30,
|
|
) -> dict[str, Any]:
|
|
"""Make HTTP requests to APIs and web services.
|
|
|
|
Args:
|
|
url: Target URL
|
|
method: HTTP method (GET, POST, PUT, DELETE, etc.)
|
|
headers: HTTP headers to include
|
|
data: Request body data (string or dict)
|
|
params: URL query parameters
|
|
timeout: Request timeout in seconds
|
|
|
|
Returns:
|
|
Dictionary with response data including status, headers, and content
|
|
"""
|
|
try:
|
|
kwargs: dict[str, Any] = {}
|
|
|
|
if headers:
|
|
kwargs["headers"] = headers
|
|
if params:
|
|
kwargs["params"] = params
|
|
if data:
|
|
if isinstance(data, dict):
|
|
kwargs["json"] = data
|
|
else:
|
|
kwargs["data"] = data
|
|
|
|
response = requests.request(method.upper(), url, timeout=timeout, **kwargs)
|
|
|
|
try:
|
|
content = response.json()
|
|
except (ValueError, requests.exceptions.JSONDecodeError):
|
|
content = response.text
|
|
|
|
return {
|
|
"success": response.status_code < 400,
|
|
"status_code": response.status_code,
|
|
"headers": dict(response.headers),
|
|
"content": content,
|
|
"url": response.url,
|
|
}
|
|
|
|
except requests.exceptions.Timeout:
|
|
return {
|
|
"success": False,
|
|
"status_code": 0,
|
|
"headers": {},
|
|
"content": f"Request timed out after {timeout} seconds",
|
|
"url": url,
|
|
}
|
|
except requests.exceptions.RequestException as e:
|
|
return {
|
|
"success": False,
|
|
"status_code": 0,
|
|
"headers": {},
|
|
"content": f"Request error: {e!s}",
|
|
"url": url,
|
|
}
|