Function assert_trace_serializable

tests/pipeline/test_pipeline_trace.py:891–909

(trace: PipelineTrace)

def assert_trace_serializable(trace: PipelineTrace) -> None:
    str(trace)
    trace.asstr(0)
    trace.asstr(1)
    trace_dict = deepcopy(trace.asdict())
    # check if we do not pop
    assert trace_dict == trace.asdict()
    with io.BytesIO() as b:
        json.typed_dump(trace, b, pretty=True)
        b.getvalue()
    json.dumps(trace)

    # load trace to duckdb
    from dlt.destinations import duckdb

    trace_pipeline = dlt.pipeline("trace", destination=duckdb(":pipeline:")).drop()
    trace_pipeline.run([trace], table_name="trace_data")

    # print(trace_pipeline.default_schema.to_pretty_yaml())