VorTECHsa/uhura

Swahili for freedom # Uhura A framework for tackling legacy data engineering code. Wrap inputs and outputs for your system and take advantage of asyncio + automated testing. Uhura makes it easy to run pipelines in a specific "mode". Modes are implemented using context managers and are stored in the `modes` module. Right now Uhura supports two main use cases: 1. A testing mode used to run a pipeline and compare its output against pre-generated fixtures. This is executed through the `fixture_builder_mode` and the `task_test_mode` contexts. 2. A mode used for testing whether functions respect specific properties. This is executed through the `test_transformers` context. More details on how to run the two modes can be found in the Examples section below ## Examples ### 1. Pipeline testing mode Imagine you have some legacy ETL code: ```python import pandas as pd def load_data() -> pd.DataFrame: ... # IO code here def load_ids_of_interest() -> set[int]: ... # More IO code async def write_filtered_data(df: pd.DataFrame): ... async def main(): data = load_data() ids = load_ids_of_interest() await write_filtered_data(data[data['id'].isin(ids)]) ``` These functions might be reading from, and writing to, live cloud resources. They may also be loading large amounts of data, all of which makes it tricky/ unsafe to run this code locally. If something goes wrong we can use logging to try and understand what's going on in the cloud. But print statement debugging is no substitute for a proper local debugger. If we had decently separated IO and business logic we might be able to use local unit tests, but the existing code might not be set up for this, and refactoring without tests in the first place is quite risky. To solve this problem, uhura allows you to wrap arbritary IO code, and substitute it when testing. It also contains tools to speed up the process of building local end-to-end tests. So what does this wrapping process look like? First we decorate our functions which provid …