Skip to content

Instantly share code, notes, and snippets.

@clodo
Last active January 6, 2019 17:54
Show Gist options
  • Save clodo/f2086101ea8c9b7c3f47917b22315c01 to your computer and use it in GitHub Desktop.
Save clodo/f2086101ea8c9b7c3f47917b22315c01 to your computer and use it in GitHub Desktop.
Python: Strategy + Simple Factory
class InvalidEventTypeError(Exception):
def __init__(self, message):
self.message = message
class Event:
def __init__(self, type, auction_id):
self.type = type
self.auction_id = auction_id
class EventService:
def __init__(self, elasticService, s3Service):
self.elasticService = elasticService
self.s3Service = s3Service
def process(self, event, strategy):
return strategy.process(event)
class StrategyFactory:
@staticmethod
def factory(type):
if type == "win":
return WinStrategy()
if type == "loss":
return LossStrategy()
if type in ["click", "play", "stop"]:
return ConversionStrategy()
raise InvalidEventTypeError("Invalid Event Type {}".format(type))
class ProcessEventStrategy:
def process(self, event):
pass
class WinStrategy(ProcessEventStrategy):
def process(self, event):
print("Processing win: {}".format(event.auction_id))
# print("Call {}".format(self.elasticService))
# print("Call {}".format(self.s3Service))
class LossStrategy(ProcessEventStrategy):
def process(self, event):
print("Processing loss: {}".format(event.auction_id))
# print("Call {}".format(self.elasticService))
# print("Call {}".format(self.s3Service))
class ConversionStrategy(ProcessEventStrategy):
def process(self, event):
print("Processing conversion: {}".format(event.auction_id))
# print("Call {}".format(self.elasticService))
# print("Call {}".format(self.s3Service))
if __name__ == '__main__':
winEvent = Event(type="win", auction_id="mopub-1")
lossEvent = Event(type="loss", auction_id="mopub-2")
clickEvent = Event(type="click", auction_id="mopub-3")
elasticService = "ElasticService"
s3Service = "S3Service"
eventService = EventService(elasticService, s3Service)
eventService.process(winEvent, StrategyFactory.factory(winEvent.type))
eventService.process(lossEvent, StrategyFactory.factory(lossEvent.type))
eventService.process(clickEvent, StrategyFactory.factory(clickEvent.type))
try:
eventService.process(clickEvent, StrategyFactory.factory("unknown"))
except InvalidEventTypeError as err:
print("Error: {}".format(err.message))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment