`workday`

Workday API connectors for Snowpark Stored Procedures.

Supports three integration points:

RAAS REST — Report-as-a-Service endpoint, single or paged.
SOAP — EIB / Core Connector operations (single call, paged, and bulk parallel).
WQL — Workday Query Language with LIMIT/OFFSET pagination and OAuth2 token refresh.

All lxml and requests imports are lazy — the module is importable without these dependencies installed.

Note

Requires the workday optional dependency group::

pip install "pinky-connect[workday]"

`raas_query(report_url, *, secret_name, params=None)`

Fetch a Workday RAAS report and return all rows.

Parameters:

Name	Type	Description	Default
`report_url`	`str`	Fully-qualified RAAS endpoint URL (`/ccx/service/…`).	required
`secret_name`	`str`	Name of the Snowflake Secret holding Workday credentials.	required
`params`	`dict[str, Any] \| None`	Optional query parameters appended to the request URL.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	List of row dictionaries as parsed from the JSON response.

Raises:

Type	Description
`HTTPError`	On non-2xx responses.

Source code in src/pinky_connect/workday.py

def raas_query(
    report_url: str,
    *,
    secret_name: str,
    params: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """Fetch a Workday RAAS report and return all rows.

    Args:
        report_url: Fully-qualified RAAS endpoint URL (``/ccx/service/…``).
        secret_name: Name of the Snowflake Secret holding Workday credentials.
        params: Optional query parameters appended to the request URL.

    Returns:
        List of row dictionaries as parsed from the JSON response.

    Raises:
        requests.HTTPError: On non-2xx responses.
    """
    raise NotImplementedError("pending migration from _BACKUP/snowflake-kit-connect-bak")

`soap_call(endpoint, operation, payload, *, secret_name)`

Execute a single Workday SOAP operation.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	Workday SOAP service URL.	required
`operation`	`str`	SOAP operation name (used as the `SOAPAction` header).	required
`payload`	`str`	Raw XML request body.	required
`secret_name`	`str`	Name of the Snowflake Secret holding Workday credentials.	required

Returns:

Type	Description
`str`	Raw XML response string.

Raises:

Type	Description
`HTTPError`	On non-2xx responses.
`XMLSyntaxError`	If the response is not valid XML.

Source code in src/pinky_connect/workday.py

def soap_call(
    endpoint: str,
    operation: str,
    payload: str,
    *,
    secret_name: str,
) -> str:
    """Execute a single Workday SOAP operation.

    Args:
        endpoint: Workday SOAP service URL.
        operation: SOAP operation name (used as the ``SOAPAction`` header).
        payload: Raw XML request body.
        secret_name: Name of the Snowflake Secret holding Workday credentials.

    Returns:
        Raw XML response string.

    Raises:
        requests.HTTPError: On non-2xx responses.
        lxml.etree.XMLSyntaxError: If the response is not valid XML.
    """
    raise NotImplementedError("pending migration from _BACKUP/snowflake-kit-connect-bak")

`soap_paged(endpoint, operation, payload_template, *, secret_name, page_size=200)`

Fetch all pages from a paged Workday SOAP operation.

Yields one XML response string per page until Is_Last_Page is true.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	Workday SOAP service URL.	required
`operation`	`str`	SOAP operation name.	required
`payload_template`	`str`	XML template with `{page}` and `{page_size}` placeholders.	required
`secret_name`	`str`	Name of the Snowflake Secret holding Workday credentials.	required
`page_size`	`int`	Records per page (Workday max: 999).	`200`

Yields:

Type	Description
`Iterable[str]`	Raw XML response string for each page.

Source code in src/pinky_connect/workday.py

def soap_paged(
    endpoint: str,
    operation: str,
    payload_template: str,
    *,
    secret_name: str,
    page_size: int = 200,
) -> Iterable[str]:
    """Fetch all pages from a paged Workday SOAP operation.

    Yields one XML response string per page until ``Is_Last_Page`` is true.

    Args:
        endpoint: Workday SOAP service URL.
        operation: SOAP operation name.
        payload_template: XML template with ``{page}`` and ``{page_size}`` placeholders.
        secret_name: Name of the Snowflake Secret holding Workday credentials.
        page_size: Records per page (Workday max: 999).

    Yields:
        Raw XML response string for each page.
    """
    raise NotImplementedError("pending migration from _BACKUP/snowflake-kit-connect-bak")

`wql_query(wql, *, secret_name, limit=1000)`

Execute a WQL query and return all matching rows.

Handles LIMIT/OFFSET pagination and OAuth2 token refresh transparently.

Parameters:

Name	Type	Description	Default
`wql`	`str`	Workday Query Language statement.	required
`secret_name`	`str`	Name of the Snowflake Secret holding Workday credentials.	required
`limit`	`int`	Page size used internally for pagination.	`1000`

Returns:

Type	Description
`list[dict[str, Any]]`	List of row dictionaries as returned by the WQL endpoint.

Raises:

Type	Description
`HTTPError`	On non-2xx responses.

Source code in src/pinky_connect/workday.py

def wql_query(
    wql: str,
    *,
    secret_name: str,
    limit: int = 1000,
) -> list[dict[str, Any]]:
    """Execute a WQL query and return all matching rows.

    Handles LIMIT/OFFSET pagination and OAuth2 token refresh transparently.

    Args:
        wql: Workday Query Language statement.
        secret_name: Name of the Snowflake Secret holding Workday credentials.
        limit: Page size used internally for pagination.

    Returns:
        List of row dictionaries as returned by the WQL endpoint.

    Raises:
        requests.HTTPError: On non-2xx responses.
    """
    raise NotImplementedError("pending migration from _BACKUP/snowflake-kit-connect-bak")