GreaterThan, LLC

As soon as I’d published the last version to PyPi, I’d started on the next round of improvements. Both the Pipeline and Filter class had a run method (diff signatures though), so how hard would it be to nest a Pipeline to be added alongside the filters? It turns out, not nearly has difficult as I’d feared. In the Pipeline, I swapped the pipeline constructor and run methods, and created a new base class for Pipelines and Filters. Voila! I took another pass through at refactoring the run methods, removing the base Settings class (as it’s really an implementation concern).

Fine, but Will It Blend?

Once the tests were passing it was time to create a “real-world” app. I got a connexion/flask microservice up and running fairly quickly with two endpoints.

Here’s what creating and running a simple pipeline looks like now:

# The pipeline
get_users_pipeline: Pipeline = Pipeline(
    name="Get Users"
    steps=[
        query_users_pipeline,  # <-- A nested pipeline!
        MapResReq.from_resreq_list_api_response,
        MapUser.from_resreq_list
    ]
)

# The api endpoint
def get_users(**kwargs):
    users = get_users_pipeline.run(kwargs, context)
    return users.dict()

And, one of the filters:

@staticmethod
def get_user(request: UserRequest, context: AppContext) -> dict:
    req = (
        f"{context.settings.resreq_api}/"
        f"{request.api_name}/{request.id}"
    )
    resreq_resp = requests.get(req)
    user = json.loads(resreq_resp.content)
    context.log.info("User received from ResReq")
    return user

As I built out each step in the pipeline, I found I had to rethink how to pass data (without any args) from one step to the next without having any knowledge of what the next step required (hint: make everything a model). For testing purposes, I could see how this new pattern would accelerate authoring tests, and help to adopt a TDD approach.

You’ll find the latest package at PyPi, or check out the GitHub repo which has the flask app in the “examples” directory.