| from typing import Dict, Any |
| from flows.base_flows.atomic import AtomicFlow |
|
|
|
|
| class SaveCodeAtomicFlow(AtomicFlow): |
| def _check_input(self, input_data: Dict[str, Any]): |
| assert "code" in input_data, "code is not passed to SaveCodeAtomicFlow" |
| assert "memory_files" in input_data, "memory_files is not passed to SaveCodeFlow" |
| assert "code_library" in input_data["memory_files"], "code_library not in memory_files" |
| |
| code_library_location = input_data["memory_files"]["code_library"] |
| with open(code_library_location, 'a') as file: |
| pass |
|
|
| def _call(self, input_data: Dict[str, Any]): |
| try: |
| code_to_append = input_data["code"] |
| code_lib_loc = input_data["memory_files"]["code_library"] |
| with open(code_lib_loc, 'a') as file: |
| file.write(code_to_append + '\n') |
| return { |
| "result": "code successfully written to the library", |
| "summary": f"ExtendLibrayFlow/SaveCodeAtomicFlow: code written to {code_lib_loc}" |
| } |
| except Exception as e: |
| error_msg = str(e) |
| return { |
| "result": f"error occurred: {error_msg}", |
| "summary": f"ExtendLibrayFlow/SaveCodeAtomicFlow: error occurred: {error_msg}" |
| } |
|
|
| def run( |
| self, |
| input_data: Dict[str, Any] |
| ): |
| self._check_input(input_data) |
| return self._call(input_data) |
|
|
