Function load

cartography/client/core/tx.py:784–850

Load node data to the graph with automatic indexing. This is the main entry point for intel modules to write node data to the graph. It automatically ensures that required indexes exist before performing the load operation, optimizing performance and maintaining data integrity.

(
    neo4j_session: neo4j.Session,
    node_schema: CartographyNodeSchema,
    dict_list: List[Dict[str, Any]],
    batch_size: int = 10000,
    **kwargs,
)

def load(
    neo4j_session: neo4j.Session,
    node_schema: CartographyNodeSchema,
    dict_list: List[Dict[str, Any]],
    batch_size: int = 10000,
    **kwargs,
) -> None:
    """
    Load node data to the graph with automatic indexing.

    This is the main entry point for intel modules to write node data to the graph.
    It automatically ensures that required indexes exist before performing the load
    operation, optimizing performance and maintaining data integrity.

    Args:
        neo4j_session (neo4j.Session): The Neo4j session for database operations.
        node_schema (CartographyNodeSchema): The node schema object that defines
            the structure of the data being loaded and generates the ingestion query.
        dict_list (List[Dict[str, Any]]): The data to load to the graph, represented
            as a list of dictionaries. Each dictionary represents one node to create
            or update.
        batch_size (int): The number of items to process per transaction. Defaults to 10000.
        **kwargs: Additional keyword arguments passed to the Neo4j query, such as
            timestamps, update tags, or other metadata.

    Examples:
        >>> node_schema = CartographyNodeSchema(
        ...     label='AWSUser',
        ...     properties={
        ...         'id': PropertyRef('UserId'),
        ...         'name': PropertyRef('UserName'),
        ...         'email': PropertyRef('Email')
        ...     }
        ... )
        >>> users_data = [
        ...     {'UserId': 'user1', 'UserName': 'Alice', 'Email': 'alice@example.com'},
        ...     {'UserId': 'user2', 'UserName': 'Bob', 'Email': 'bob@example.com'}
        ... ]
        >>> load(session, node_schema, users_data, lastupdated=current_time)

    Note:
        - If ``dict_list`` is empty, the function returns early to save processing time.
        - The function automatically creates necessary indexes before loading data.
        - The ingestion query is generated automatically from the node schema.
        - Data is processed in batches for optimal performance.
    """
    if batch_size <= 0:
        raise ValueError(f"batch_size must be greater than 0, got {batch_size}")
    if len(dict_list) == 0:
        # If there is no data to load, save some time.
        return
    ensure_indexes(neo4j_session, node_schema)
    ingestion_query = build_ingestion_query(node_schema)
    load_graph_data(
        neo4j_session, ingestion_query, dict_list, batch_size=batch_size, **kwargs
    )

    # Apply conditional labels if any are defined