Skip to content

Core

griddy / core**

Core module for the Griddy SDK.

This module provides shared functionality used across all SDK modules:

  • BaseClient: HTTP client with rate limiting, retries, and error handling
  • Exceptions: Custom exception classes for different error scenarios
  • Models: Base Pydantic models for data validation
Example

from griddy.core import BaseClient, GriddyError client = BaseClient(base_url="https://api.example.com") try: ... data = client.get("/endpoint") ... except GriddyError as e: ... print(f"Error: {e.message}")


Module Contents

Core module for the Griddy SDK.

This module provides shared functionality used across all SDK modules:

  • BaseClient: HTTP client with rate limiting, retries, and error handling
  • Exceptions: Custom exception classes for different error scenarios
  • Models: Base Pydantic models for data validation
Example

from griddy.core import BaseClient, GriddyError client = BaseClient(base_url="https://api.example.com") try: ... data = client.get("/endpoint") ... except GriddyError as e: ... print(f"Error: {e.message}")

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
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
@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
@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
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()

APIError

APIError(
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
)

Bases: GriddyError


              flowchart TD
              griddy.core.APIError[APIError]
              griddy.core.exceptions.GriddyError[GriddyError]

                              griddy.core.exceptions.GriddyError --> griddy.core.APIError
                


              click griddy.core.APIError href "" "griddy.core.APIError"
              click griddy.core.exceptions.GriddyError href "" "griddy.core.exceptions.GriddyError"
            

Raised when an API request fails.

This exception is raised for general API errors, including server errors (5xx status codes) and client errors that don't fit more specific categories.

Example

try: ... result = nfl.games.get_games(season=2024) ... except APIError as e: ... if e.status_code and e.status_code >= 500: ... print("Server error - retry later") ... else: ... print(f"API error: {e.message}")

Source code in griddy/core/exceptions.py
def __init__(
    self,
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
) -> None:
    """Initialize a GriddyError.

    Args:
        message: Human-readable error description.
        status_code: HTTP status code from the API response.
        response_data: Raw response data from the API.
    """
    super().__init__(message)
    self.message = message
    self.status_code = status_code
    self.response_data = response_data or {}

AuthenticationError

AuthenticationError(
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
)

Bases: GriddyError


              flowchart TD
              griddy.core.AuthenticationError[AuthenticationError]
              griddy.core.exceptions.GriddyError[GriddyError]

                              griddy.core.exceptions.GriddyError --> griddy.core.AuthenticationError
                


              click griddy.core.AuthenticationError href "" "griddy.core.AuthenticationError"
              click griddy.core.exceptions.GriddyError href "" "griddy.core.exceptions.GriddyError"
            

Raised when authentication fails.

This exception is raised when the API returns a 401 status code, indicating the request lacks valid authentication credentials.

Example

try: ... result = nfl.games.get_games(season=2024) ... except AuthenticationError: ... print("Token expired. Please re-authenticate.")

Source code in griddy/core/exceptions.py
def __init__(
    self,
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
) -> None:
    """Initialize a GriddyError.

    Args:
        message: Human-readable error description.
        status_code: HTTP status code from the API response.
        response_data: Raw response data from the API.
    """
    super().__init__(message)
    self.message = message
    self.status_code = status_code
    self.response_data = response_data or {}

GriddyError

GriddyError(
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
)

Bases: Exception


              flowchart TD
              griddy.core.GriddyError[GriddyError]

              

              click griddy.core.GriddyError href "" "griddy.core.GriddyError"
            

Base exception for all Griddy SDK errors.

All SDK-specific exceptions inherit from this class, allowing you to catch all SDK errors with a single except clause.

ATTRIBUTE DESCRIPTION
message

Human-readable error description.

status_code

HTTP status code from the API response, if applicable.

response_data

Raw response data from the API, if available.

Example

try: ... result = nfl.games.get_games(season=2024) ... except GriddyError as e: ... print(f"Error: {e.message}") ... print(f"Status: {e.status_code}") ... print(f"Data: {e.response_data}")

Initialize a GriddyError.

PARAMETER DESCRIPTION
message

Human-readable error description.

TYPE: str

status_code

HTTP status code from the API response.

TYPE: int | None DEFAULT: None

response_data

Raw response data from the API.

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

Source code in griddy/core/exceptions.py
def __init__(
    self,
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
) -> None:
    """Initialize a GriddyError.

    Args:
        message: Human-readable error description.
        status_code: HTTP status code from the API response.
        response_data: Raw response data from the API.
    """
    super().__init__(message)
    self.message = message
    self.status_code = status_code
    self.response_data = response_data or {}

Functions

NotFoundError

NotFoundError(
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
)

Bases: GriddyError


              flowchart TD
              griddy.core.NotFoundError[NotFoundError]
              griddy.core.exceptions.GriddyError[GriddyError]

                              griddy.core.exceptions.GriddyError --> griddy.core.NotFoundError
                


              click griddy.core.NotFoundError href "" "griddy.core.NotFoundError"
              click griddy.core.exceptions.GriddyError href "" "griddy.core.exceptions.GriddyError"
            

Raised when a requested resource is not found.

This exception is raised when the API returns a 404 status code, indicating the requested resource does not exist.

Example

try: ... player = nfl.players.get_player(player_id="invalid_id") ... except NotFoundError as e: ... print(f"Player not found: {e.message}") ... player = None

Source code in griddy/core/exceptions.py
def __init__(
    self,
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
) -> None:
    """Initialize a GriddyError.

    Args:
        message: Human-readable error description.
        status_code: HTTP status code from the API response.
        response_data: Raw response data from the API.
    """
    super().__init__(message)
    self.message = message
    self.status_code = status_code
    self.response_data = response_data or {}

RateLimitError

RateLimitError(
    message: str,
    retry_after: int | None = None,
    **kwargs: Any
)

Bases: GriddyError


              flowchart TD
              griddy.core.RateLimitError[RateLimitError]
              griddy.core.exceptions.GriddyError[GriddyError]

                              griddy.core.exceptions.GriddyError --> griddy.core.RateLimitError
                


              click griddy.core.RateLimitError href "" "griddy.core.RateLimitError"
              click griddy.core.exceptions.GriddyError href "" "griddy.core.exceptions.GriddyError"
            

Raised when the API rate limit is exceeded.

This exception is raised when the API returns a 429 status code, indicating too many requests have been made in a given time period.

ATTRIBUTE DESCRIPTION
retry_after

Number of seconds to wait before retrying, if provided by the API in the Retry-After header.

Example

import time try: ... result = nfl.games.get_games(season=2024) ... except RateLimitError as e: ... wait_time = e.retry_after or 60 ... print(f"Rate limited. Waiting {wait_time} seconds...") ... time.sleep(wait_time)

Initialize a RateLimitError.

PARAMETER DESCRIPTION
message

Human-readable error description.

TYPE: str

retry_after

Seconds to wait before retrying, from Retry-After header.

TYPE: int | None DEFAULT: None

**kwargs

Additional arguments passed to GriddyError.

TYPE: Any DEFAULT: {}

Source code in griddy/core/exceptions.py
def __init__(
    self,
    message: str,
    retry_after: int | None = None,
    **kwargs: Any,
) -> None:
    """Initialize a RateLimitError.

    Args:
        message: Human-readable error description.
        retry_after: Seconds to wait before retrying, from Retry-After header.
        **kwargs: Additional arguments passed to GriddyError.
    """
    super().__init__(message, **kwargs)
    self.retry_after = retry_after

Functions

ValidationError

ValidationError(
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
)

Bases: GriddyError


              flowchart TD
              griddy.core.ValidationError[ValidationError]
              griddy.core.exceptions.GriddyError[GriddyError]

                              griddy.core.exceptions.GriddyError --> griddy.core.ValidationError
                


              click griddy.core.ValidationError href "" "griddy.core.ValidationError"
              click griddy.core.exceptions.GriddyError href "" "griddy.core.exceptions.GriddyError"
            

Raised when request validation fails.

This exception is raised when the request parameters fail validation, either client-side or when the API returns a 400 status code.

Example

try: ... result = nfl.games.get_games(season="invalid") ... except ValidationError as e: ... print(f"Invalid parameters: {e.message}")

Source code in griddy/core/exceptions.py
def __init__(
    self,
    message: str,
    status_code: int | None = None,
    response_data: dict[str, Any] | None = None,
) -> None:
    """Initialize a GriddyError.

    Args:
        message: Human-readable error description.
        status_code: HTTP status code from the API response.
        response_data: Raw response data from the API.
    """
    super().__init__(message)
    self.message = message
    self.status_code = status_code
    self.response_data = response_data or {}

BaseModel

Bases: BaseModel


              flowchart TD
              griddy.core.BaseModel[BaseModel]

              

              click griddy.core.BaseModel href "" "griddy.core.BaseModel"
            

Base model for all Griddy data structures.

Classes

Config

Pydantic configuration.