Skip to content

Commit

Permalink
add docs and refactor kwarg filter
Browse files Browse the repository at this point in the history
  • Loading branch information
JR-1991 committed Mar 1, 2024
1 parent 0fd58ee commit 2b4c740
Showing 1 changed file with 66 additions and 2 deletions.
68 changes: 66 additions & 2 deletions src/pyDataverse/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -268,9 +268,23 @@ def _sync_request(
method,
**kwargs,
):
"""
Sends a synchronous request to the specified URL using the specified HTTP method.
Args:
method (function): The HTTP method to use for the request.
**kwargs: Additional keyword arguments to be passed to the method.
Returns:
requests.Response: The response object returned by the request.
Raises:
ApiAuthorizationError: If the response status code is 401 (Authorization error).
ConnectError: If a connection to the API cannot be established.
"""
assert "url" in kwargs, "URL is required for a request."

kwargs = {k: v for k, v in kwargs.items() if v is not None}
kwargs = self._filter_kwargs(kwargs)

try:
resp = method(**kwargs)
Expand All @@ -297,7 +311,57 @@ async def _async_request(
method,
**kwargs,
):
return await method(**kwargs)
"""
Sends an asynchronous request to the specified URL using the specified HTTP method.
Args:
method (callable): The HTTP method to use for the request.
**kwargs: Additional keyword arguments to be passed to the method.
Raises:
ApiAuthorizationError: If the response status code is 401 (Authorization error).
ConnectError: If a connection to the API cannot be established.
Returns:
The response object.
"""
assert "url" in kwargs, "URL is required for a request."

kwargs = self._filter_kwargs(kwargs)

try:
resp = await method(**kwargs)

if resp.status_code == 401:
error_msg = resp.json()["message"]
raise ApiAuthorizationError(
"ERROR: HTTP 401 - Authorization error {0}. MSG: {1}".format(
kwargs["url"], error_msg
)
)

return resp

except ConnectError:
raise ConnectError(
"ERROR - Could not establish connection to api '{0}'.".format(
kwargs["url"]
)
)

@staticmethod
def _filter_kwargs(kwargs: Dict[str, Any]) -> Dict[str, Any]:
"""
Filters out any keyword arguments that are `None` from the specified dictionary.
Args:
kwargs (Dict[str, Any]): The dictionary to filter.
Returns:
Dict[str, Any]: The filtered dictionary.
"""
return {k: v for k, v in kwargs.items() if v is not None}

async def __aenter__(self):
"""
Expand Down

0 comments on commit 2b4c740

Please sign in to comment.