How to use the jsonrpcclient.response.SuccessResponse function in jsonrpcclient

To help you get started, we’ve selected a few jsonrpcclient examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github bcb / jsonrpcclient / tests / test_response.py View on Github external
def test_success_response_null_id():
    # Acceptable.
    response = SuccessResponse(**{"jsonrpc": "2.0", "result": "foo", "id": None})
    assert response.ok == True
    assert response.id == None
    assert repr(response) == ""
    assert str(response) == '{"jsonrpc": "2.0", "result": "foo", "id": null}'
github bcb / jsonrpcclient / tests / test_response.py View on Github external
def test_success_response():
    response = SuccessResponse(**{"jsonrpc": "2.0", "result": 5, "id": 1})
    assert response.ok == True
    assert response.id == 1
    assert response.result == 5
    assert repr(response) == ""
    assert str(response) == '{"jsonrpc": "2.0", "result": 5, "id": 1}'
github bcb / jsonrpcclient / tests / test_response.py View on Github external
def test_total_responses_list():
    response = SuccessResponse(**{"jsonrpc": "2.0", "result": "foo", "id": 1})
    assert total_results([response, response]) == 2
github bcb / jsonrpcclient / tests / test_response.py View on Github external
def test_total_responses_one():
    response = SuccessResponse(**{"jsonrpc": "2.0", "result": "foo", "id": 1})
    assert total_results(response) == 1
github bcb / jsonrpcclient / jsonrpcclient / response.py View on Github external
ok = True

    def __init__(self, result: Any, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        self.result = result

    def __repr__(self) -> str:
        return "".format(self.id, self.result)

    def __str__(self) -> str:
        return serialize(
            sort_response(dict(jsonrpc=self.jsonrpc, result=self.result, id=self.id))
        )


class NotificationResponse(SuccessResponse):
    """
    Represents a JSON-RPC notification response object.
    """

    ok = True

    def __init__(self) -> None:
        super().__init__(jsonrpc="2.0", result=None, id=NOID)

    def __repr__(self) -> str:
        return ""

    def __str__(self) -> str:
        return ""
github bcb / jsonrpcclient / jsonrpcclient / parse.py View on Github external
def get_response(response: Dict[str, Any]) -> JSONRPCResponse:
    """
    Converts a deserialized response into a JSONRPCResponse object.

    The dictionary be either an error or success response, never a notification.

    Args:
        response: Deserialized response dictionary. We can assume the response is valid
            JSON-RPC here, since it passed the jsonschema validation.
    """
    if "error" in response:
        return ErrorResponse(**response)
    return SuccessResponse(**response)
github Opentrons / opentrons / api / src / opentrons / hardware_control / socket_client.py View on Github external
assert isinstance(request, dict),\
            'How did you manage to already serialize this request'
        method = request['method']
        if method in self._funcmap and request.get('params'):
            new_params = _apply_arg_transforms(
                request['params'],
                self._funcmap[method])
            LOG.debug(
                f"request param xform: {request['params']} -> {new_params}")
            request['params'] = new_params
        elif method not in self._funcmap:
            LOG.warning(
                f"jrpc client: no funcmap entry for {request['method']}")

        response = await super().send(request, *args, **kwargs)
        if isinstance(response.data, SuccessResponse)\
           and method in self._funcmap:
            # We need to transform the return type
            xformed = self._funcmap[method].return_xform(response.data.result)
            LOG.debug(f'response xform: {response.data.result}->{xformed}')
            response.data.result = xformed
        return response