prefect.artifacts

Interface for creating and reading artifacts.

create_link_artifact async

Create a link artifact.

Parameters:

Name	Type	Description	Default
link	str	The link to create.	required
link_text	Optional[str]	The link text.	None
key	Optional[str]	A user-provided string identifier. Required for the artifact to show in the Artifacts page in the UI. The key must only contain lowercase letters, numbers, and dashes.	None
description	Optional[str]	A user-specified description of the artifact.	None

Returns:

Type	Description
UUID	The table artifact ID.

Source code in prefect/artifacts.py

@sync_compatible
async def create_link_artifact(
    link: str,
    link_text: Optional[str] = None,
    key: Optional[str] = None,
    description: Optional[str] = None,
) -> UUID:
    """
    Create a link artifact.

    Arguments:
        link: The link to create.
        link_text: The link text.
        key: A user-provided string identifier.
          Required for the artifact to show in the Artifacts page in the UI.
          The key must only contain lowercase letters, numbers, and dashes.
        description: A user-specified description of the artifact.


    Returns:
        The table artifact ID.
    """
    formatted_link = f"[{link_text}]({link})" if link_text else f"[{link}]({link})"
    artifact = await _create_artifact(
        key=key,
        type="markdown",
        description=description,
        data=formatted_link,
    )

    return artifact.id

create_markdown_artifact async

Create a markdown artifact.

Parameters:

Name	Type	Description	Default
markdown	str	The markdown to create.	required
key	Optional[str]	A user-provided string identifier. Required for the artifact to show in the Artifacts page in the UI. The key must only contain lowercase letters, numbers, and dashes.	None
description	Optional[str]	A user-specified description of the artifact.	None

Returns:

Type	Description
UUID	The table artifact ID.

Source code in prefect/artifacts.py

@sync_compatible
async def create_markdown_artifact(
    markdown: str,
    key: Optional[str] = None,
    description: Optional[str] = None,
) -> UUID:
    """
    Create a markdown artifact.

    Arguments:
        markdown: The markdown to create.
        key: A user-provided string identifier.
          Required for the artifact to show in the Artifacts page in the UI.
          The key must only contain lowercase letters, numbers, and dashes.
        description: A user-specified description of the artifact.

    Returns:
        The table artifact ID.
    """
    artifact = await _create_artifact(
        key=key,
        type="markdown",
        description=description,
        data=markdown,
    )

    return artifact.id

create_table_artifact async

Create a table artifact.

Parameters:

Name	Type	Description	Default
table	Union[Dict[str, List[Any]], List[Dict[str, Any]], List[List[Any]]]	The table to create.	required
key	Optional[str]	A user-provided string identifier. Required for the artifact to show in the Artifacts page in the UI. The key must only contain lowercase letters, numbers, and dashes.	None
description	Optional[str]	A user-specified description of the artifact.	None

Returns:

Type	Description
UUID	The table artifact ID.

Source code in prefect/artifacts.py

@sync_compatible
async def create_table_artifact(
    table: Union[Dict[str, List[Any]], List[Dict[str, Any]], List[List[Any]]],
    key: Optional[str] = None,
    description: Optional[str] = None,
) -> UUID:
    """
    Create a table artifact.

    Arguments:
        table: The table to create.
        key: A user-provided string identifier.
          Required for the artifact to show in the Artifacts page in the UI.
          The key must only contain lowercase letters, numbers, and dashes.
        description: A user-specified description of the artifact.

    Returns:
        The table artifact ID.
    """

    def _sanitize_nan_values(item):
        """
        Sanitize NaN values in a given item. The item can be a dict, list or float.
        """

        if isinstance(item, list):
            return [_sanitize_nan_values(sub_item) for sub_item in item]

        elif isinstance(item, dict):
            return {k: _sanitize_nan_values(v) for k, v in item.items()}

        elif isinstance(item, float) and math.isnan(item):
            return None

        else:
            return item

    sanitized_table = _sanitize_nan_values(table)

    if isinstance(table, dict) and all(isinstance(v, list) for v in table.values()):
        pass
    elif isinstance(table, list) and all(isinstance(v, (list, dict)) for v in table):
        pass
    else:
        raise TypeError(INVALID_TABLE_TYPE_ERROR)

    formatted_table = json.dumps(sanitized_table)

    artifact = await _create_artifact(
        key=key,
        type="table",
        description=description,
        data=formatted_table,
    )

    return artifact.id