netrequests

Minimal requests-compatible shim for IronPython / pyRevit using .NET HttpClient.

Supports: - requests.get(...) - requests.post(...) - requests.put(...) - requests.patch(...) - requests.delete(...) - params={} - json={} - data= - headers={} - timeout= - stream=True - response.status_code - response.ok - response.text - response.content - response.json() - response.iter_lines() - response.close() - requests.RequestException - requests.exceptions.RequestException

Attributes

exceptions = _Exceptions() module-attribute

Classes

RequestException

Bases: Exception

Timeout

Bases: RequestException

Response(dotnet_response, stream=False, client=None)

Bases: object

Source code in pyrevitlib/pyrevit/netrequests.py

def __init__(self, dotnet_response, stream=False, client=None):
    self._response = dotnet_response
    self._stream = stream
    self._client = client
    self._reader = None

    try:
        self.status_code = int(dotnet_response.StatusCode)
    except Exception:
        self.status_code = 0

    try:
        self.reason = str(dotnet_response.ReasonPhrase)
    except Exception:
        self.reason = ""

    self.headers = {}

    try:
        for h in dotnet_response.Headers:
            try:
                values = [str(x) for x in h.Value]
                self.headers[str(h.Key)] = ",".join(values)
            except Exception:
                pass
    except Exception:
        pass

    self._content = b""
    self._text = ""

    try:
        if not dotnet_response.Content:
            return
        try:
            for h in dotnet_response.Content.Headers:
                try:
                    values = [str(x) for x in h.Value]
                    self.headers[str(h.Key)] = ",".join(values)
                except Exception:
                    pass
        except Exception:
            pass

            # Important:
            # If this is a streaming response, do NOT read the full content here.
            # SSE endpoints may never complete, so ReadAsStringAsync().Result
            # would block forever.
            if stream:
                return
            try:
                self._content = (
                    dotnet_response.Content.ReadAsByteArrayAsync().Result
                )
            except Exception:
                self._content = b""

            try:
                self._text = dotnet_response.Content.ReadAsStringAsync().Result
            except Exception:
                self._text = ""

    except Exception:
        pass

Attributes

status_code = int(dotnet_response.StatusCode) instance-attribute

reason = str(dotnet_response.ReasonPhrase) instance-attribute

headers = {} instance-attribute

ok property

text property

content property

raw property

Methods:

json()

Source code in pyrevitlib/pyrevit/netrequests.py

def json(self):
    import json

    if self._stream:
        raise RequestException("Cannot parse streamed response as JSON.")

    if not self._text:
        return {}

    try:
        return json.loads(self._text)

    except Exception as ex:
        raise RequestException("Invalid JSON: {}".format(ex))

raise_for_status()

Source code in pyrevitlib/pyrevit/netrequests.py

def raise_for_status(self):
    if not self.ok:
        raise RequestException("HTTP {} {}".format(self.status_code, self.reason))

iter_lines()

Minimal implementation for requests.Response.iter_lines().

Intended for SSE-style responses:

for raw_line in response.iter_lines():
    ...

Returns strings in IronPython/.NET rather than bytes. Your existing code already handles both.

Source code in pyrevitlib/pyrevit/netrequests.py

def iter_lines(self):
    """
    Minimal implementation for requests.Response.iter_lines().

    Intended for SSE-style responses:

        for raw_line in response.iter_lines():
            ...

    Returns strings in IronPython/.NET rather than bytes.
    Your existing code already handles both.
    """

    try:
        if not self._response or not self._response.Content:
            return

        stream = self._response.Content.ReadAsStreamAsync().Result

        self._reader = StreamReader(stream)

        while True:
            try:
                line = self._reader.ReadLine()
            except Exception:
                break

            if line is None:
                break

            yield line

    except Exception:
        return

close()

Source code in pyrevitlib/pyrevit/netrequests.py

def close(self):
    try:
        if self._reader:
            self._reader.Close()
    except Exception:
        pass

    try:
        if self._response:
            self._response.Dispose()
    except Exception:
        pass

    try:
        if self._client:
            self._client.Dispose()
    except Exception:
        pass

Functions:

request(method, url, headers=None, params=None, data=None, json=None, stream=False, timeout=None, verify=None, allow_redirects=True, **kwargs)

Source code in pyrevitlib/pyrevit/netrequests.py

def request(
    method,
    url,
    headers=None,
    params=None,
    data=None,
    json=None,
    stream=False,
    timeout=None,
    verify=None,
    allow_redirects=True,
    **kwargs
):

    if verify is False:
        raise NotImplementedError("verify=False is not supported")

    if allow_redirects is False:
        raise NotImplementedError("allow_redirects=False is not supported")

    client = _shared_client
    temporary_client = False
    req = None

    try:
        url = _apply_params(url, params)

        if timeout is not None:
            try:
                client = HttpClient()
                client.Timeout = TimeSpan.FromSeconds(float(timeout))
                temporary_client = True
            except Exception:
                client = _shared_client
                temporary_client = False

        req = _create_request(method, url, headers=headers, data=data, json_data=json)

        try:
            if stream:
                resp = client.SendAsync(
                    req, HttpCompletionOption.ResponseHeadersRead
                ).Result
            else:
                resp = client.SendAsync(req).Result

        except Exception as ex:
            raise RequestException(
                "HTTP request failed: {}".format(_get_exception_message(ex))
            )

        finally:
            try:
                if req:
                    req.Dispose()
            except Exception:
                pass

        if resp is None:
            raise RequestException("No response received")

        # If streaming, keep the temporary client alive until response.close().
        if stream and temporary_client:
            return Response(resp, stream=True, client=client)

        response = Response(resp, stream=stream, client=None)

        return response

    except RequestException:
        raise

    except Exception as ex:
        import traceback

        raise RequestException(
            "{}\n\n{}".format(_get_exception_message(ex), traceback.format_exc())
        )

    finally:
        # For non-streaming temporary clients, safe to dispose here because
        # response body has already been read into memory.
        if temporary_client and not stream:
            try:
                client.Dispose()
            except Exception:
                pass

get(url, headers=None, params=None, stream=False, timeout=None, verify=None, allow_redirects=True, **kwargs)

Source code in pyrevitlib/pyrevit/netrequests.py

def get(
    url,
    headers=None,
    params=None,
    stream=False,
    timeout=None,
    verify=None,
    allow_redirects=True,
    **kwargs
):

    return request(
        "GET",
        url,
        headers=headers,
        params=params,
        stream=stream,
        timeout=timeout,
        verify=verify,
        allow_redirects=allow_redirects,
        **kwargs
    )

post(url, headers=None, params=None, data=None, json=None, stream=False, timeout=None, verify=None, allow_redirects=True, **kwargs)

Source code in pyrevitlib/pyrevit/netrequests.py

def post(
    url,
    headers=None,
    params=None,
    data=None,
    json=None,
    stream=False,
    timeout=None,
    verify=None,
    allow_redirects=True,
    **kwargs
):

    return request(
        "POST",
        url,
        headers=headers,
        params=params,
        data=data,
        json=json,
        stream=stream,
        timeout=timeout,
        verify=verify,
        allow_redirects=allow_redirects,
        **kwargs
    )

put(url, headers=None, params=None, data=None, json=None, stream=False, timeout=None, verify=None, allow_redirects=True, **kwargs)

Source code in pyrevitlib/pyrevit/netrequests.py

def put(
    url,
    headers=None,
    params=None,
    data=None,
    json=None,
    stream=False,
    timeout=None,
    verify=None,
    allow_redirects=True,
    **kwargs
):

    return request(
        "PUT",
        url,
        headers=headers,
        params=params,
        data=data,
        json=json,
        stream=stream,
        timeout=timeout,
        verify=verify,
        allow_redirects=allow_redirects,
        **kwargs
    )

patch(url, headers=None, params=None, data=None, json=None, stream=False, timeout=None, verify=None, allow_redirects=True, **kwargs)

Source code in pyrevitlib/pyrevit/netrequests.py

def patch(
    url,
    headers=None,
    params=None,
    data=None,
    json=None,
    stream=False,
    timeout=None,
    verify=None,
    allow_redirects=True,
    **kwargs
):

    return request(
        "PATCH",
        url,
        headers=headers,
        params=params,
        data=data,
        json=json,
        stream=stream,
        timeout=timeout,
        verify=verify,
        allow_redirects=allow_redirects,
        **kwargs
    )

delete(url, headers=None, params=None, timeout=None, verify=None, allow_redirects=True, **kwargs)

Source code in pyrevitlib/pyrevit/netrequests.py

def delete(
    url,
    headers=None,
    params=None,
    timeout=None,
    verify=None,
    allow_redirects=True,
    **kwargs
):

    return request(
        "DELETE",
        url,
        headers=headers,
        params=params,
        timeout=timeout,
        verify=verify,
        allow_redirects=allow_redirects,
        **kwargs
    )

head(url, headers=None, params=None, timeout=None, verify=None, allow_redirects=True, **kwargs)

Source code in pyrevitlib/pyrevit/netrequests.py

def head(
    url,
    headers=None,
    params=None,
    timeout=None,
    verify=None,
    allow_redirects=True,
    **kwargs
):

    return request(
        "HEAD",
        url,
        headers=headers,
        params=params,
        timeout=timeout,
        verify=verify,
        allow_redirects=allow_redirects,
        **kwargs
    )