Skip to content

Commit

Permalink
Add an optional global timeout
Browse files Browse the repository at this point in the history
  • Loading branch information
bfontaine committed Nov 15, 2023
1 parent e3643d0 commit 78b762b
Showing 1 changed file with 10 additions and 2 deletions.
12 changes: 10 additions & 2 deletions api_session/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from typing import Optional, Union, Text, Dict, Any
from typing import Optional, Union, Text, Dict, Any, Tuple
from urllib3 import Timeout

import requests

Expand All @@ -18,7 +19,8 @@ class APISession(requests.Session):
def __init__(self, base_url: str, user_agent: Optional[str] = None, read_only=False, *,
offline=False,
none_on_404=True,
none_on_empty=False):
none_on_empty=False,
timeout: Optional[Union[int, Tuple[int, int], Timeout]] = None):
"""
:param base_url: Base URL of the API.
:param user_agent: Optional user-agent header to use.
Expand All @@ -27,6 +29,7 @@ def __init__(self, base_url: str, user_agent: Optional[str] = None, read_only=Fa
be overridden by passing it explicitly when calling the method.
:param none_on_empty: set the default for the argument of the same name in ``.get_json_api`` calls. This can
still be overridden by passing it explicitly when calling the method.
:param timeout: default timeout to use for all requests
"""
super().__init__()

Expand All @@ -35,6 +38,7 @@ def __init__(self, base_url: str, user_agent: Optional[str] = None, read_only=Fa
self.offline = offline
self.none_on_404 = none_on_404
self.none_on_empty = none_on_empty
self.timeout = timeout

if user_agent is not None:
self.headers['User-Agent'] = user_agent
Expand Down Expand Up @@ -62,6 +66,10 @@ def request(self, method: Union[str, bytes], url: Union[str, bytes, Text], *args

if self.read_only and not bypass_read_only and method.upper() not in self.READ_METHODS:
raise AssertionError("Can't perform %r action in read-only mode!" % method)

if "timeout" not in kwargs and self.timeout is not None:
kwargs["timeout"] = self.timeout

return super().request(method, url, *args, **kwargs)

def request_api(self, method: str, path: str, *args, throw: Optional[bool] = None, **kwargs):
Expand Down

0 comments on commit 78b762b

Please sign in to comment.