Skip to content

Base Client

griddy / core / base_client**

base_client

Base HTTP client for all Griddy SDK modules.

This module provides the BaseClient class, which implements common HTTP functionality including rate limiting, automatic retries, and error handling.

Example

from griddy.core import BaseClient client = BaseClient( ... base_url="https://api.example.com", ... timeout=30, ... rate_limit_delay=1.0, ... ) response = client.get("/endpoint", params={"key": "value"}) client.close()

Classes

BaseClient 

BaseClient(
    base_url: str,
    timeout: int = 30,
    max_retries: int = 3,
    rate_limit_delay: float = 1.0,
    headers: Dict[str, str] | None = None,
    cookies_file: str | None = None,
)

Base HTTP client with common functionality for all data source clients.

This class provides a foundation for building API clients with built-in support for rate limiting, automatic retries on transient failures, and consistent error handling.

ATTRIBUTE DESCRIPTION
base_url

The base URL for all API requests.
timeout

Request timeout in seconds.
rate_limit_delay

Minimum delay between requests in seconds.
session

The underlying requests Session object.
cookies_file

Path to a cookies file for authentication.
Example

client = BaseClient( ... base_url="https://api.nfl.com", ... timeout=30, ... max_retries=3, ... rate_limit_delay=1.0, ... ) try: ... data = client.get("/games", params={"season": 2024}) ... finally: ... client.close()

Initialize the base client.

Creates an HTTP session with automatic retry capabilities for transient errors (429, 500, 502, 503, 504 status codes).

PARAMETER DESCRIPTION
base_url

Base URL for the API (e.g., "https://api.nfl.com"). Trailing slashes are automatically removed.

TYPE: str

timeout

Request timeout in seconds. Defaults to 30.

TYPE: int DEFAULT: 30

max_retries

Maximum number of retries for failed requests. Defaults to 3.

TYPE: int DEFAULT: 3

rate_limit_delay

Minimum delay between requests in seconds to avoid rate limiting. Defaults to 1.0.

TYPE: float DEFAULT: 1.0

headers

Additional headers to include in all requests. These are merged with default headers (Accept, Content-Type).

TYPE: Dict[str, str] | None DEFAULT: None

cookies_file

Path to a Netscape-format cookies file for authentication.

TYPE: str | None DEFAULT: None

Example

client = BaseClient( ... base_url="https://api.example.com", ... timeout=60, ... max_retries=5, ... rate_limit_delay=2.0, ... headers={"Authorization": "Bearer token"}, ... )

Source code in griddy/core/base_client.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def __init__(
    self,
    base_url: str,
    timeout: int = 30,
    max_retries: int = 3,
    rate_limit_delay: float = 1.0,
    headers: Dict[str, str] | None = None,
    cookies_file: str | None = None,
) -> None:
    """Initialize the base client.

    Creates an HTTP session with automatic retry capabilities for transient
    errors (429, 500, 502, 503, 504 status codes).

    Args:
        base_url: Base URL for the API (e.g., "https://api.nfl.com").
            Trailing slashes are automatically removed.
        timeout: Request timeout in seconds. Defaults to 30.
        max_retries: Maximum number of retries for failed requests.
            Defaults to 3.
        rate_limit_delay: Minimum delay between requests in seconds
            to avoid rate limiting. Defaults to 1.0.
        headers: Additional headers to include in all requests.
            These are merged with default headers (Accept, Content-Type).
        cookies_file: Path to a Netscape-format cookies file for
            authentication.

    Example:
        >>> client = BaseClient(
        ...     base_url="https://api.example.com",
        ...     timeout=60,
        ...     max_retries=5,
        ...     rate_limit_delay=2.0,
        ...     headers={"Authorization": "Bearer token"},
        ... )
    """
    self.base_url = base_url.rstrip("/")
    self.timeout = timeout
    self.rate_limit_delay = rate_limit_delay
    self._last_request_time = 0.0
    self.cookies_file = cookies_file

    # Create session with retry strategy
    self.session = requests.Session()

    # Configure retry strategy
    retry_strategy = Retry(
        total=max_retries,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "OPTIONS"],
        backoff_factor=1.0,
    )

    adapter = HTTPAdapter(max_retries=retry_strategy)
    self.session.mount("http://", adapter)
    self.session.mount("https://", adapter)

    # Set default headers
    default_headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
    }

    if headers:
        default_headers.update(headers)

    self.session.headers.update(default_headers)
Functions
get 
get(
    endpoint: str,
    params: Dict[str, Any] | None = None,
    headers: Dict[str, str] | None = None,
) -> Dict[str, Any] | List[Any]

Make a GET request to the API.

Automatically applies rate limiting and handles errors.

PARAMETER DESCRIPTION
endpoint

API endpoint path (e.g., "/games" or "players/123"). Leading slashes are handled automatically.

TYPE: str

params

Query parameters to include in the request.

TYPE: Dict[str, Any] | None DEFAULT: None

headers

Additional headers to include in this request only.

TYPE: Dict[str, str] | None DEFAULT: None

RETURNS DESCRIPTION
Dict[str, Any] | List[Any]

Parsed JSON response data.
RAISES DESCRIPTION
NotFoundError

When the resource is not found (404).
AuthenticationError

When authentication fails (401).
RateLimitError

When rate limit is exceeded (429).
APIError

For all other error status codes.
Example

data = client.get( ... "/games", ... params={"season": 2024, "week": 1}, ... )

Source code in griddy/core/base_client.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
@retry_on_rate_limit(max_retries=3)
def get(
    self,
    endpoint: str,
    params: Dict[str, Any] | None = None,
    headers: Dict[str, str] | None = None,
) -> Dict[str, Any] | List[Any]:
    """Make a GET request to the API.

    Automatically applies rate limiting and handles errors.

    Args:
        endpoint: API endpoint path (e.g., "/games" or "players/123").
            Leading slashes are handled automatically.
        params: Query parameters to include in the request.
        headers: Additional headers to include in this request only.

    Returns:
        Parsed JSON response data.

    Raises:
        NotFoundError: When the resource is not found (404).
        AuthenticationError: When authentication fails (401).
        RateLimitError: When rate limit is exceeded (429).
        APIError: For all other error status codes.

    Example:
        >>> data = client.get(
        ...     "/games",
        ...     params={"season": 2024, "week": 1},
        ... )
    """
    self._enforce_rate_limit()

    url = urljoin(self.base_url + "/", endpoint.lstrip("/"))

    response = self.session.get(
        url,
        params=params,
        headers=headers,
        timeout=self.timeout,
    )

    return self._handle_response(response)
post 
post(
    endpoint: str,
    data: Dict[str, Any] | None = None,
    json_data: Dict[str, Any] | None = None,
    headers: Dict[str, str] | None = None,
) -> Dict[str, Any]

Make a POST request to the API.

Automatically applies rate limiting and handles errors.

PARAMETER DESCRIPTION
endpoint

API endpoint path.

TYPE: str

data

Form data to send in the request body.

TYPE: Dict[str, Any] | None DEFAULT: None

json_data

JSON data to send in the request body.

TYPE: Dict[str, Any] | None DEFAULT: None

headers

Additional headers to include in this request only.

TYPE: Dict[str, str] | None DEFAULT: None

RETURNS DESCRIPTION
Dict[str, Any]

Parsed JSON response data.
RAISES DESCRIPTION
NotFoundError

When the resource is not found (404).
AuthenticationError

When authentication fails (401).
RateLimitError

When rate limit is exceeded (429).
APIError

For all other error status codes.
Example

result = client.post( ... "/auth/token", ... json_data={"username": "user", "password": "pass"}, ... )

Source code in griddy/core/base_client.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
@retry_on_rate_limit(max_retries=3)
def post(
    self,
    endpoint: str,
    data: Dict[str, Any] | None = None,
    json_data: Dict[str, Any] | None = None,
    headers: Dict[str, str] | None = None,
) -> Dict[str, Any]:
    """Make a POST request to the API.

    Automatically applies rate limiting and handles errors.

    Args:
        endpoint: API endpoint path.
        data: Form data to send in the request body.
        json_data: JSON data to send in the request body.
        headers: Additional headers to include in this request only.

    Returns:
        Parsed JSON response data.

    Raises:
        NotFoundError: When the resource is not found (404).
        AuthenticationError: When authentication fails (401).
        RateLimitError: When rate limit is exceeded (429).
        APIError: For all other error status codes.

    Example:
        >>> result = client.post(
        ...     "/auth/token",
        ...     json_data={"username": "user", "password": "pass"},
        ... )
    """
    self._enforce_rate_limit()

    url = urljoin(self.base_url + "/", endpoint.lstrip("/"))

    response = self.session.post(
        url,
        data=data,
        json=json_data,
        headers=headers,
        timeout=self.timeout,
    )

    return self._handle_response(response)
close 
close() -> None

Close the HTTP session and release resources.

Should be called when the client is no longer needed to properly clean up connections.

Example

client = BaseClient(base_url="https://api.example.com") try: ... data = client.get("/endpoint") ... finally: ... client.close()

Source code in griddy/core/base_client.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def close(self) -> None:
    """Close the HTTP session and release resources.

    Should be called when the client is no longer needed to properly
    clean up connections.

    Example:
        >>> client = BaseClient(base_url="https://api.example.com")
        >>> try:
        ...     data = client.get("/endpoint")
        ... finally:
        ...     client.close()
    """
    self.session.close()

Functions