Extractor API Reference

This function serves as the main entry point for extracting element data from Relatics and transforming it into a set of normalized pandas DataFrames. It orchestrates the full extraction workflow across a specified list of elements within a workspace, including data retrieval, schema validation, table normalization, and application of transformations of the data into relational tables.

The function supports both sequential and parallel execution, making it suitable for small ad hoc extracts as well as larger data pipelines. Optionally, selected to-one relations can be inlined into the resulting element tables, simplifying downstream analysis and reporting. The output is a dictionary of transformed DataFrames that can be used directly for further processing, analytics, or data integration tasks.

Overview

relatics_toolkit.extract_element_tables(client, workspace_id, element_ids, operation, parallel=False, max_workers=None, inline_relations=None)

Extracts and transforms Relatics elements into normalized tables.

For the specified workspace and element IDs, retrieves the corresponding Relatics XML payloads, validates the extracted schema, normalizes the resulting tables, and applies business transformations.

Processing can be executed sequentially or in parallel.

Parameters:

Name	Type	Description	Default
client	RelaticsClient	Configured Relatics API client.	required
workspace_id	str	Workspace ID containing the elements that should be extracted.	required
element_ids	list[str]	Element IDs that should be extracted from the workspace.	required
operation	str	Relatics operation name used to retrieve the element data.	required
parallel	bool	Whether element extraction should be executed in parallel.	False
max_workers	int | None	Maximum number of worker threads used when parallel is True. If None, the ThreadPoolExecutor default is used.	None
inline_relations	list[str] | None	Relation names of relations to R2 elements whose values should be materialized directly in the resulting element tables (must be to-one relations).	None

Returns:

Type	Description
dict[str, DataFrame]	Dictionary mapping table names to transformed pandas DataFrames.

Raises:

Type	Description
Exception	Any exception raised during retrieval, validation, normalization, or transformation of element data.

Source code in src/relatics_toolkit/extraction_service.py

def extract_element_tables(
    client: RelaticsClient,
    workspace_id: str,
    element_ids: list[str],
    operation: str,
    parallel: bool = False,
    max_workers: int | None = None,
    inline_relations: list[str] | None = None,
) -> dict[str, pd.DataFrame]:
    """
    Extracts and transforms Relatics elements into normalized tables.

    For the specified workspace and element IDs, retrieves the
    corresponding Relatics XML payloads, validates the extracted schema,
    normalizes the resulting tables, and applies business transformations.

    Processing can be executed sequentially or in parallel.

    Args:
        client: Configured Relatics API client.
        workspace_id: Workspace ID containing the elements that should
            be extracted.
        element_ids: Element IDs that should be extracted from the
            workspace.
        operation: Relatics operation name used to retrieve the
            element data.
        parallel: Whether element extraction should be executed in
            parallel.
        max_workers: Maximum number of worker threads used when
            parallel is True. If None, the ThreadPoolExecutor
            default is used.
        inline_relations: Relation names of relations to R2 elements
            whose values should be materialized directly in the
            resulting element tables (must be to-one relations).

    Returns:
        Dictionary mapping table names to transformed pandas DataFrames.

    Raises:
        Exception: Any exception raised during retrieval, validation,
            normalization, or transformation of element data.
    """
    logger.info(
        "Starting extraction for %s elements in workspace %s (parallel=%s)",
        len(element_ids),
        workspace_id,
        parallel,
    )

    tables: dict[str, pd.DataFrame] = {}

    if parallel:
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_map = {
                executor.submit(
                    _process_element,
                    element_id=element_id,
                    client=client,
                    workspace_id=workspace_id,
                    operation=operation,
                    inline_relations=inline_relations,
                ): element_id
                for element_id in element_ids
            }

            for future in as_completed(future_map):
                element_id = future_map[future]

                try:
                    _add_tables(tables, future.result())
                except Exception:
                    logger.exception(
                        "Failed processing workspace_id=%s element_id=%s",
                        workspace_id,
                        element_id,
                    )
                    raise
    else:
        for element_id in element_ids:
            try:
                _add_tables(
                    tables,
                    _process_element(
                        element_id=element_id,
                        client=client,
                        workspace_id=workspace_id,
                        operation=operation,
                        inline_relations=inline_relations,
                    ),
                )
            except Exception:
                logger.exception(
                    "Failed processing workspace_id=%s element_id=%s",
                    workspace_id,
                    element_id,
                )
                raise

    logger.info(
        "Extraction completed successfully. Generated %s tables.",
        len(tables),
    )

    return tables