triad

import pandas as pd from triad.collections import FunctionWrapper, Schema # Example of FunctionWrapper for input/output validation @FunctionWrapper( params_re="(pa:pd.DataFrame,pb:pd.DataFrame)(.*)", return_re="(res:pd.DataFrame)" ) def process_dataframes(pa: pd.DataFrame, pb: pd.DataFrame, *args) -> pd.DataFrame: """A dummy function to demonstrate FunctionWrapper validation.""" print(f"[FunctionWrapper] Received dataframes with columns: {pa.columns.tolist()}, {pb.columns.tolist()}") print(f"[FunctionWrapper] Received other args: {args}") return pd.DataFrame({"sum_rows": [len(pa) + len(pb)]}) # Create some dummy dataframes df1 = pd.DataFrame({"col1": [1, 2], "col2": ["a", "b"]}) df2 = pd.DataFrame({"colA": [3, 4, 5], "colB": ["c", "d", "e"]}) # Call the wrapped function result = process_dataframes(df1, df2, "extra_param", 42) print(f"\nFunction result:\n{result}") # Example of Schema usage s = Schema("id:int,name:str,value:double") print(f"\nCreated Schema: {s.as_str}") # Convert a dictionary to a Pandas DataFrame conforming to the schema data = {"id": [101, 102], "name": ["Alice", "Bob"], "value": [1.1, 2.2]} df_from_schema = s.as_pandas_df(data) print(f"\nDataFrame from Schema:\n{df_from_schema}")