---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
Cell In[22], line 1
----> 1 get_ipython().run_cell_magic('cell_to_module', 'adjustments --display display_config --config config', 'import stat\nimport pandas as pd\nfrom datetime import datetime\nfrom pathlib import Path\nfrom typing import Dict, List, Optional, Tuple\nfrom duco_api.duco_api import DucoCube\n\nfrom hamilton.htypes import Parallelizable, Collect\nfrom duco_api.util import unzip_bytes\nfrom hamilton.function_modifiers import pipe_output, pipe_input, step, extract_fields, config\nfrom pandas.tseries.offsets import BMonthBegin\n\n\ndef _most_recent_business_month_start(run_date: Optional[datetime] = None) -> pd.Timestamp:\n    """Return the most recent business month start date in the past with time zeroed out."""\n    \n    today = pd.Timestamp.today() if run_date is None else run_date\n    \n    # Calculate the first business day of the current month\n    first_business_day_this_month = (today.replace(day=1) + BMonthBegin()).normalize()\n    \n    if today >= first_business_day_this_month:\n        # If today is on or after the first business day of the current month\n        return first_business_day_this_month\n    else:\n        # Otherwise, return the first business day of the previous month\n        first_business_day_last_month = (today.replace(day=1) - pd.DateOffset(months=1) + BMonthBegin()).normalize()\n        return first_business_day_last_month\n    \n\ndef _filter_run_latest(df: pd.DataFrame) -> pd.DataFrame:\n    """Return last run id for process."""\n    return df[df["id"].idxmax()]\n\n\ndef _filter_run_bom(df: pd.DataFrame, run_date: Optional[datetime] = None) -> pd.DataFrame:\n    """Return last run id for process finishing \n    on the business day first of month.\n    """\n\n    df["completed_at"] = pd.to_datetime(df["completed_at"]).dt.normalize()\n\n    biz_bom = _most_recent_business_month_start().strftime("%Y-%m-%d")\n    filtered_df = df.loc[df["completed_at"] == biz_bom]\n    if filtered_df.empty:\n        print("NO DATA")\n    \n    return filtered_df.head(1)\n\n\ndef _filter_run_id(df: pd.DataFrame) -> pd.DataFrame:\n    """Fetch extract given run id."""\n    # TODO\n\n\ndef child_rec_process_code(child_rec_process_codes: list) -> Parallelizable[str]:\n    for process_code in child_rec_process_codes:\n        yield process_code\n\n\ndef process_code_runs_endpoint(child_rec_process_code: str) -> str:\n    return f"/processes/{child_rec_process_code}/runs"\n\n\ndef child_process_runs(\n    process_code_runs_endpoint: str, child_rec_process_code: str, duco: DucoCube\n) -> pd.DataFrame:\n    # TODO - get latest runs\n    data = duco.fetch(process_code_runs_endpoint, "get")\n    _df = pd.json_normalize(data)\n    _df.insert(0, "process_code", child_rec_process_code)\n    return _df\n\n\n@extract_fields({\n    "child_process_run_id": int,\n    "child_process_completed_at": pd.Timestamp,\n})\n@pipe_input(\n    step(_filter_run_bom).when(last_run="bom"),\n    step(_filter_run_latest).when(last_run="last"),\n)\ndef child_process_run_info(child_process_runs: pd.DataFrame) -> Dict[int, pd.Timestamp]:\n    """Return run id for process from `last_run` logic."""\n    return {\n        "child_process_run_id": child_process_runs["id"].iloc[0],\n        "child_process_completed_at": pd.to_datetime(child_process_runs["completed_at"].iloc[0]),\n    }\n\n\ndef last_run_result_endpoint(\n    child_rec_process_code: str, child_process_run_id: int, breaks_only: bool = True\n) -> str:\n    query_string = "?breaks_only=true" if breaks_only else ""\n    return f"/processes/{child_rec_process_code}/runs/{child_process_run_id}/xlsx_results{query_string}"\n\n\ndef last_run_result_inputs(\n    last_run_result_endpoint: str, \n    child_process_completed_at: pd.Timestamp\n) -> Tuple[str, pd.Timestamp]:\n    return last_run_result_endpoint, child_process_completed_at\n\n\ndef all_last_run_result_inputs(\n    last_run_result_inputs: Collect[Tuple[str, pd.Timestamp]]\n) -> List[Tuple[str, pd.Timestamp]]:\n    return last_run_result_inputs\n\n\n@config.when_not(last_run="run-id")\ndef last_run_results__default(\n    all_last_run_result_inputs: List[Tuple[str, pd.Timestamp]], \n    duco: DucoCube,\n    save_location: str,\n) -> List[pd.DataFrame]:\n    result_dfs = []\n    for endpoint, completed_at in all_last_run_result_inputs:\n        response = duco.fetch(endpoint, "get", json_response=False)\n        unzipped_results = unzip_bytes(response.content, ".", prefix=f"{completed_at:%Y%m%d}")\n        if unzipped_results:\n            file_path = unzipped_results[0]\n            _df = pd.read_excel(file_path)\n            result_dfs.append(_df)\n        \n    return result_dfs\n\n\n@config.when(last_run="run-id")\ndef last_run_results__runid(\n    process_code: str,\n    run_id: str,\n    duco: DucoCube,\n    save_location: str,\n    breaks_only: bool = True\n) -> List[pd.DataFrame]:\n    """Download run results given `process_code` and `run_id`"""\n    endpoint = f"/processes/{process_code}/runs/{run_id}"\n    resp = duco.fetch(endpoint, "get", json_response=True)\n    completed_at = pd.to_datetime(resp["completed_at"])\n    \n    query_string = "?breaks_only=true" if breaks_only else ""\n    response = duco.fetch(f"{endpoint}/xlsx_results{query_string}", "get", json_response=False)\n    results_dir = Path(save_location) / f"{completed_at:%Y-%m-%d}"\n    results_dir.mkdir(parents=True, exist_ok=True)\n    unzipped_results = unzip_bytes(response.content, save_location, prefix=f"{completed_at:%Y%m%d}")\n\n    if unzipped_results:\n        file_path = unzipped_results[0]\n        _df = pd.read_excel(file_path)\n\n    return _df\n')

File c:\codebase\api-duco\duco-venv\Lib\site-packages\IPython\core\interactiveshell.py:2541, in InteractiveShell.run_cell_magic(self, magic_name, line, cell)
   2539 with self.builtin_trap:
   2540     args = (magic_arg_s, cell)
-> 2541     result = fn(*args, **kwargs)
   2543 # The code below prevents the output from being displayed
   2544 # when using magics with decorator @output_can_be_silenced
   2545 # when the last Python token in the expression is a ';'.
   2546 if getattr(fn, magic.MAGIC_OUTPUT_CAN_BE_SILENCED, False):

File c:\codebase\api-duco\duco-venv\Lib\site-packages\hamilton\plugins\jupyter_magic.py:355, in HamiltonMagics.cell_to_module(self, line, cell)
    353 # build the Driver. the Builder is copied to avoid conflict with the user namespace
    354 builder = base_builder.copy()
--> 355 dr = builder.with_config(config).with_modules(cell_module).build()
    357 # determine final vars
    358 if args.execute not in [True, None]:

File c:\codebase\api-duco\duco-venv\Lib\site-packages\hamilton\driver.py:1907, in Builder.with_config(self, config)
   1900 def with_config(self, config: Dict[str, Any]) -> "Builder":
   1901     """Adds the specified configuration to the config.
   1902     This can be called multilple times -- later calls will take precedence.
   1903 
   1904     :param config: Config to use.
   1905     :return: self
   1906     """
-> 1907     self.config.update(config)
   1908     return self

TypeError: 'ABCMeta' object is not iterable