GreaterThan, LLC

The is only one notable change in 0.3.1: The pipelayer.Step base class was converted to an interface (a Protocol). This means for your implementation, all you need to do is create a class that implements a run method with this signature:

class MyFilter:
    def run(data: Any, context: Context) -> Any:
        ...

As I continued to work on the microservice example (using v0.3.0), I found I was not creating filters by sub-classing pipelayer.Filter. Doing so would have meant I had to create a separate class for each filter, and so any logical grouping was placed on the directory structure of the project and not the domain namespace. For example:

from service.config import AppContext

class UserRequest(Filter):

    @staticmethod
    def run(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 opposed to this:

from service.config import AppContext
from service.model.resreq_model import ResReqModel

class MapResReq(Filter):

    @staticmethod
    def from_resreq_api_response(response: dict, 
                                 context: AppContext) -> ResReqModel:
        resreq = ResReqModel.parse_obj(response)
        return resreq

A subtle difference, but here’s how these filters would look in a pipeline:

get_user_pipeline = Pipeline(
    name="Get User Pipeline",
    steps=[
        MapRequest.from_get_user_request,
        UserRequest,                         # pipelayer.Filter
        MapResReq.from_resreq_api_response,  # static method
        MapUser.from_resreq_model
    ]
)

Since class names are generally nouns and functions are verbs, I think using a custom static method reads more clearly.

Now, one of the benefits of implementing a sub-class of Filter is that can enforce a separation of concerns. It would be very easy to create a single class with every filter in your app using the static method approach.

NOTE: If your list of filters can change at runtime, implementing the pipelayer.Step interface (or sub-classing pipelayer.Filter) is the easiest way to implement a pipeline.