Black_Raven
dd7f07b2d0
Executor performs HTTP requests:
- Uses httpx for HTTP calls
- Resolves {{variable}} in URLs
- Returns Response with status, body, headers
36 lines
1.0 KiB
Python
36 lines
1.0 KiB
Python
import httpx
|
|
from .context import Context
|
|
from .itm import RequestModel
|
|
|
|
|
|
class Response:
|
|
def __init__(self, status: int, body: dict, headers: dict):
|
|
self.status = status
|
|
self.body = body
|
|
self.headers = headers
|
|
|
|
|
|
class Executor:
|
|
def __init__(self, context: Context):
|
|
self.context = context
|
|
|
|
def execute(self, request: RequestModel) -> Response:
|
|
url = self.context.resolve(request.path)
|
|
if not url.startswith("http"):
|
|
url = f"{self.context.config.base_url}{url}"
|
|
|
|
with httpx.Client() as client:
|
|
resp = client.request(
|
|
method=request.method,
|
|
url=url,
|
|
headers=request.headers,
|
|
json=request.body if request.body_type == "json" else None,
|
|
timeout=request.timeout,
|
|
)
|
|
|
|
return Response(
|
|
status=resp.status_code,
|
|
body=resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {},
|
|
headers=dict(resp.headers),
|
|
)
|