Function unpack_col_dict

python/pathway/stdlib/utils/col.py:97–188

Unpacks columns from a json object Input: - column: Column expression of column containing some pw.Json with an object - schema: Schema for columns to extract Output: - Table with columns given by the schema Example: >>> import pathway as pw >>> t = pw.debug.table_

(
    column: pw.ColumnReference,
    schema: type[pw.Schema],
)

@check_arg_types
@trace_user_frame
def unpack_col_dict(
    column: pw.ColumnReference,
    schema: type[pw.Schema],
) -> pw.Table:
    """Unpacks columns from a json object

    Input:
    - column: Column expression of column containing some pw.Json with an object
    - schema: Schema for columns to extract

    Output:
    - Table with columns given by the schema

    Example:
    >>> import pathway as pw
    >>> t = pw.debug.table_from_rows(
    ...     schema=pw.schema_from_types(data=pw.Json),
    ...     rows=[
    ...         ({"field_a": 13, "field_b": "foo", "field_c": False},),
    ...         ({"field_a": 17, "field_c": True, "field_d": 3.4},)
    ...     ]
    ... )
    >>> class DataSchema(pw.Schema):
    ...     field_a: int
    ...     field_b: str | None
    ...     field_c: bool
    ...     field_d: float | None
    >>> t2 = pw.utils.col.unpack_col_dict(t.data, schema=DataSchema)
    >>> pw.debug.compute_and_print(t2, include_id=False)
    field_a | field_b | field_c | field_d
    13      | foo     | False   |
    17      |         | True    | 3.4
    """
    typehints = schema._dtypes()

    def _convert_from_json(name: str, col: pw.ColumnExpression):
        _type = dt.unoptionalize(typehints[name])
        is_optional = isinstance(typehints[name], dt.Optional)
        result: pw.ColumnExpression

        def _optional(
            col: pw.ColumnExpression,
            op: Callable[[pw.ColumnExpression], pw.ColumnExpression],
        ) -> pw.ColumnExpression:
            if is_optional:
                return pw.if_else(col == pw.Json.NULL, None, op(col))
            else:
                return op(col)

        match _type:
            case dt.JSON:
                result = col
            case dt.BOOL:
                result = col.as_bool()
            case dt.FLOAT:
                result = col.as_float()
            case dt.INT:
                result = col.as_int()