How I applied factory pattern to query prices

What is factory pattern

Tying your code to a concrete class can make it more fragile and less flexible.

Head First book gives example like this.

Duck duck;

if (picnic){
    duck = new MallardDuck();
} else if (hunting) {
    duck = new DecoyDuck();
} else if (inBathTub){
    duck = RubberDuck();
}

Here we have got several concrete classes being instantiated, and the decision of which to instantiate is made at runtime depending on some set of conditions.

When you see code like this, you know that when it comes time for changes or extensions, you'll have to reopen this code and examine what needs to be added (or deleted). Often this kind of code ends up in several parts of the application, making maintenance and updates more difficult and error-prone.

By coding to an interface, you know you can insulate yourself from many of the changes that might happen to a system down the road. Why? If your code is written to an interface, then it will work with any new classes implementing that interface through polymorphism. Howevere when you have code that makes use of lots of concrete classes, you are looking for trouble because that code may have to be changed as new concrete classes are added.

Whenever you are dealing with types of classes and having to instantiate different types of classes based on some condition, we can extract that part and create a Factory Class. Factory Class is responsible to instantiate different types of classes based on input.

How I applied Factory pattern to my use case

I often query prices for equities of FI securities. Equity prices are estored in eq_prices table and FI prices are stored in fi_prices table. You can see that depending on a condition, equity vs FI I will need to query different tables.

My factory class looks like below

class PriceQueryFactory:
    def createPriceQuery(self, asset_type, dbh):
        if asset_type == PriceAssetType.eq:
            return EquityPriceQuery(dbh)
        elif asset_type == PriceAssetType.fi:
            return FIPriceQuery(dbh)

And this is how I use the factory class in action

def get_cusip_prices_adate_with_px_hi(cusips, adate, dbh, con, px_hi='NAV', asset_type=PriceAssetType.eq, chunk_size=400):
    price_query_factory = PriceQueryFactory()
    price_query_obj = price_query_factory.createPriceQuery(asset_type, dbh)
    price_query_obj = PxHierPriceQuery(price_query_obj, px_hi)
    return get_db_records_in_chunks(cusips, price_query_obj.get_px_asof, con, chunk_size, adate)

This way I am decoupling the function get_cusip_prices_adate_with_px_hi from actual implementations of like EquityPriceQuery or FIPriceQuery. I can easily swap EquityPriceQuery with EquityPriceQueryV2 without impacting the function.