API Reference

Credential Management

To use the AdpApiClient you first must configure your API credentials. These credentials are managed through an AdpCredentials object, which can be configured manually or from environment variables.

Container for ADP API authentication credentials.

Holds the OAuth 2.0 client credentials and paths to the mTLS certificate and private key required for ADP API authentication.

Attributes:

Name	Type	Description
`client_id`	`str`	OAuth 2.0 client ID from ADP
`client_secret`	`str`	OAuth 2.0 client secret from ADP
`cert_path`	`str \| None`	Path to the mTLS certificate file (.pem). Defaults to 'certificate.pem'
`key_path`	`str \| None`	Path to the private key file. Defaults to 'adp.key'

Example

credentials = AdpCredentials( ... client_id="my_client_id", ... client_secret="my_secret", ... cert_path="/path/to/cert.pem", ... key_path="/path/to/key.key" ... )

Source code in src/adpapi/client.py

@dataclass
class AdpCredentials:
    """Container for ADP API authentication credentials.

    Holds the OAuth 2.0 client credentials and paths to the mTLS certificate
    and private key required for ADP API authentication.

    Attributes:
        client_id: OAuth 2.0 client ID from ADP
        client_secret: OAuth 2.0 client secret from ADP
        cert_path: Path to the mTLS certificate file (.pem). Defaults to 'certificate.pem'
        key_path: Path to the private key file. Defaults to 'adp.key'

    Example:
        >>> credentials = AdpCredentials(
        ...     client_id="my_client_id",
        ...     client_secret="my_secret",
        ...     cert_path="/path/to/cert.pem",
        ...     key_path="/path/to/key.key"
        ... )
    """

    client_id: str
    client_secret: str
    cert_path: str | None = CERT_DEFAULT
    key_path: str | None = KEY_DEFAULT

    @staticmethod
    def from_env() -> "AdpCredentials":
        """Load credentials from environment variables.

        Reads authentication credentials from the following environment variables:
        - CLIENT_ID (required): OAuth 2.0 client ID
        - CLIENT_SECRET (required): OAuth 2.0 client secret
        - CERT_PATH (optional): Path to mTLS certificate, defaults to 'certificate.pem'
        - KEY_PATH (optional): Path to private key, defaults to 'adp.key'

        Returns:
            AdpCredentials instance populated from environment variables

        Raises:
            ValueError: If CLIENT_ID or CLIENT_SECRET are not set

        Example:
            >>> from dotenv import load_dotenv
            >>> load_dotenv()
            >>> credentials = AdpCredentials.from_env()
        """
        client_id = os.getenv("CLIENT_ID")
        client_secret = os.getenv("CLIENT_SECRET")

        # Read optional mTLS certificate/key paths (defaults assume files in project root).
        cert_path = os.getenv("CERT_PATH")
        key_path = os.getenv("KEY_PATH")

        if cert_path is None:
            logger.warning(
                f"No environment variables found for CERT_PATH, defaulting to {CERT_DEFAULT}"
            )

        if key_path is None:
            logger.warning(
                f"No environment variables found for KEY_PATH, defaulting to {KEY_DEFAULT}"
            )

        if client_id is None or client_secret is None:
            raise ValueError("CLIENT_ID and CLIENT_SECRET environment variables must be set")

        return AdpCredentials(client_id, client_secret, cert_path, key_path)

`from_env()` `staticmethod`

Load credentials from environment variables.

Reads authentication credentials from the following environment variables: - CLIENT_ID (required): OAuth 2.0 client ID - CLIENT_SECRET (required): OAuth 2.0 client secret - CERT_PATH (optional): Path to mTLS certificate, defaults to 'certificate.pem' - KEY_PATH (optional): Path to private key, defaults to 'adp.key'

Returns:

Type	Description
`AdpCredentials`	AdpCredentials instance populated from environment variables

Raises:

Type	Description
`ValueError`	If CLIENT_ID or CLIENT_SECRET are not set

Example

from dotenv import load_dotenv load_dotenv() credentials = AdpCredentials.from_env()

Source code in src/adpapi/client.py

@staticmethod
def from_env() -> "AdpCredentials":
    """Load credentials from environment variables.

    Reads authentication credentials from the following environment variables:
    - CLIENT_ID (required): OAuth 2.0 client ID
    - CLIENT_SECRET (required): OAuth 2.0 client secret
    - CERT_PATH (optional): Path to mTLS certificate, defaults to 'certificate.pem'
    - KEY_PATH (optional): Path to private key, defaults to 'adp.key'

    Returns:
        AdpCredentials instance populated from environment variables

    Raises:
        ValueError: If CLIENT_ID or CLIENT_SECRET are not set

    Example:
        >>> from dotenv import load_dotenv
        >>> load_dotenv()
        >>> credentials = AdpCredentials.from_env()
    """
    client_id = os.getenv("CLIENT_ID")
    client_secret = os.getenv("CLIENT_SECRET")

    # Read optional mTLS certificate/key paths (defaults assume files in project root).
    cert_path = os.getenv("CERT_PATH")
    key_path = os.getenv("KEY_PATH")

    if cert_path is None:
        logger.warning(
            f"No environment variables found for CERT_PATH, defaulting to {CERT_DEFAULT}"
        )

    if key_path is None:
        logger.warning(
            f"No environment variables found for KEY_PATH, defaulting to {KEY_DEFAULT}"
        )

    if client_id is None or client_secret is None:
        raise ValueError("CLIENT_ID and CLIENT_SECRET environment variables must be set")

    return AdpCredentials(client_id, client_secret, cert_path, key_path)

Client

The main entry point for interacting with the ADP API. Using a context manager is recommended so the HTTP session is always closed cleanly:

from adpapi import AdpApiClient, AdpCredentials

credentials = AdpCredentials(
    client_id, client_secret, key_path, cert_path
)
with AdpApiClient(credentials) as api:
    api.call_endpoint(...)
    api.call_rest_endpoint(...)

Configuring Retry Behavior

Customize which HTTP status codes trigger automatic retries:

# Default: retries on [429, 500, 502, 503, 504]
client = AdpApiClient(credentials)

# Custom retry status codes
client = AdpApiClient(credentials, retry_on_statuses=[429, 503])

# Disable retries
client = AdpApiClient(credentials, retry_on_statuses=[])

The AdpApiClient surfaces two main entry points:

.call_endpoint() — for paginated OData queries (lists, searches)
.call_rest_endpoint() — for direct resource lookups by ID, with path parameter substitution

Both methods support select and filters parameters for OData column selection and filtering.

Call a paginated OData endpoint with automatic pagination handling.

Use this method for list/search operations that return multiple records. The client automatically handles pagination by incrementing $skip until the API returns HTTP 204 (No Content) or max_requests is reached.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	API endpoint path (e.g., '/hr/v2/workers'). Can optionally include the full URL, but path-only is preferred.	required
`select`	`list[str] \| None`	List of OData columns to retrieve. Uses dot notation in Python (e.g., 'workers/person/legalName') which is converted to OData's forward-slash notation. If None, all columns are returned.	`None`
`filters`	`str \| FilterExpression \| None`	OData filter expression as a string or FilterExpression object. Use FilterExpression for type-safe filter building.	`None`
`masked`	`bool`	Whether to request masked data (hides PII). Set to False to request unmasked data if your tenant permissions allow it. Defaults to True.	`True`
`timeout`	`int`	Request timeout in seconds. Defaults to 30.	`DEFAULT_TIMEOUT`
`page_size`	`int`	Number of records per request (max 100). Defaults to 100.	`100`
`max_requests`	`int \| None`	Maximum number of paginated requests to make. Useful for testing or limiting data pulls. If None, fetches all available records.	`None`
`method`	`str`	HTTP method to use. Defaults to 'GET'. Non-GET methods will only make a single request (no pagination).	`'GET'`

Returns:

Type	Description
`list[dict]`	List of dictionaries, where each dictionary is a response from the API.
`list[dict]`	For paginated GET requests, this will contain one dict per page.

Raises:

Type	Description
`ValueError`	If endpoint format is invalid (must start with '/' or base URL)
`RequestException`	If the HTTP request fails
`JSONDecodeError`	If the response body is not valid JSON

Example

Basic usage

workers = client.call_endpoint("/hr/v2/workers")

With column selection

workers = client.call_endpoint( ... "/hr/v2/workers", ... select=["workers/person/legalName", "workers/associateOID"] ... )

With filtering

from adpapi import FilterExpression active = FilterExpression.field("workers/status").eq("Active") workers = client.call_endpoint("/hr/v2/workers", filters=active)

Limited fetch for testing

workers = client.call_endpoint("/hr/v2/workers", max_requests=2)

Source code in src/adpapi/client.py

def call_endpoint(
    self,
    endpoint: str,
    select: list[str] | None = None,
    filters: str | FilterExpression | None = None,
    masked: bool = True,
    timeout: int = DEFAULT_TIMEOUT,
    page_size: int = 100,
    max_requests: int | None = None,
    method: str = "GET",
) -> list[dict]:
    """Call a paginated OData endpoint with automatic pagination handling.

    Use this method for list/search operations that return multiple records.
    The client automatically handles pagination by incrementing $skip until
    the API returns HTTP 204 (No Content) or max_requests is reached.

    Args:
        endpoint: API endpoint path (e.g., '/hr/v2/workers'). Can optionally include
            the full URL, but path-only is preferred.
        select: List of OData columns to retrieve. Uses dot notation in Python
            (e.g., 'workers/person/legalName') which is converted to OData's
            forward-slash notation. If None, all columns are returned.
        filters: OData filter expression as a string or FilterExpression object.
            Use FilterExpression for type-safe filter building.
        masked: Whether to request masked data (hides PII). Set to False to request
            unmasked data if your tenant permissions allow it. Defaults to True.
        timeout: Request timeout in seconds. Defaults to 30.
        page_size: Number of records per request (max 100). Defaults to 100.
        max_requests: Maximum number of paginated requests to make. Useful for
            testing or limiting data pulls. If None, fetches all available records.
        method: HTTP method to use. Defaults to 'GET'. Non-GET methods will only
            make a single request (no pagination).

    Returns:
        List of dictionaries, where each dictionary is a response from the API.
        For paginated GET requests, this will contain one dict per page.

    Raises:
        ValueError: If endpoint format is invalid (must start with '/' or base URL)
        requests.RequestException: If the HTTP request fails
        json.JSONDecodeError: If the response body is not valid JSON

    Example:
        >>> # Basic usage
        >>> workers = client.call_endpoint("/hr/v2/workers")
        >>>
        >>> # With column selection
        >>> workers = client.call_endpoint(
        ...     "/hr/v2/workers",
        ...     select=["workers/person/legalName", "workers/associateOID"]
        ... )
        >>>
        >>> # With filtering
        >>> from adpapi import FilterExpression
        >>> active = FilterExpression.field("workers/status").eq("Active")
        >>> workers = client.call_endpoint("/hr/v2/workers", filters=active)
        >>>
        >>> # Limited fetch for testing
        >>> workers = client.call_endpoint("/hr/v2/workers", max_requests=2)
    """

    # Request Cleanup and Validation Logic
    if page_size > 100:
        logger.warning("Page size > 100 not supported by API endpoint. Limiting to 100.")
        page_size = 100

    # Output/Request Initialization
    endpoint = self._clean_endpoint(endpoint)
    url = self.base_url + endpoint
    request_method = self._resolve_method(method)

    query_params = self._build_query_params(select=select, filters=filters)
    output = []
    skip = 0

    call_session = self._build_call_session(masked=masked, timeout=timeout)

    if request_method == RequestMethod.GET:
        query_params["$top"] = page_size
    elif max_requests is not None and max_requests > 1:
        logger.warning(
            "max_requests > 1 was provided for a non-GET request; only one request will be made."
        )

    while True:
        params = dict(query_params)
        if request_method == RequestMethod.GET:
            params["$skip"] = skip

        call_session.set_params(params)
        self._ensure_valid_token(timeout)
        response = call_session._request(url, method=request_method)

        if request_method == RequestMethod.GET and response.status_code == 204:
            logger.debug("End of pagination reached (204 No Content)")
            break

        # json.JSONDecodeError handling is centralized in _parse_json_response.
        data = self._parse_json_response(response)
        output.append(data)

        if request_method != RequestMethod.GET:
            break

        if max_requests is not None and len(output) >= max_requests:
            logger.debug(f"Max Requests reached: {max_requests}")
            break
        skip += page_size

    return output

Call a REST endpoint with path parameter substitution and optional parallelization.

Use this method for direct resource lookups when you already know the resource identifier(s). Supports batch fetching multiple resources and parallel execution for improved throughput.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	API endpoint path template with placeholders in curly braces (e.g., '/hr/v2/workers/{associateOID}').	required
`method`	`str`	HTTP method to use. Defaults to 'GET'.	`'GET'`
`masked`	`bool`	Whether to request masked data (hides PII). Set to False to request unmasked data if your tenant permissions allow it. Defaults to True.	`True`
`timeout`	`int`	Request timeout in seconds. Defaults to 30.	`DEFAULT_TIMEOUT`
`params`	`dict \| None`	Additional query parameters to include in the request.	`None`
`select`	`list[str] \| None`	List of OData columns to retrieve. Works the same as in call_endpoint().	`None`
`filters`	`str \| FilterExpression \| None`	OData filter expression as a string or FilterExpression object.	`None`
`max_workers`	`int`	Number of threads for parallel requests. Use 1 for sequential (default). Recommended range is 5-10 for parallel execution.	`1`
`inject_path_params`	`bool`	When True, the resolved path parameters are merged into each response dictionary. Useful when the API response doesn't include the requested identifier (e.g., associate OIDs). Defaults to False.	`False`
`**kwargs`	`Any`	Path parameters to substitute into the endpoint template. - Single values: workerId='123' → '/hr/v2/workers/123' - Lists: workerId=['123', '456'] → multiple requests for each ID - Multiple params: workerId='123', jobId='J1' → '/hr/v2/workers/123/jobs/J1'	`{}`

Returns:

Type	Description
`list[dict]`	List of dictionaries, one for each resolved endpoint URL. For batch requests
`list[dict]`	with lists, returns one response per list item.

Raises:

Type	Description
`ValueError`	If required path parameters are missing from kwargs, or if endpoint format is invalid.
`RequestException`	If the HTTP request fails.
`JSONDecodeError`	If the response body is not valid JSON.

Example

Single resource fetch

worker = client.call_rest_endpoint( ... "/hr/v2/workers/{associateOID}", ... associateOID="G3349PRDL000001" ... )

Batch fetch (sequential)

workers = client.call_rest_endpoint( ... "/hr/v2/workers/{associateOID}", ... associateOID=["G3349PRDL000001", "G3349PRDL000002"] ... )

Parallel batch fetch (5-10x faster)

workers = client.call_rest_endpoint( ... "/hr/v2/workers/{associateOID}", ... max_workers=10, ... associateOID=list_of_50_ids ... )

With column selection and ID injection

workers = client.call_rest_endpoint( ... "/hr/v2/workers/{associateOID}", ... select=["workers/person/legalName"], ... inject_path_params=True, ... associateOID=["G3349PRDL000001", "G3349PRDL000002"] ... )

Each response now includes: {"associateOID": "G3349PRDL000001", ...}

Multiple path parameters

job = client.call_rest_endpoint( ... "/hr/v2/workers/{associateOID}/jobs/{jobId}", ... associateOID="G3349PRDL000001", ... jobId="J42" ... )

Source code in src/adpapi/client.py

def call_rest_endpoint(
    self,
    endpoint: str,
    method: str = "GET",
    masked: bool = True,
    timeout: int = DEFAULT_TIMEOUT,
    params: dict | None = None,
    select: list[str] | None = None,
    filters: str | FilterExpression | None = None,
    max_workers: int = 1,
    inject_path_params: bool = False,
    **kwargs: Any,
) -> list[dict]:
    """Call a REST endpoint with path parameter substitution and optional parallelization.

    Use this method for direct resource lookups when you already know the resource
    identifier(s). Supports batch fetching multiple resources and parallel execution
    for improved throughput.

    Args:
        endpoint: API endpoint path template with placeholders in curly braces
            (e.g., '/hr/v2/workers/{associateOID}').
        method: HTTP method to use. Defaults to 'GET'.
        masked: Whether to request masked data (hides PII). Set to False to request
            unmasked data if your tenant permissions allow it. Defaults to True.
        timeout: Request timeout in seconds. Defaults to 30.
        params: Additional query parameters to include in the request.
        select: List of OData columns to retrieve. Works the same as in call_endpoint().
        filters: OData filter expression as a string or FilterExpression object.
        max_workers: Number of threads for parallel requests. Use 1 for sequential
            (default). Recommended range is 5-10 for parallel execution.
        inject_path_params: When True, the resolved path parameters are merged into
            each response dictionary. Useful when the API response doesn't include
            the requested identifier (e.g., associate OIDs). Defaults to False.
        **kwargs: Path parameters to substitute into the endpoint template.
            - Single values: workerId='123' → '/hr/v2/workers/123'
            - Lists: workerId=['123', '456'] → multiple requests for each ID
            - Multiple params: workerId='123', jobId='J1' → '/hr/v2/workers/123/jobs/J1'

    Returns:
        List of dictionaries, one for each resolved endpoint URL. For batch requests
        with lists, returns one response per list item.

    Raises:
        ValueError: If required path parameters are missing from kwargs, or if endpoint
            format is invalid.
        requests.RequestException: If the HTTP request fails.
        json.JSONDecodeError: If the response body is not valid JSON.

    Example:
        >>> # Single resource fetch
        >>> worker = client.call_rest_endpoint(
        ...     "/hr/v2/workers/{associateOID}",
        ...     associateOID="G3349PRDL000001"
        ... )
        >>>
        >>> # Batch fetch (sequential)
        >>> workers = client.call_rest_endpoint(
        ...     "/hr/v2/workers/{associateOID}",
        ...     associateOID=["G3349PRDL000001", "G3349PRDL000002"]
        ... )
        >>>
        >>> # Parallel batch fetch (5-10x faster)
        >>> workers = client.call_rest_endpoint(
        ...     "/hr/v2/workers/{associateOID}",
        ...     max_workers=10,
        ...     associateOID=list_of_50_ids
        ... )
        >>>
        >>> # With column selection and ID injection
        >>> workers = client.call_rest_endpoint(
        ...     "/hr/v2/workers/{associateOID}",
        ...     select=["workers/person/legalName"],
        ...     inject_path_params=True,
        ...     associateOID=["G3349PRDL000001", "G3349PRDL000002"]
        ... )
        >>> # Each response now includes: {"associateOID": "G3349PRDL000001", ...}
        >>>
        >>> # Multiple path parameters
        >>> job = client.call_rest_endpoint(
        ...     "/hr/v2/workers/{associateOID}/jobs/{jobId}",
        ...     associateOID="G3349PRDL000001",
        ...     jobId="J42"
        ... )
    """
    endpoint = self._clean_endpoint(endpoint)
    is_valid, missing_params = validate_path_parameters(endpoint, kwargs)
    if not is_valid:
        raise ValueError(f"Missing required path parameters: {', '.join(missing_params)}")

    urls = substitute_path_parameters(endpoint, kwargs)
    if not urls:
        return []

    request_method = self._resolve_method(method)
    query_params = self._build_query_params(params=params, select=select, filters=filters)
    call_session = self._build_call_session(masked=masked, timeout=timeout, params=query_params)

    # Ensure a valid token once before all requests to avoid race conditions
    # with concurrent threads each trying to refresh the token simultaneously.
    self._ensure_valid_token(timeout)

    def _fetch(url: str) -> dict:
        full_url = self.base_url + url
        response = call_session._request(url=full_url, method=request_method)
        return self._parse_json_response(response)

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        output = list(executor.map(_fetch, urls))

    if inject_path_params:
        param_sets = resolve_path_parameter_sets(endpoint, kwargs)
        for response, param_set in zip(output, param_sets, strict=False):
            response.update(param_set)

    return output

Filters

OData filter expressions for querying. Use FilterExpression.field() as the primary entry point:

from adpapi import FilterExpression

# Recommended: use FilterExpression.field()
filter = FilterExpression.field('fieldName').eq('targetValue')

# Advanced: construct a Field directly for reuse
from adpapi.odata_filters import Field
field = Field('fieldName')

Filters can be used with both call_endpoint and call_rest_endpoint.

See full details on supported OData operations:

Bases: Expr

Represents a field reference in an OData filter expression.

Fields are identified by their path (e.g., 'worker.person.firstName'). This class provides a fluent API for building filter conditions on fields.

Attributes:

Name	Type	Description
`path`	`str`	The dot-separated path to the field, supporting nested properties.

Example

field = Field('worker.hireDate') field.eq('2020-01-01').to_odata() "(worker/hireDate eq '2020-01-01')"

Source code in src/adpapi/odata_filters.py

@dataclass(frozen=True)
class Field(Expr):
    """Represents a field reference in an OData filter expression.

    Fields are identified by their path (e.g., 'worker.person.firstName').
    This class provides a fluent API for building filter conditions on fields.

    Attributes:
        path (str): The dot-separated path to the field, supporting nested properties.

    Example:
        >>> field = Field('worker.hireDate')
        >>> field.eq('2020-01-01').to_odata()
        "(worker/hireDate eq '2020-01-01')"
    """

    path: str

    # comparisons
    def eq(self, val: Any) -> "BinaryOp":
        """Create an equality comparison filter (field = value).

        Args:
            val: The value to compare against. Can be string, number, boolean, or None.

        Returns:
            BinaryOp: A binary operation representing the equality condition.

        Example:
            >>> FilterExpression.field('status').eq('Active').to_odata()
            "(status eq 'Active')"
        """
        return BinaryOp(self, "eq", literal(val))

    def ne(self, val: Any) -> "BinaryOp":
        """Create a not-equal comparison filter (field != value).

        Args:
            val: The value to compare against. Can be string, number, boolean, or None.

        Returns:
            BinaryOp: A binary operation representing the not-equal condition.

        Example:
            >>> FilterExpression.field('status').ne('Inactive').to_odata()
            "(status ne 'Inactive')"
        """
        return BinaryOp(self, "ne", literal(val))

    def gt(self, val: Any) -> "BinaryOp":
        """Create a greater-than comparison filter (field > value).

        Args:
            val: The value to compare against. Typically a number or date string.

        Returns:
            BinaryOp: A binary operation representing the greater-than condition.

        Example:
            >>> FilterExpression.field('salary').gt(50000).to_odata()
            "(salary gt 50000)"
        """
        return BinaryOp(self, "gt", literal(val))

    def ge(self, val: Any) -> "BinaryOp":
        """Create a greater-than-or-equal comparison filter (field >= value).

        Args:
            val: The value to compare against. Typically a number or date string.

        Returns:
            BinaryOp: A binary operation representing the greater-than-or-equal condition.

        Example:
            >>> FilterExpression.field('hireDate').ge('2020-01-01').to_odata()
            "(hireDate ge '2020-01-01')"
        """
        return BinaryOp(self, "ge", literal(val))

    def lt(self, val: Any) -> "BinaryOp":
        """Create a less-than comparison filter (field < value).

        Args:
            val: The value to compare against. Typically a number or date string.

        Returns:
            BinaryOp: A binary operation representing the less-than condition.

        Example:
            >>> FilterExpression.field('salary').lt(100000).to_odata()
            "(salary lt 100000)"
        """
        return BinaryOp(self, "lt", literal(val))

    def le(self, val: Any) -> "BinaryOp":
        """Create a less-than-or-equal comparison filter (field <= value).

        Args:
            val: The value to compare against. Typically a number or date string.

        Returns:
            BinaryOp: A binary operation representing the less-than-or-equal condition.

        Example:
            >>> FilterExpression.field('retirementDate').le('2025-12-31').to_odata()
            "(retirementDate le '2025-12-31')"
        """
        return BinaryOp(self, "le", literal(val))

    # string functions
    def contains(self, val: Any) -> "Func":
        """Create a substring contains filter for string fields.

        Args:
            val: The substring to search for within the field value.

        Returns:
            Func: A function call representing the contains operation.

        Example:
            >>> FilterExpression.field('lastName').contains('Smith').to_odata()
            "contains(lastName, 'Smith')"
        """
        return Func("contains", [self, literal(val)])

    def startswith(self, val: Any) -> "Func":
        """Create a string starts-with filter.

        Args:
            val: The prefix to search for at the start of the field value.

        Returns:
            Func: A function call representing the startswith operation.

        Example:
            >>> FilterExpression.field('firstName').startswith('John').to_odata()
            "startswith(firstName, 'John')"
        """
        return Func("startswith", [self, literal(val)])

    def endswith(self, val: Any) -> "Func":
        """Create a string ends-with filter.

        Args:
            val: The suffix to search for at the end of the field value.

        Returns:
            Func: A function call representing the endswith operation.

        Example:
            >>> FilterExpression.field('email').endswith('@company.com').to_odata()
            "endswith(email, '@company.com')"
        """
        return Func("endswith", [self, literal(val)])

    # emulate IN as disjunction
    def isin(self, values: list[Any]) -> "Expr":
        """Create an IN filter for multiple values (field IN (val1, val2, ...)).

        Since OData v4 doesn't have a native IN operator, this is implemented as
        a series of OR conditions joined together.

        Args:
            values: A list of values to check against. If empty, returns false.

        Returns:
            Expr: An expression representing the IN operation. For empty lists,
                  returns an always-false condition (1 eq 0).

        Example:
            >>> statuses = ['Active', 'OnLeave', 'Pending']
            >>> FilterExpression.field('status').isin(statuses).to_odata()
            "((status eq 'Active') or ((status eq 'OnLeave') or (status eq 'Pending')))"
        """
        if not values:
            # empty IN -> false; represent as (1 eq 0)
            return BinaryOp(Literal(1), "eq", Literal(0))
        expr: Expr = BinaryOp(self, "eq", literal(values[0]))
        for v in values[1:]:
            clause = BinaryOp(self, "eq", literal(v))
            expr = BinaryOp(expr, "or", clause)
        return expr

    def to_odata(self) -> str:
        """Convert this field reference to an OData path string.

        Converts dot notation to forward slash notation for OData v4 compliance.

        Returns:
            str: The OData-compliant field path.

        Example:
            >>> Field('worker.person.firstName').to_odata()
            'worker/person/firstName'
        """
        # Convert dot notation to forward slash for OData v4 compliance
        # Input: "workers.workAssignments.reportsTo.positionID"
        # Output: "workers/workAssignments/reportsTo/positionID"
        return self.path.replace(".", "/")

`contains(val)`

Create a substring contains filter for string fields.

Parameters:

Name	Type	Description	Default
`val`	`Any`	The substring to search for within the field value.	required

Returns:

Name	Type	Description
`Func`	`Func`	A function call representing the contains operation.

Example

FilterExpression.field('lastName').contains('Smith').to_odata() "contains(lastName, 'Smith')"

Source code in src/adpapi/odata_filters.py

def contains(self, val: Any) -> "Func":
    """Create a substring contains filter for string fields.

    Args:
        val: The substring to search for within the field value.

    Returns:
        Func: A function call representing the contains operation.

    Example:
        >>> FilterExpression.field('lastName').contains('Smith').to_odata()
        "contains(lastName, 'Smith')"
    """
    return Func("contains", [self, literal(val)])

`endswith(val)`

Create a string ends-with filter.

Parameters:

Name	Type	Description	Default
`val`	`Any`	The suffix to search for at the end of the field value.	required

Returns:

Name	Type	Description
`Func`	`Func`	A function call representing the endswith operation.

Example

FilterExpression.field('email').endswith('@company.com').to_odata() "endswith(email, '@company.com')"

Source code in src/adpapi/odata_filters.py

def endswith(self, val: Any) -> "Func":
    """Create a string ends-with filter.

    Args:
        val: The suffix to search for at the end of the field value.

    Returns:
        Func: A function call representing the endswith operation.

    Example:
        >>> FilterExpression.field('email').endswith('@company.com').to_odata()
        "endswith(email, '@company.com')"
    """
    return Func("endswith", [self, literal(val)])

`eq(val)`

Create an equality comparison filter (field = value).

Parameters:

Name	Type	Description	Default
`val`	`Any`	The value to compare against. Can be string, number, boolean, or None.	required

Returns:

Name	Type	Description
`BinaryOp`	`BinaryOp`	A binary operation representing the equality condition.

Example

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

Source code in src/adpapi/odata_filters.py

def eq(self, val: Any) -> "BinaryOp":
    """Create an equality comparison filter (field = value).

    Args:
        val: The value to compare against. Can be string, number, boolean, or None.

    Returns:
        BinaryOp: A binary operation representing the equality condition.

    Example:
        >>> FilterExpression.field('status').eq('Active').to_odata()
        "(status eq 'Active')"
    """
    return BinaryOp(self, "eq", literal(val))

`ge(val)`

Create a greater-than-or-equal comparison filter (field >= value).

Parameters:

Name	Type	Description	Default
`val`	`Any`	The value to compare against. Typically a number or date string.	required

Returns:

Name	Type	Description
`BinaryOp`	`BinaryOp`	A binary operation representing the greater-than-or-equal condition.

Example

FilterExpression.field('hireDate').ge('2020-01-01').to_odata() "(hireDate ge '2020-01-01')"

Source code in src/adpapi/odata_filters.py

def ge(self, val: Any) -> "BinaryOp":
    """Create a greater-than-or-equal comparison filter (field >= value).

    Args:
        val: The value to compare against. Typically a number or date string.

    Returns:
        BinaryOp: A binary operation representing the greater-than-or-equal condition.

    Example:
        >>> FilterExpression.field('hireDate').ge('2020-01-01').to_odata()
        "(hireDate ge '2020-01-01')"
    """
    return BinaryOp(self, "ge", literal(val))

`gt(val)`

Create a greater-than comparison filter (field > value).

Parameters:

Name	Type	Description	Default
`val`	`Any`	The value to compare against. Typically a number or date string.	required

Returns:

Name	Type	Description
`BinaryOp`	`BinaryOp`	A binary operation representing the greater-than condition.

Example

FilterExpression.field('salary').gt(50000).to_odata() "(salary gt 50000)"

Source code in src/adpapi/odata_filters.py

def gt(self, val: Any) -> "BinaryOp":
    """Create a greater-than comparison filter (field > value).

    Args:
        val: The value to compare against. Typically a number or date string.

    Returns:
        BinaryOp: A binary operation representing the greater-than condition.

    Example:
        >>> FilterExpression.field('salary').gt(50000).to_odata()
        "(salary gt 50000)"
    """
    return BinaryOp(self, "gt", literal(val))

`isin(values)`

Create an IN filter for multiple values (field IN (val1, val2, ...)).

Since OData v4 doesn't have a native IN operator, this is implemented as a series of OR conditions joined together.

Parameters:

Name	Type	Description	Default
`values`	`list[Any]`	A list of values to check against. If empty, returns false.	required

Returns:

Name	Type	Description
`Expr`	`Expr`	An expression representing the IN operation. For empty lists, returns an always-false condition (1 eq 0).

Example

statuses = ['Active', 'OnLeave', 'Pending'] FilterExpression.field('status').isin(statuses).to_odata() "((status eq 'Active') or ((status eq 'OnLeave') or (status eq 'Pending')))"

Source code in src/adpapi/odata_filters.py

def isin(self, values: list[Any]) -> "Expr":
    """Create an IN filter for multiple values (field IN (val1, val2, ...)).

    Since OData v4 doesn't have a native IN operator, this is implemented as
    a series of OR conditions joined together.

    Args:
        values: A list of values to check against. If empty, returns false.

    Returns:
        Expr: An expression representing the IN operation. For empty lists,
              returns an always-false condition (1 eq 0).

    Example:
        >>> statuses = ['Active', 'OnLeave', 'Pending']
        >>> FilterExpression.field('status').isin(statuses).to_odata()
        "((status eq 'Active') or ((status eq 'OnLeave') or (status eq 'Pending')))"
    """
    if not values:
        # empty IN -> false; represent as (1 eq 0)
        return BinaryOp(Literal(1), "eq", Literal(0))
    expr: Expr = BinaryOp(self, "eq", literal(values[0]))
    for v in values[1:]:
        clause = BinaryOp(self, "eq", literal(v))
        expr = BinaryOp(expr, "or", clause)
    return expr

`le(val)`

Create a less-than-or-equal comparison filter (field <= value).

Parameters:

Name	Type	Description	Default
`val`	`Any`	The value to compare against. Typically a number or date string.	required

Returns:

Name	Type	Description
`BinaryOp`	`BinaryOp`	A binary operation representing the less-than-or-equal condition.

Example

FilterExpression.field('retirementDate').le('2025-12-31').to_odata() "(retirementDate le '2025-12-31')"

Source code in src/adpapi/odata_filters.py

def le(self, val: Any) -> "BinaryOp":
    """Create a less-than-or-equal comparison filter (field <= value).

    Args:
        val: The value to compare against. Typically a number or date string.

    Returns:
        BinaryOp: A binary operation representing the less-than-or-equal condition.

    Example:
        >>> FilterExpression.field('retirementDate').le('2025-12-31').to_odata()
        "(retirementDate le '2025-12-31')"
    """
    return BinaryOp(self, "le", literal(val))

`lt(val)`

Create a less-than comparison filter (field < value).

Parameters:

Name	Type	Description	Default
`val`	`Any`	The value to compare against. Typically a number or date string.	required

Returns:

Name	Type	Description
`BinaryOp`	`BinaryOp`	A binary operation representing the less-than condition.

Example

FilterExpression.field('salary').lt(100000).to_odata() "(salary lt 100000)"

Source code in src/adpapi/odata_filters.py

def lt(self, val: Any) -> "BinaryOp":
    """Create a less-than comparison filter (field < value).

    Args:
        val: The value to compare against. Typically a number or date string.

    Returns:
        BinaryOp: A binary operation representing the less-than condition.

    Example:
        >>> FilterExpression.field('salary').lt(100000).to_odata()
        "(salary lt 100000)"
    """
    return BinaryOp(self, "lt", literal(val))

`ne(val)`

Create a not-equal comparison filter (field != value).

Parameters:

Name	Type	Description	Default
`val`	`Any`	The value to compare against. Can be string, number, boolean, or None.	required

Returns:

Name	Type	Description
`BinaryOp`	`BinaryOp`	A binary operation representing the not-equal condition.

Example

FilterExpression.field('status').ne('Inactive').to_odata() "(status ne 'Inactive')"

Source code in src/adpapi/odata_filters.py

def ne(self, val: Any) -> "BinaryOp":
    """Create a not-equal comparison filter (field != value).

    Args:
        val: The value to compare against. Can be string, number, boolean, or None.

    Returns:
        BinaryOp: A binary operation representing the not-equal condition.

    Example:
        >>> FilterExpression.field('status').ne('Inactive').to_odata()
        "(status ne 'Inactive')"
    """
    return BinaryOp(self, "ne", literal(val))

`startswith(val)`

Create a string starts-with filter.

Parameters:

Name	Type	Description	Default
`val`	`Any`	The prefix to search for at the start of the field value.	required

Returns:

Name	Type	Description
`Func`	`Func`	A function call representing the startswith operation.

Example

FilterExpression.field('firstName').startswith('John').to_odata() "startswith(firstName, 'John')"

Source code in src/adpapi/odata_filters.py

def startswith(self, val: Any) -> "Func":
    """Create a string starts-with filter.

    Args:
        val: The prefix to search for at the start of the field value.

    Returns:
        Func: A function call representing the startswith operation.

    Example:
        >>> FilterExpression.field('firstName').startswith('John').to_odata()
        "startswith(firstName, 'John')"
    """
    return Func("startswith", [self, literal(val)])

`to_odata()`

Convert this field reference to an OData path string.

Converts dot notation to forward slash notation for OData v4 compliance.

Returns:

Name	Type	Description
`str`	`str`	The OData-compliant field path.

Example

Field('worker.person.firstName').to_odata() 'worker/person/firstName'

Source code in src/adpapi/odata_filters.py

def to_odata(self) -> str:
    """Convert this field reference to an OData path string.

    Converts dot notation to forward slash notation for OData v4 compliance.

    Returns:
        str: The OData-compliant field path.

    Example:
        >>> Field('worker.person.firstName').to_odata()
        'worker/person/firstName'
    """
    # Convert dot notation to forward slash for OData v4 compliance
    # Input: "workers.workAssignments.reportsTo.positionID"
    # Output: "workers/workAssignments/reportsTo/positionID"
    return self.path.replace(".", "/")

Logging

Quick basic logging configuration with a rotating file handler and stream handling.

Source code in src/adpapi/logger.py

def configure_logging():
    logger = logging.getLogger("adpapi")
    logger.setLevel(logging.DEBUG)

    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")

    # File handler
    file_handler = logging.FileHandler("app.log", mode="w")
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

    # Console handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

API Reference

Credential Management

from_env() staticmethod

Client

Configuring Retry Behavior

Basic usage

With column selection

With filtering

Limited fetch for testing

Single resource fetch

Batch fetch (sequential)

Parallel batch fetch (5-10x faster)

With column selection and ID injection

Each response now includes: {"associateOID": "G3349PRDL000001", ...}

Multiple path parameters

Filters

contains(val)

endswith(val)

eq(val)

ge(val)

gt(val)

isin(values)

le(val)

lt(val)

ne(val)

startswith(val)

to_odata()

Logging

`from_env()` `staticmethod`

`contains(val)`

`endswith(val)`

`eq(val)`

`ge(val)`

`gt(val)`

`isin(values)`

`le(val)`

`lt(val)`

`ne(val)`

`startswith(val)`

`to_odata()`