Artifacts

import upnext

@worker.task
async def generate_report(report_type: str) -> dict:
    data = {"type": report_type, "rows": 42}

    await upnext.create_artifact(
        name="report.json",
        data=data,
        artifact_type=upnext.ArtifactType.JSON,
    )

    return data

await upnext.create_artifact(
    name="output.json",
    data={"key": "value"},
)

upnext.create_artifact_sync(
    name="output.json",
    data={"key": "value"},
)

await upnext.create_artifact(
    name="results.json",
    data={"users": 150, "errors": 3},
)

await upnext.create_artifact(
    name="log.txt",
    data="Processing completed successfully.\n3 warnings.",
    artifact_type=upnext.ArtifactType.TEXT,
)

import base64

with open("chart.png", "rb") as f:
    encoded = base64.b64encode(f.read()).decode("ascii")

await upnext.create_artifact(
    name="chart.png",
    data=encoded,
    artifact_type=upnext.ArtifactType.PNG,
)

UPNEXT_ARTIFACT_STORAGE_BACKEND=s3