ETL

Helper to make ETL pipelines easier.

Our plan

We want to try to build up some functionality to create etl pipelines for pandas operations. We want to define classic extract, transform and load functions which we can string together in a pipeline. We would also like to add some functionality to inspect the intermediate dataframes and some key properties.

Our background

We work as Data Science Consultants mainly with SME’s in manufacturing. We often have to build up etl pipelines from various sources to then use for analytics or ML workloads. The pipelines can get pretty complex with custom business logic, that why we want to have a way to quickly show all the steps we take.

Tools available from dialoghelper:

• &curr_dialog: Get the current dialog info.
• &msg_idx: Get absolute index of message in dialog.
• &add_html: Send HTML to the browser to be swapped into the DOM using hx-swap-oob.
• &find_msg_id: Get the current message id.
• &find_msgs: Find messages in current specific dialog that contain the given information.
  • (solveit can often get this id directly from its context, and will not need to use this if the required information is already available to it.)
• &read_msg: Get the message indexed in the current dialog.
  • To get the exact message use n=0 and relative=True together with msgid.
  • To get a relative message use n (relative position index).
  • To get the nth message use n with relative=False, e.g n=0 first message, n=-1 last message.
• &del_msg: Delete a message from the dialog.
• &add_msg: Add/update a message to the queue to show after code execution completes.
• &update_msg: Update an existing message.
• &url2note: Read URL as markdown, and add a note below current message with the result
• &msg_insert_line: Insert text at a specific location in a message.
• &msg_str_replace: Find and replace text in a message.
• &msg_strs_replace: Find and replace multiple strings in a message.
• &msg_replace_lines: Replace a range of lines in a message with new content.
  • Always first use read_msg( msgid=msgid, n=0, relative=True, nums=True) to view the content with line numbers.

Tools available from fastcore.tools:

• &rg: Run the rg command with the args in argstr (no need to backslash escape)
• &sed: Run the sed command with the args in argstr (e.g for reading a section of a file)
• &view: View directory or file contents with optional line range and numbers
• &create: Creates a new file with the given content at the specified path
• &insert: Insert new_str at specified line number
• &str_replace: Replace first occurrence of old_str with new_str in file
• &strs_replace: Replace for each str pair in old_strs,new_strs
• &replace_lines: Replace lines in file using start and end line-numbers

# Sample manufacturing data
data = {
    'order_id': [101, 102, 103, 104, 105],
    'product': ['Widget A', 'Widget B', 'Widget A', 'Widget C', 'Widget B'],
    'quantity': [50, 30, 75, 20, 45],
    'defects': [2, 1, 3, 0, 2],
    'production_time': [120, 95, 150, 80, 110]  # in minutes
}

df = pd.DataFrame(data)
df

	order_id	product	quantity	defects	production_time
0	101	Widget A	50	2	120
1	102	Widget B	30	1	95
2	103	Widget A	75	3	150
3	104	Widget C	20	0	80
4	105	Widget B	45	2	110

---

source

get_demo_data

def get_demo_data():

df[df["product"] != "Widget C"].groupby(["product"])[["quantity", "defects" ,"production_time"]].sum()

	quantity	defects	production_time
product
Widget A	125	5	270
Widget B	75	3	205

Solution space exploration

Say we have a pipeline like this. What are some approaches to implement our plan?

Show a simple example for each please

How would we deal with methods vs functions in each approach?

Can you write our small example transformation for each approach

Our preferred approach

For a the exploration and discussion with AI on the approaches see the notebooks in the nbs folder.

So we would write a decorator that adds a logging keyword to each function, then we can determine for each function in the pipeline what to log/how much to log. We want to keep it as simple as possible. The pipeline should just take a df and a list of functions as steps of the pipeline.

def pipeline(df, steps, vrbs_default=True):
    for func, func_kwargs in steps:
        vrbs =  func_kwargs.get("vrbs", vrbs_default)
        func_kwargs.update({"vrbs": vrbs})
        df = func(df, **func_kwargs)
    return df

def track(func):
    def wrapper(in_df, vrbs=False, *args, **kwargs):
        if vrbs:
            print(f"\n{'*'*10} Step: {func.__name__} {'*'*10}")
            print(f"\nInput DataFrame shape: {in_df.shape}")
            print(in_df)
        out_df = func(in_df, *args, **kwargs)
        if vrbs:
            print(f"\nOutput DataFrame shape: {out_df.shape}")
            print(out_df)
        return out_df
    return wrapper

@track
def filter_products(df):
    return df[df["product"] != "Widget C"]

@track
def aggregate_by_product(df):
    return df.groupby(["product"])[["quantity", "defects", "production_time"]].sum()

We include a way to handle pipeline overwrites of the vrbs argument for specific transformation steps. In this example, no information for aggregate_by_product is printed:

steps = [
    (filter_products, {}),
    (aggregate_by_product, {"vrbs": False}),
]

pipeline(df, steps)

********** Step: filter_products **********

Input DataFrame shape: (5, 5)
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110

Output DataFrame shape: (4, 5)
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110

	quantity	defects	production_time
product
Widget A	125	5	270
Widget B	75	3	205

We want to add basic profiling of time and memory.

start_time = datetime.now()
sleep(3.14)
end_time = datetime.now()
total_time = end_time - start_time
total_time

datetime.timedelta(seconds=3, microseconds=140759)

start_time

datetime.datetime(2026, 9, 14, 12, 30, 41, 547669)

def track(func):
    def wrapper(in_df, vrbs=False, *args, **kwargs):
        start_time = datetime.now()
        if vrbs:
            
            print(f"\n{'*'*10} Step: {func.__name__} {'*'*10}")
            print(f"\nInput DataFrame shape: {in_df.shape}")
            print(f"Start time: {start_time}")
            print(in_df)
        out_df = func(in_df, *args, **kwargs)
        

        end_time = datetime.now()
        total_time = end_time - start_time
        if vrbs:
            print(f"\nOutput DataFrame shape: {out_df.shape}")
            print(f"End time: {end_time}")
            print(f"Total time: {total_time}")
            print(out_df)
        return out_df
    return wrapper

@track
def filter_products(df):
    return df[df["product"] != "Widget C"]

@track
def aggregate_by_product(df):
    return df.groupby(["product"])[["quantity", "defects", "production_time"]].sum()

steps = [
    (filter_products, {}),
    (aggregate_by_product, {}),
]

_df = pipeline(df, steps)

********** Step: filter_products **********

Input DataFrame shape: (5, 5)
Start time: 2026-09-14 12:30:44.757964
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110

Output DataFrame shape: (4, 5)
End time: 2026-09-14 12:30:44.759521
Total time: 0:00:00.001557
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110

********** Step: aggregate_by_product **********

Input DataFrame shape: (4, 5)
Start time: 2026-09-14 12:30:44.760193
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110

Output DataFrame shape: (2, 3)
End time: 2026-09-14 12:30:44.763016
Total time: 0:00:00.002823
          quantity  defects  production_time
product                                     
Widget A       125        5              270
Widget B        75        3              205

first reflection and next steps

• right now we are still printing the whole dataframe. We want to change that to just show the first 3, 5 random and the last 3 rows, maybe with some visual clue to show that the rows are truncated.

• we would also like show the dataframes to the right of the aggregate infos (like shape, etc.)

• include information like df.info()

• include information like df.describe()

• include information like diff columns (num cols changed, cols just in before, cols just in after)

• include information like diff rows (maybe just the number)

• source code of functions

• description of step, based on docstring

• nicer representation maybe in html or display or something like that

• new wrapper to include assertions and/or sanity checks (propertiy based, for example same number of mat_ids)

• function generate pipeline report, in excel (or something else)

Okay our first idea after picking the problem back up is to store the information in a dictionary and then create different functions to display the information.

def track(func):
    @wraps(func)
    def wrapper(in_df, vrbs=False, *args, **kwargs):
        meta_dict = {
            'step_name':func.__name__, # name of the pipeline step
            'in_time':datetime.now(), # time when the pipeline step starts
            'in_df_shape':in_df.shape, # shape of the input dataframe
            'in_df_head':in_df.head(3), # head of the input dataframe
            'in_df_sample':in_df.sample(min(in_df.shape[0], 5)), # sample of the input dataframe
            'in_df_tail':in_df.tail(3), # tail of the input dataframe
        }

        out_df = func(in_df, *args, **kwargs)

        out_time = datetime.now()
        total_time = out_time - meta_dict['in_time']
            
        meta_dict.update({
            'out_time':datetime.now(), # time when the pipeline step stops
            'out_df_shape':out_df.shape, # shape of the output dataframe
            'out_df_head':out_df.head(3), # head of the output dataframe
            'out_df_sample':out_df.sample(min(out_df.shape[0], 5)), # sample of the output dataframe
            'out_df_tail':out_df.tail(3), # tail of the output dataframe
            'total_time':total_time, # difference between in_time and out_time
        })

        if vrbs:
            print('Here we use a fuction to display the information')
            print(meta_dict)
        
        return out_df
    return wrapper

@track
def filter_products(df):
    return df[df["product"] != "Widget C"]

@track
def aggregate_by_product(df):
    return df.groupby(["product"])[["quantity", "defects", "production_time"]].sum()

steps = [
    (filter_products, {}),
    (aggregate_by_product, {}),
]

_df = pipeline(df, steps)

Here we use a fuction to display the information
{'step_name': 'filter_products', 'in_time': datetime.datetime(2026, 9, 14, 12, 30, 44, 823024), 'in_df_shape': (5, 5), 'in_df_head':    order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150, 'in_df_sample':    order_id   product  quantity  defects  production_time
4       105  Widget B        45        2              110
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95, 'in_df_tail':    order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110, 'out_time': datetime.datetime(2026, 9, 14, 12, 30, 44, 824010), 'out_df_shape': (4, 5), 'out_df_head':    order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150, 'out_df_sample':    order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
4       105  Widget B        45        2              110
0       101  Widget A        50        2              120
2       103  Widget A        75        3              150, 'out_df_tail':    order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110, 'total_time': datetime.timedelta(microseconds=983)}
Here we use a fuction to display the information
{'step_name': 'aggregate_by_product', 'in_time': datetime.datetime(2026, 9, 14, 12, 30, 44, 830224), 'in_df_shape': (4, 5), 'in_df_head':    order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150, 'in_df_sample':    order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95, 'in_df_tail':    order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110, 'out_time': datetime.datetime(2026, 9, 14, 12, 30, 44, 839045), 'out_df_shape': (2, 3), 'out_df_head':           quantity  defects  production_time
product                                     
Widget A       125        5              270
Widget B        75        3              205, 'out_df_sample':           quantity  defects  production_time
product                                     
Widget A       125        5              270
Widget B        75        3              205, 'out_df_tail':           quantity  defects  production_time
product                                     
Widget A       125        5              270
Widget B        75        3              205, 'total_time': datetime.timedelta(microseconds=8813)}

display functions

Here we created a sample meta_dict to play around and create different functions to display the information

meta_dict = {
    'step_name': 'filter_products',
    'step_description': 'Exclude products which are Widget C',
    'in_time': datetime.now(),
    'in_df_shape': (5, 5),
    'in_df_head': df.head(3),
    'in_df_sample': df.sample(min(df.shape[0], 5)),
    'in_df_tail': df.tail(3),
    'out_time': datetime.now(),
    'out_df_shape': (4, 5),
    'out_df_head': df[df["product"] != "Widget C"].head(3),
    'out_df_sample': df[df["product"] != "Widget C"].sample(min(4, 5)),
    'out_df_tail': df[df["product"] != "Widget C"].tail(3),
    'total_time': timedelta(microseconds=1287)
}

meta_dict

{'step_name': 'filter_products',
 'step_description': 'Exclude products which are Widget C',
 'in_time': datetime.datetime(2026, 9, 14, 12, 30, 44, 851928),
 'in_df_shape': (5, 5),
 'in_df_head':    order_id   product  quantity  defects  production_time
 0       101  Widget A        50        2              120
 1       102  Widget B        30        1               95
 2       103  Widget A        75        3              150,
 'in_df_sample':    order_id   product  quantity  defects  production_time
 4       105  Widget B        45        2              110
 3       104  Widget C        20        0               80
 2       103  Widget A        75        3              150
 1       102  Widget B        30        1               95
 0       101  Widget A        50        2              120,
 'in_df_tail':    order_id   product  quantity  defects  production_time
 2       103  Widget A        75        3              150
 3       104  Widget C        20        0               80
 4       105  Widget B        45        2              110,
 'out_time': datetime.datetime(2026, 9, 14, 12, 30, 44, 852409),
 'out_df_shape': (4, 5),
 'out_df_head':    order_id   product  quantity  defects  production_time
 0       101  Widget A        50        2              120
 1       102  Widget B        30        1               95
 2       103  Widget A        75        3              150,
 'out_df_sample':    order_id   product  quantity  defects  production_time
 4       105  Widget B        45        2              110
 0       101  Widget A        50        2              120
 1       102  Widget B        30        1               95
 2       103  Widget A        75        3              150,
 'out_df_tail':    order_id   product  quantity  defects  production_time
 1       102  Widget B        30        1               95
 2       103  Widget A        75        3              150
 4       105  Widget B        45        2              110,
 'total_time': datetime.timedelta(microseconds=1287)}

---

source

StepMeta

def StepMeta(
    step_name:str, step_description:str, in_time:datetime.datetime, in_df_shape:tuple, in_df_head:object,
    in_df_sample:object, in_df_tail:object, out_time:datetime.datetime, out_df_shape:tuple, out_df_head:object,
    out_df_sample:object, out_df_tail:object, total_time:datetime.timedelta
)->None:

Metadata collected for a single ETL pipeline step.

step_meta = StepMeta(
    step_name='filter_products',
    step_description='Exclude products which are Widget C',
    in_time=datetime.now(),
    in_df_shape=df.shape,
    in_df_head=df.head(3),
    in_df_sample=df.sample(min(df.shape[0], 5)),
    in_df_tail=df.tail(3),
    out_time=datetime.now(),
    out_df_shape=(4, 5),
    out_df_head=df[df["product"] != "Widget C"].head(3),
    out_df_sample=df[df["product"] != "Widget C"].sample(4),
    out_df_tail=df[df["product"] != "Widget C"].tail(3),
    total_time=timedelta(microseconds=1287)
)
step_meta

StepMeta(step_name='filter_products', step_description='Exclude products which are Widget C', in_time=datetime.datetime(2026, 9, 14, 12, 30, 44, 897277), in_df_shape=(5, 5), in_df_head=   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150, in_df_sample=   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110
0       101  Widget A        50        2              120, in_df_tail=   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110, out_time=datetime.datetime(2026, 9, 14, 12, 30, 44, 897946), out_df_shape=(4, 5), out_df_head=   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150, out_df_sample=   order_id   product  quantity  defects  production_time
4       105  Widget B        45        2              110
2       103  Widget A        75        3              150
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95, out_df_tail=   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110, total_time=datetime.timedelta(microseconds=1287))

print name and time

print(15*'*' + ' ' + step_meta.step_name + ' ' + 15*'*')

*************** filter_products ***************

print(f'Total Time: {step_meta.total_time}')
print(f'Start: {step_meta.in_time}')
print(f'End: {step_meta.out_time}')

Total Time: 0:00:00.001287
Start: 2026-09-14 12:30:44.897277
End: 2026-09-14 12:30:44.897946

---

source

format_timedelta

def format_timedelta(
    td
):

print(f'Total Time: {format_timedelta(step_meta.total_time)}')

Total Time: 1.29 ms

---

source

print_step_name

def print_step_name(
    step_meta:__main__.StepMeta
):

print_step_name(step_meta)

*************** filter_products ***************

---

source

print_time

def print_time(
    step_meta:__main__.StepMeta
):

print_time(step_meta)

Total Time: 1.29 ms

Start: 2026-09-14 12:30:44.897277
  End: 2026-09-14 12:30:44.897946

print sample df

pd.concat([step_meta.in_df_head, step_meta.in_df_sample, step_meta.in_df_tail])

	order_id	product	quantity	defects	production_time
0	101	Widget A	50	2	120
1	102	Widget B	30	1	95
2	103	Widget A	75	3	150
1	102	Widget B	30	1	95
2	103	Widget A	75	3	150
3	104	Widget C	20	0	80
4	105	Widget B	45	2	110
0	101	Widget A	50	2	120
2	103	Widget A	75	3	150
3	104	Widget C	20	0	80
4	105	Widget B	45	2	110

pd.DataFrame(np.nan, index=range(1), columns=step_meta.in_df_head.columns)

	order_id	product	quantity	defects	production_time
0	NaN	NaN	NaN	NaN	NaN

pd.DataFrame(np.nan, index=range(1), columns=step_meta.in_df_head.columns).fillna('...')

	order_id	product	quantity	defects	production_time
0	...	...	...	...	...

pd.DataFrame(np.nan, index=range(3), columns=meta_dict['in_df_head'].columns).fillna('.')

	order_id	product	quantity	defects	production_time
0	.	.	.	.	.
1	.	.	.	.	.
2	.	.	.	.	.

pd.DataFrame(np.nan, index=range(1), columns=step_meta.in_df_head.columns).fillna(':')

	order_id	product	quantity	defects	production_time
0	:	:	:	:	:

def fill_between_df_parts(df):
    return pd.DataFrame(np.nan, index=[''], columns=df.columns).fillna(':')

fill_between_df_parts(meta_dict['in_df_head'])

	order_id	product	quantity	defects	production_time
	:	:	:	:	:

pd.concat(
    [step_meta.in_df_head, 
    fill_between_df_parts(step_meta.in_df_head),
    step_meta.in_df_sample, 
    fill_between_df_parts(step_meta.in_df_head),
    step_meta.in_df_tail]
    )

	order_id	product	quantity	defects	production_time
0	101	Widget A	50	2	120
1	102	Widget B	30	1	95
2	103	Widget A	75	3	150
	:	:	:	:	:
1	102	Widget B	30	1	95
2	103	Widget A	75	3	150
3	104	Widget C	20	0	80
4	105	Widget B	45	2	110
0	101	Widget A	50	2	120
	:	:	:	:	:
2	103	Widget A	75	3	150
3	104	Widget C	20	0	80
4	105	Widget B	45	2	110

The display helpers print head, sample, and tail separately, so they work with both pandas and eager Polars DataFrames without conversion. print_sample_from_meta_dict selects the input or output previews with mode; display_sample_from_df samples a DataFrame directly and prints it rather than returning a combined DataFrame.

---

source

print_sample_from_meta_dict

def print_sample_from_meta_dict(
    step_meta:__main__.StepMeta, mode:str='in'
):

print_sample_from_meta_dict(step_meta)

head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110
0       101  Widget A        50        2              120
tail:
   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110

---

source

display_sample_from_df

def display_sample_from_df(
    df
):

display_sample_from_df(df)

head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
4       105  Widget B        45        2              110
3       104  Widget C        20        0               80
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
tail:
   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110

We will have to also consider what happen if we have to many columns.

Our idea is to truncate them as well and just add a dummy columns with ‘…’ as sign that its truncated

print shape and shape change

print('(rows, columns) =', step_meta.in_df_shape)
print('        |          ')
print('        V          ')
print('(rows, columns) =', step_meta.out_df_shape)

(rows, columns) = (5, 5)
        |          
        V          
(rows, columns) = (4, 5)

print('(rows, columns) =', step_meta.in_df_shape)
print(' ↓ '*8)
print('(rows, columns) =', step_meta.out_df_shape)

(rows, columns) = (5, 5)
 ↓  ↓  ↓  ↓  ↓  ↓  ↓  ↓ 
(rows, columns) = (4, 5)

in_str = f"{step_meta.in_df_shape[0]} rows, {step_meta.in_df_shape[1]} columns"
print(in_str)
print('↓ ↓ ↓'.center(len(in_str), ' '))
print(f"{step_meta.out_df_shape[0]} rows, {step_meta.out_df_shape[1]} columns")

5 rows, 5 columns
      ↓ ↓ ↓      
4 rows, 5 columns

is there a convenietn way to space out a number of strings over a specific length?

s1, s2, s3 = "5 rows", "4 columns", "time: 1.2ms"
total_width = 50
print(f"{s1:<{total_width//3}}{s2:<{total_width//3}}{s3}")

5 rows          4 columns       time: 1.2ms

width = 20
print(s1.ljust(width) + s2.center(width) + s3.rjust(width))

5 rows                   4 columns               time: 1.2ms

---

source

space_strings

def space_strings(
    strings, total_width
):

space_strings(['|', '|', '|'], 20)

'|        |        |'

len(space_strings(['|', '|', '|'], 20))

19

in_str = f"{step_meta.in_df_shape[0]} rows, {step_meta.in_df_shape[1]} columns"
print(in_str)
print(space_strings(['↓','↓','↓'], len(in_str)))
print(f"{step_meta.out_df_shape[0]} rows, {step_meta.out_df_shape[1]} columns")

5 rows, 5 columns
↓       ↓       ↓
4 rows, 5 columns

in_rows, in_cols = step_meta.in_df_shape
out_rows, out_cols = step_meta.out_df_shape

in_str = f"{in_rows} rows, {in_cols} columns"
print(in_str)
print(space_strings(['↓','↓','↓'], len(in_str)))

print(f'{str(out_rows-in_rows).center(len(in_str)//2)}', f'{str(out_cols-in_cols).center(len(in_str)//2)}')

print(space_strings(['↓','↓','↓'], len(in_str)))
print(f"{out_rows} rows, {out_cols} columns")

5 rows, 5 columns
↓       ↓       ↓
   -1       0    
↓       ↓       ↓
4 rows, 5 columns

in_rows, in_cols = step_meta.in_df_shape
out_rows, out_cols = step_meta.out_df_shape
diff_rows, diff_cols = out_rows - in_rows, out_cols - in_cols

diff_rows_str = f"{diff_rows:+d}" if diff_rows != 0 else "0"
diff_cols_str = f"{diff_cols:+d}" if diff_cols != 0 else "0"

# Right-align each column
row_width = max(len(str(in_rows)), len(str(out_rows)), len(diff_rows_str))
col_width = max(len(str(in_cols)), len(str(out_cols)), len(diff_cols_str))

print(f"Input:  {in_rows:>{row_width}} rows, {in_cols:>{col_width}} cols")
print(f"        {' '*row_width}   ↓   {' '*col_width}   ↓")
print(f"Diff:   {diff_rows_str:>{row_width}} rows, {diff_cols_str:>{col_width}} cols")
print(f"        {' '*row_width}   ↓   {' '*col_width}   ↓")
print(f"Output: {out_rows:>{row_width}} rows, {out_cols:>{col_width}} cols")

Input:   5 rows, 5 cols
             ↓       ↓
Diff:   -1 rows, 0 cols
             ↓       ↓
Output:  4 rows, 5 cols

print(f"""
Input:  {in_rows:>{row_width}} rows, {in_cols:>{col_width}} cols
        {' '*row_width}   ↓   {' '*col_width}   ↓
Diff:   {diff_rows_str:>{row_width}} rows, {diff_cols_str:>{col_width}} cols
        {' '*row_width}   ↓   {' '*col_width}   ↓
Output: {out_rows:>{row_width}} rows, {out_cols:>{col_width}} cols
""")

Input:   5 rows, 5 cols
             ↓       ↓
Diff:   -1 rows, 0 cols
             ↓       ↓
Output:  4 rows, 5 cols

---

source

print_shape_change

def print_shape_change(
    step_meta:__main__.StepMeta
):

print_shape_change(step_meta)

        Input:   5 rows, 5 cols
                     ↓       ↓
        Diff:   -1 rows, 0 cols
                     ↓       ↓
        Output:  4 rows, 5 cols
        

print docstring

@track
def my_func():
    """This is the docstring."""
    pass

print(my_func.__doc__)

This is the docstring.

---

source

print_step_description

def print_step_description(
    step_meta:__main__.StepMeta
):

print_step_description(step_meta)

'''Exclude products which are Widget C'''

print all function

---

source

print_step_info

def print_step_info(
    step_meta:__main__.StepMeta
):

print_step_info(step_meta)

*************** filter_products ***************
'''Exclude products which are Widget C'''


Total Time: 1.29 ms

Start: 2026-09-14 12:30:44.897277
  End: 2026-09-14 12:30:44.897946

Input DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110
0       101  Widget A        50        2              120
tail:
   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110

        Input:   5 rows, 5 cols
                     ↓       ↓
        Diff:   -1 rows, 0 cols
                     ↓       ↓
        Output:  4 rows, 5 cols
        
Output DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
4       105  Widget B        45        2              110
2       103  Widget A        75        3              150
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
tail:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110

Testing the pipeline

---

source

pipeline

def pipeline(
    df, steps, vrbs_default:bool=True
):

---

source

track

def track(
    func
):

@track
def filter_products(df):
    '''Exclude products which are Widget C.'''
    return df[df["product"] != "Widget C"]

@track
def aggregate_by_product(df):
    '''aggregate by product and sum columns "quantity", "defects", "production_time"'''
    return df.groupby(["product"])[["quantity", "defects", "production_time"]].sum()

steps = [
    (filter_products, {'vrbs':True}),
    (aggregate_by_product, {}),
]

_df = pipeline(df, steps, vrbs_default=False)

*************** filter_products ***************
'''Exclude products which are Widget C.'''


Total Time: 1.05 ms

Start: 2026-09-14 12:30:45.689253
  End: 2026-09-14 12:30:45.690299

Input DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110
3       104  Widget C        20        0               80
1       102  Widget B        30        1               95
0       101  Widget A        50        2              120
tail:
   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110

        Input:   5 rows, 5 cols
                     ↓       ↓
        Diff:   -1 rows, 0 cols
                     ↓       ↓
        Output:  4 rows, 5 cols
        
Output DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
4       105  Widget B        45        2              110
2       103  Widget A        75        3              150
0       101  Widget A        50        2              120
tail:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110

_df = pipeline(df, steps, vrbs_default=True)

@track
def take_rows(df, n):
    return df.head(n)

preview = df.pipe(take_rows, 4, vrbs=True).pipe(take_rows, 2, vrbs=True)
assert preview.shape == (2, df.shape[1])
preview

*************** filter_products ***************
'''Exclude products which are Widget C.'''


Total Time: 1.29 ms

Start: 2026-09-14 12:30:45.716039
  End: 2026-09-14 12:30:45.717331

Input DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
1       102  Widget B        30        1               95
4       105  Widget B        45        2              110
3       104  Widget C        20        0               80
0       101  Widget A        50        2              120
tail:
   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110

        Input:   5 rows, 5 cols
                     ↓       ↓
        Diff:   -1 rows, 0 cols
                     ↓       ↓
        Output:  4 rows, 5 cols
        
Output DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
0       101  Widget A        50        2              120
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110
tail:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110


*************** aggregate_by_product ***************
'''aggregate by product and sum columns "quantity", "defects", "production_time"'''


Total Time: 10.25 ms

Start: 2026-09-14 12:30:45.723587
  End: 2026-09-14 12:30:45.733836

Input DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110
tail:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110

        Input:   4 rows,  5 cols
                     ↓        ↓
        Diff:   -2 rows, -2 cols
                     ↓        ↓
        Output:  2 rows,  3 cols
        
Output DataFrame:
head:
          quantity  defects  production_time
product                                     
Widget A       125        5              270
Widget B        75        3              205
sample:
          quantity  defects  production_time
product                                     
Widget A       125        5              270
Widget B        75        3              205
tail:
          quantity  defects  production_time
product                                     
Widget A       125        5              270
Widget B        75        3              205


*************** take_rows ***************
''''''


Total Time: 548 µs

Start: 2026-09-14 12:30:45.740618
  End: 2026-09-14 12:30:45.741166

Input DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
4       105  Widget B        45        2              110
3       104  Widget C        20        0               80
0       101  Widget A        50        2              120
tail:
   order_id   product  quantity  defects  production_time
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
4       105  Widget B        45        2              110

        Input:   5 rows, 5 cols
                     ↓       ↓
        Diff:   -1 rows, 0 cols
                     ↓       ↓
        Output:  4 rows, 5 cols
        
Output DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
0       101  Widget A        50        2              120
3       104  Widget C        20        0               80
tail:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80


*************** take_rows ***************
''''''


Total Time: 461 µs

Start: 2026-09-14 12:30:45.746451
  End: 2026-09-14 12:30:45.746912

Input DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
sample:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80
0       101  Widget A        50        2              120
tail:
   order_id   product  quantity  defects  production_time
1       102  Widget B        30        1               95
2       103  Widget A        75        3              150
3       104  Widget C        20        0               80

        Input:   4 rows, 5 cols
                     ↓       ↓
        Diff:   -2 rows, 0 cols
                     ↓       ↓
        Output:  2 rows, 5 cols
        
Output DataFrame:
head:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
sample:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95
tail:
   order_id   product  quantity  defects  production_time
0       101  Widget A        50        2              120
1       102  Widget B        30        1               95

	order_id	product	quantity	defects	production_time
0	101	Widget A	50	2	120
1	102	Widget B	30	1	95

Summarize what we did so far

🤖Reply🤖

🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠

Here’s a summary of what you’ve built so far:

Goal: A lightweight ETL pipeline framework for pandas that lets you chain transformation steps and inspect intermediate results — aimed at your work as Data Science Consultants in manufacturing.

What’s been built:

1. pipeline(df, steps, vrbs_default) — Runs a list of (function, kwargs) steps sequentially on a DataFrame, with per-step control over verbosity via the vrbs flag.

2. @track decorator — Wraps each step function to optionally collect and display metadata (timing, shapes, data samples) using @wraps to preserve docstrings.

3. StepMeta dataclass — A structured container for all the metadata captured per step: name, description, timing, input/output shapes, head/sample/tail of DataFrames.

4. Display helper functions:

   • print_step_name — banner with step name
   • print_step_description — docstring display
   • print_time / format_timedelta — human-readable timing (µs, ms, s, min, h)
   • print_shape_change — nice visual showing row/col diffs with arrows
   • print_sample_from_meta_dict — head + sample + tail with : separator rows
   • display_sample_from_df — same but directly from a DataFrame
   • print_step_info — combines all the above into one output
   • fill_between_df_parts / space_strings — formatting utilities

Still on the to-do list (from your reflection note): - Column truncation for wide DataFrames - Side-by-side layout (metadata next to DataFrames) - df.info() / df.describe() integration - Column diff (added/removed columns) - Source code display of step functions - HTML/rich display - Assertion/sanity-check wrapper - Pipeline report generation (e.g. Excel)