Skip to content

rs_workflows/utils/artifact_verbose.md

<< Back to index

Report Manager to create artifact table

ReportManager

Manages execution reports for a multi-step flow and publishes them as Prefect artifacts.

Attributes

number_steps : int Total number of steps expected in the flow. report : list[dict] Accumulated list of step results.

Source code in docs/rs-client-libraries/rs_workflows/utils/artifact_verbose.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class ReportManager:
    """
    Manages execution reports for a multi-step flow and publishes them
    as Prefect artifacts.

    Attributes
    ----------
    number_steps : int
        Total number of steps expected in the flow.
    report : list[dict]
        Accumulated list of step results.
    """

    def __init__(self):
        self.report = []

    def success_step(self, step: int, description: str) -> None:
        """
        Register a successful step in the report.
        """
        item = {
            "step": step,
            "description": description,
            "status": "OK",
        }
        self.report.append(item)

    def failed_step(self, step: int, description: str) -> None:
        """
        Register a failed step in the report.
        """
        item = {
            "step": step,
            "description": description,
            "status": "NOK",
        }
        self.report.append(item)

    async def push_report(self, key_value: str, description_value: str):
        """
        Publish the accumulated report as a Prefect table artifact.

        Notes
        -----
        - Prefect artifact keys must contain only lowercase letters,
          numbers, and dashes.
        - A timestamp is appended to the description for traceability.

        Returns
        -------
        ArtifactResponse
            The Prefect artifact object returned by create_table_artifact().
        """
        now = datetime.now()
        timestamp = now.strftime("%A %d %B %Y, %H:%M:%S")

        await acreate_table_artifact(
            key=key_value.lower(),
            table=self.report,
            description=f"{description_value} - {timestamp}",
        )

failed_step(step, description)

Register a failed step in the report.

Source code in docs/rs-client-libraries/rs_workflows/utils/artifact_verbose.py 
49
50
51
52
53
54
55
56
57
58
def failed_step(self, step: int, description: str) -> None:
    """
    Register a failed step in the report.
    """
    item = {
        "step": step,
        "description": description,
        "status": "NOK",
    }
    self.report.append(item)

push_report(key_value, description_value) async

Publish the accumulated report as a Prefect table artifact.

Notes
  • Prefect artifact keys must contain only lowercase letters, numbers, and dashes.
  • A timestamp is appended to the description for traceability.
Returns

ArtifactResponse The Prefect artifact object returned by create_table_artifact().

Source code in docs/rs-client-libraries/rs_workflows/utils/artifact_verbose.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
async def push_report(self, key_value: str, description_value: str):
    """
    Publish the accumulated report as a Prefect table artifact.

    Notes
    -----
    - Prefect artifact keys must contain only lowercase letters,
      numbers, and dashes.
    - A timestamp is appended to the description for traceability.

    Returns
    -------
    ArtifactResponse
        The Prefect artifact object returned by create_table_artifact().
    """
    now = datetime.now()
    timestamp = now.strftime("%A %d %B %Y, %H:%M:%S")

    await acreate_table_artifact(
        key=key_value.lower(),
        table=self.report,
        description=f"{description_value} - {timestamp}",
    )

success_step(step, description)

Register a successful step in the report.

Source code in docs/rs-client-libraries/rs_workflows/utils/artifact_verbose.py 
38
39
40
41
42
43
44
45
46
47
def success_step(self, step: int, description: str) -> None:
    """
    Register a successful step in the report.
    """
    item = {
        "step": step,
        "description": description,
        "status": "OK",
    }
    self.report.append(item)