Concepts and Design

Authentication Flow

The ADP API Client uses the OAuth 2.0 Client Credentials flow with certificate-based authentication. This flow is designed for server-to-server communication where the application acts on its own behalf.

Process

The client presents its credentials (client ID and secret) along with client certificate
The ADP authentication service validates the certificate and credentials
An access token is issued with a specific validity period
The client uses this token to make API requests
When the token approaches expiration, it's automatically refreshed

Credential Loading

The AdpCredentials object holds all API credentials, and supports loading from environment variables

from dotenv import load_dotenv
from adpapi.client import AdpCredentials

creds = AdpCredentials.from_env()

NOTE: .from_env expects the following environment variables - CLIENT_ID and CLIENT_SECRET, mandatory, link to the API project - CERT_PATH and KEY_PATH, optional, default to certificate.pem and adp.key, respectively if not provided

`call_endpoint` vs `call_rest_endpoint`

The client exposes two methods that serve different purposes:

call_endpoint — for paginated OData endpoints. Use this when you want to list or search records. It automatically handles $top/$skip pagination and supports $select and $filter query parameters.

call_rest_endpoint — for direct resource lookups. Use this when you already know the resource identifier and want to fetch it by ID. It supports path parameter templates (e.g. /hr/v2/workers/{associateOID}) and can batch-fetch a list of IDs in a single call.

Parallel Batch Requests

call_rest_endpoint supports concurrent fetching through the max_workers parameter. Under the hood, this uses Python's concurrent.futures.ThreadPoolExecutor to dispatch multiple HTTP requests simultaneously.

Why threads?

API calls are I/O-bound (most time is spent waiting for network responses), so Python's threading model works well here despite the GIL. Each thread simply waits for its own HTTP response, allowing many requests to be in-flight at once.

Token safety

Before dispatching the thread pool, call_rest_endpoint calls _ensure_valid_token() once. This guarantees that all threads share the same fresh token and avoids race conditions from multiple threads trying to refresh simultaneously.

Performance characteristics

Benchmarks with real ADP API calls show:

Workers	10 IDs	50 IDs
1 (sequential)	~11s	~44s
10 (parallel)	~4s	~8s

Scaling beyond 10 workers typically offers diminishing returns because the bottleneck shifts to ADP server processing time and rate limits.

OData Filtering

The ADP API uses OData query syntax for filtering results. The FilterExpression class provides a Pythonic way to build these filters:

Bases: Expr

Public API for creating and managing OData filter expressions.

Attributes:

Name	Type	Description
`_node`	`Expr`	The internal AST node representing the expression.

Examples:

Build filters programmatically:

>>> f = FilterExpression.field('firstName').eq('John')
>>> f.to_odata()
"(firstName eq 'John')"

Combine with logical operators:

>>> f1 = FilterExpression.field('age').gt(18)
>>> f2 = FilterExpression.field('status').eq('Active')
>>> combined = f1 & f2
>>> combined.to_odata()
"((age gt 18) and (status eq 'Active'))"

Parse existing OData filter strings:

>>> f = FilterExpression.from_string("firstName eq 'John'")
>>> f.to_odata()
"(firstName eq 'John')"

Source code in src/adpapi/odata_filters.py

class FilterExpression(Expr):
    """Public API for creating and managing OData filter expressions.

    Attributes:
        _node (Expr): The internal AST node representing the expression.

    Examples:
        Build filters programmatically:
        >>> f = FilterExpression.field('firstName').eq('John')
        >>> f.to_odata()
        "(firstName eq 'John')"

        Combine with logical operators:
        >>> f1 = FilterExpression.field('age').gt(18)
        >>> f2 = FilterExpression.field('status').eq('Active')
        >>> combined = f1 & f2
        >>> combined.to_odata()
        "((age gt 18) and (status eq 'Active'))"

        Parse existing OData filter strings:
        >>> f = FilterExpression.from_string("firstName eq 'John'")
        >>> f.to_odata()
        "(firstName eq 'John')"
    """

    def __init__(self, node: Expr):
        """Initialize a FilterExpression with an AST node.

        Args:
            node: An Expr AST node representing the filter expression.
        """
        self._node = node

    # façade pass-through
    def to_odata(self) -> str:
        """Convert this filter expression to an OData v4 filter string.

        Returns:
            str: The complete OData filter string ready for use in API requests.

        Example:
            >>> FilterExpression.field('status').eq('Active').to_odata()
            "(status eq 'Active')"
        """
        return self._node.to_odata()

    # convenience constructors
    @staticmethod
    def field(path: str) -> Field:
        """Create a field reference for building filter conditions.

        This is the primary entry point for building filters. The returned Field
        object provides a fluent API with comparison and string function methods.

        Args:
            path (str): Dot-separated field path (e.g., 'worker.person.firstName').
                       Supports nested properties accessible through the API.

        Returns:
            Field: A Field object with methods for building conditions.

        Example:
            >>> f = FilterExpression.field('lastName')
            >>> f = f.eq('Smith')
            >>> f.to_odata()
            "(lastName eq 'Smith')"

        Commonly used field paths:
            - 'worker.firstName' - Worker's first name
            - 'worker.lastName' - Worker's last name
            - 'hireDate' - Date of hire
            - 'department' - Department assignment
            - 'salary' - Salary information
        """
        return Field(path)

    # parse a limited OData subset into an AST
    @staticmethod
    def from_string(s: str) -> "FilterExpression":
        """Parse an OData filter string into a FilterExpression.

        This parser supports a subset of OData v4 filter syntax, including:
        - Comparison operators: eq, ne, gt, ge, lt, le
        - Logical operators: and, or, not
        - Boolean operators with parentheses for grouping
        - String functions: contains(), startswith(), endswith()
        - Literal values: strings, numbers, booleans, null
        - Field paths with dot notation

        Args:
            s (str): An OData filter string to parse.

        Returns:
            FilterExpression: A parsed and structured filter expression.

        Raises:
            ValueError: If the filter string has syntax errors or uses unsupported
                       OData features.

        Example:
            >>> f = FilterExpression.from_string(
            ...     "(firstName eq 'John') and (lastName eq 'Doe')"
            ... )
            >>> f.to_odata()
            "((firstName eq 'John') and (lastName eq 'Doe'))"

            Parse a string function:
            >>> f = FilterExpression.from_string(
            ...     "contains(email, '@company.com')"
            ... )
            >>> f.to_odata()
            "contains(email, '@company.com')"
        """
        node = _FilterParser(s).parse()
        return FilterExpression(node)

    # combinators keep returning FilterExpression
    def __and__(self, other: Expr) -> "FilterExpression":
        """Combine this expression with another using logical AND.

        Args:
            other: Another FilterExpression or Expr to combine with AND.

        Returns:
            FilterExpression: A new combined filter expression.

        Example:
            >>> expr1 = FilterExpression.field('age').gt(18)
            >>> expr2 = FilterExpression.field('status').eq('Active')
            >>> combined = expr1 & expr2
            >>> combined.to_odata()
            "((age gt 18) and (status eq 'Active'))"
        """
        return FilterExpression(BinaryOp(self._node, "and", _unwrap(other)))

    def __or__(self, other: Expr) -> "FilterExpression":
        """Combine this expression with another using logical OR.

        Args:
            other: Another FilterExpression or Expr to combine with OR.

        Returns:
            FilterExpression: A new combined filter expression.

        Example:
            >>> expr1 = FilterExpression.field('status').eq('Active')
            >>> expr2 = FilterExpression.field('status').eq('Pending')
            >>> combined = expr1 | expr2
            >>> combined.to_odata()
            "((status eq 'Active') or (status eq 'Pending'))"
        """
        return FilterExpression(BinaryOp(self._node, "or", _unwrap(other)))

    def __invert__(self) -> "FilterExpression":
        """Invert this expression using logical NOT.

        Returns:
            FilterExpression: A new inverted filter expression.

        Example:
            >>> expr = FilterExpression.field('isTerminated').eq(True)
            >>> inverted = ~expr
            >>> inverted.to_odata()
            "(not (isTerminated eq true))"
        """
        return FilterExpression(UnaryOp("not", self._node))

`and(other)`

Combine this expression with another using logical AND.

Parameters:

Name	Type	Description	Default
`other`	`Expr`	Another FilterExpression or Expr to combine with AND.	required

Returns:

Name	Type	Description
`FilterExpression`	`FilterExpression`	A new combined filter expression.

Example

expr1 = FilterExpression.field('age').gt(18) expr2 = FilterExpression.field('status').eq('Active') combined = expr1 & expr2 combined.to_odata() "((age gt 18) and (status eq 'Active'))"

Source code in src/adpapi/odata_filters.py

def __and__(self, other: Expr) -> "FilterExpression":
    """Combine this expression with another using logical AND.

    Args:
        other: Another FilterExpression or Expr to combine with AND.

    Returns:
        FilterExpression: A new combined filter expression.

    Example:
        >>> expr1 = FilterExpression.field('age').gt(18)
        >>> expr2 = FilterExpression.field('status').eq('Active')
        >>> combined = expr1 & expr2
        >>> combined.to_odata()
        "((age gt 18) and (status eq 'Active'))"
    """
    return FilterExpression(BinaryOp(self._node, "and", _unwrap(other)))

`init(node)`

Initialize a FilterExpression with an AST node.

Parameters:

Name	Type	Description	Default
`node`	`Expr`	An Expr AST node representing the filter expression.	required

Source code in src/adpapi/odata_filters.py

def __init__(self, node: Expr):
    """Initialize a FilterExpression with an AST node.

    Args:
        node: An Expr AST node representing the filter expression.
    """
    self._node = node

`invert()`

Invert this expression using logical NOT.

Returns:

Name	Type	Description
`FilterExpression`	`FilterExpression`	A new inverted filter expression.

Example

expr = FilterExpression.field('isTerminated').eq(True) inverted = ~expr inverted.to_odata() "(not (isTerminated eq true))"

Source code in src/adpapi/odata_filters.py

def __invert__(self) -> "FilterExpression":
    """Invert this expression using logical NOT.

    Returns:
        FilterExpression: A new inverted filter expression.

    Example:
        >>> expr = FilterExpression.field('isTerminated').eq(True)
        >>> inverted = ~expr
        >>> inverted.to_odata()
        "(not (isTerminated eq true))"
    """
    return FilterExpression(UnaryOp("not", self._node))

`or(other)`

Combine this expression with another using logical OR.

Parameters:

Name	Type	Description	Default
`other`	`Expr`	Another FilterExpression or Expr to combine with OR.	required

Returns:

Name	Type	Description
`FilterExpression`	`FilterExpression`	A new combined filter expression.

Example

expr1 = FilterExpression.field('status').eq('Active') expr2 = FilterExpression.field('status').eq('Pending') combined = expr1 | expr2 combined.to_odata() "((status eq 'Active') or (status eq 'Pending'))"

Source code in src/adpapi/odata_filters.py

def __or__(self, other: Expr) -> "FilterExpression":
    """Combine this expression with another using logical OR.

    Args:
        other: Another FilterExpression or Expr to combine with OR.

    Returns:
        FilterExpression: A new combined filter expression.

    Example:
        >>> expr1 = FilterExpression.field('status').eq('Active')
        >>> expr2 = FilterExpression.field('status').eq('Pending')
        >>> combined = expr1 | expr2
        >>> combined.to_odata()
        "((status eq 'Active') or (status eq 'Pending'))"
    """
    return FilterExpression(BinaryOp(self._node, "or", _unwrap(other)))

`field(path)` `staticmethod`

Create a field reference for building filter conditions.

This is the primary entry point for building filters. The returned Field object provides a fluent API with comparison and string function methods.

Parameters:

Name	Type	Description	Default
`path`	`str`	Dot-separated field path (e.g., 'worker.person.firstName'). Supports nested properties accessible through the API.	required

Returns:

Name	Type	Description
`Field`	`Field`	A Field object with methods for building conditions.

Example

f = FilterExpression.field('lastName') f = f.eq('Smith') f.to_odata() "(lastName eq 'Smith')"

Commonly used field paths

'worker.firstName' - Worker's first name
'worker.lastName' - Worker's last name
'hireDate' - Date of hire
'department' - Department assignment
'salary' - Salary information

Source code in src/adpapi/odata_filters.py

@staticmethod
def field(path: str) -> Field:
    """Create a field reference for building filter conditions.

    This is the primary entry point for building filters. The returned Field
    object provides a fluent API with comparison and string function methods.

    Args:
        path (str): Dot-separated field path (e.g., 'worker.person.firstName').
                   Supports nested properties accessible through the API.

    Returns:
        Field: A Field object with methods for building conditions.

    Example:
        >>> f = FilterExpression.field('lastName')
        >>> f = f.eq('Smith')
        >>> f.to_odata()
        "(lastName eq 'Smith')"

    Commonly used field paths:
        - 'worker.firstName' - Worker's first name
        - 'worker.lastName' - Worker's last name
        - 'hireDate' - Date of hire
        - 'department' - Department assignment
        - 'salary' - Salary information
    """
    return Field(path)

`from_string(s)` `staticmethod`

Parse an OData filter string into a FilterExpression.

This parser supports a subset of OData v4 filter syntax, including: - Comparison operators: eq, ne, gt, ge, lt, le - Logical operators: and, or, not - Boolean operators with parentheses for grouping - String functions: contains(), startswith(), endswith() - Literal values: strings, numbers, booleans, null - Field paths with dot notation

Parameters:

Name	Type	Description	Default
`s`	`str`	An OData filter string to parse.	required

Returns:

Name	Type	Description
`FilterExpression`	`FilterExpression`	A parsed and structured filter expression.

Raises:

Type	Description
`ValueError`	If the filter string has syntax errors or uses unsupported OData features.

Example

f = FilterExpression.from_string( ... "(firstName eq 'John') and (lastName eq 'Doe')" ... ) f.to_odata() "((firstName eq 'John') and (lastName eq 'Doe'))"

Parse a string function:

f = FilterExpression.from_string( ... "contains(email, '@company.com')" ... ) f.to_odata() "contains(email, '@company.com')"

Source code in src/adpapi/odata_filters.py

@staticmethod
def from_string(s: str) -> "FilterExpression":
    """Parse an OData filter string into a FilterExpression.

    This parser supports a subset of OData v4 filter syntax, including:
    - Comparison operators: eq, ne, gt, ge, lt, le
    - Logical operators: and, or, not
    - Boolean operators with parentheses for grouping
    - String functions: contains(), startswith(), endswith()
    - Literal values: strings, numbers, booleans, null
    - Field paths with dot notation

    Args:
        s (str): An OData filter string to parse.

    Returns:
        FilterExpression: A parsed and structured filter expression.

    Raises:
        ValueError: If the filter string has syntax errors or uses unsupported
                   OData features.

    Example:
        >>> f = FilterExpression.from_string(
        ...     "(firstName eq 'John') and (lastName eq 'Doe')"
        ... )
        >>> f.to_odata()
        "((firstName eq 'John') and (lastName eq 'Doe'))"

        Parse a string function:
        >>> f = FilterExpression.from_string(
        ...     "contains(email, '@company.com')"
        ... )
        >>> f.to_odata()
        "contains(email, '@company.com')"
    """
    node = _FilterParser(s).parse()
    return FilterExpression(node)

`to_odata()`

Convert this filter expression to an OData v4 filter string.

Returns:

Name	Type	Description
`str`	`str`	The complete OData filter string ready for use in API requests.

Example

FilterExpression.field('status').eq('Active').to_odata() "(status eq 'Active')"

Source code in src/adpapi/odata_filters.py

def to_odata(self) -> str:
    """Convert this filter expression to an OData v4 filter string.

    Returns:
        str: The complete OData filter string ready for use in API requests.

    Example:
        >>> FilterExpression.field('status').eq('Active').to_odata()
        "(status eq 'Active')"
    """
    return self._node.to_odata()

Token Lifecycle

The client automatically manages token refresh. It tracks token expiration and refreshes the token with a 5-minute buffer before actual expiration to prevent failed requests.

Error Handling

The client includes built-in retry logic for transient failures using exponential backoff. This helps handle temporary network issues gracefully.

Logging

Comprehensive logging is available at multiple levels (DEBUG, INFO, WARNING, ERROR) to help with troubleshooting:

Source code in src/adpapi/logger.py

def configure_logging():
    logging.basicConfig(
        filename="app.log",
        filemode="w",
        level=logging.DEBUG,
        format="%(asctime)s - %(levelname)s - %(message)s",
    )
    # Add console handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
    logging.getLogger().addHandler(console_handler)

Session Management

The ApiSession dataclass manages HTTP sessions, retry logic, and authentication headers. It encapsulates:

A requests.Session object for connection pooling
Client certificate information for SSL/TLS authentication
A callback function for dynamic header generation (useful for token refresh)
Request timeout handling

Source code in src/adpapi/sessions.py

@dataclass
class ApiSession:
    session: requests.Session
    cert: tuple[str, str]
    get_headers: Callable[[], dict] | None = None
    headers: dict | None = None
    params: dict | None = None
    timeout: int = 30
    data: Any | None = None

    def __post_init__(self):
        if self.get_headers is None:
            # Default to empty header generation
            self.get_headers = lambda: {}
        if self.params is None:
            self.params = {}

    def set_params(self, params: dict):
        self.params = params

    def set_data(self, data: Any):
        self.data = data

    def _get_request_function(self, method: RequestMethod) -> Callable:
        match method:
            case RequestMethod.GET:
                return self.session.get
            case RequestMethod.POST:
                return self.session.post
            case RequestMethod.PUT:
                return self.session.put
            case RequestMethod.DELETE:
                return self.session.delete

        raise ValueError(f"Unsupported method {method}")

    def _request(self, url: str, method: RequestMethod = RequestMethod.GET) -> requests.Response:
        """Execute HTTP request with specified method, headers, params, and optional data.

        Args:
            url: The request URL
            method: HTTP method (GET, POST, PUT, DELETE)

        Returns:
            requests.Response object

        Raises:
            requests.RequestException: If request fails
        """
        request_fn = self._get_request_function(method)
        # Generate headers on call time for up-to-date token
        assert self.get_headers is not None
        headers = self.get_headers()
        kwargs = {
            "headers": headers,
            "params": self.params,
            "cert": self.cert,
            "timeout": self.timeout,
        }
        if self.data is not None:
            kwargs["json"] = self.data
        response = request_fn(url, **kwargs)

        try:
            response.raise_for_status()

        except requests.RequestException as e:
            headers = dict(response.headers)
            data = response.json()
            logger.error(
                f"Request failed for {method} request to url: {url} with params {self.params}\n"
                f"Response Headers: {json.dumps(headers, indent=2)}\n"
                f"Response Body: {json.dumps(data, indent=2)}\n"
                f"Error:\n{e}"
            )
            raise

        return response

    def get(self, url: str) -> requests.Response:
        return self._request(url, RequestMethod.GET)

    def post(self, url: str, data: Any | None = None) -> requests.Response:
        if data is not None:
            self.set_data(data)
        return self._request(url, RequestMethod.POST)

    def put(self, url: str, data: Any | None = None) -> requests.Response:
        if data is not None:
            self.set_data(data)
        return self._request(url, RequestMethod.PUT)

    def delete(self, url: str) -> requests.Response:
        return self._request(url, RequestMethod.DELETE)

Concepts and Design

Authentication Flow

Process

Credential Loading

call_endpoint vs call_rest_endpoint

Parallel Batch Requests

Why threads?

Token safety

Performance characteristics

OData Filtering

__and__(other)

__init__(node)

__invert__()

__or__(other)

field(path) staticmethod

from_string(s) staticmethod

to_odata()

Token Lifecycle

Error Handling

Logging

Session Management

`call_endpoint` vs `call_rest_endpoint`

`and(other)`

`init(node)`

`invert()`

`or(other)`

`field(path)` `staticmethod`

`from_string(s)` `staticmethod`

`to_odata()`