OnlyOnePixel

Share this post

How To Build A Log Orchestrator in Python

onlyonepixel.substack.com
Software Engineering

How To Build A Log Orchestrator in Python

Logging To Multiple Destinations In A Python Application

Uchechukwu Emmanuel
Apr 19, 2022
2
Share this post

How To Build A Log Orchestrator in Python

onlyonepixel.substack.com

Hey guys and girls! If you are reading this, I’m assuming you at least know what logging means in programming, but for the sake of those who might be new to the concept, I’ll give a brief introduction to logging in python. Please feel free to skip the intro if you already know what it’s all about.

WHAT IS LOGGING

A log is a record of events that occur in an operating system or other software programs. Logging is the act of keeping a log.

LOGGING IN PYTHON

Python comes with a logging module in the standard library that provides a flexible framework for emitting log messages from Python programs. This module is widely used by libraries and is the first go-to point for most developers when it comes to logging.

The module provides a way for applications to configure different log handlers and a way of routing log messages to these handlers. This allows for a highly flexible configuration that can deal with a lot of different use cases.

To emit a log message, a caller first requests a named logger. The name can be used by the application to configure different rules for different loggers. This logger can then be used to emit simply-formatted messages at different log levels (DEBUG, INFO, ERROR, etc.), which again can be used by the application to handle messages of higher priority different than those of a lower priority. While it might sound complicated, it can be as simple as this:

import logging
log = logging.getLogger("my-logger")
log.info("Hello, world")

Internally, the message is turned into a LogRecord object and routed to a Handler object registered for this logger. The handler will then use a Formatter to turn the LogRecord into a string and emit that string.

If at this point, you still don’t know what I am talking about, please click me. I will wait for you to catch up and come back.

OKAY LET’S GET TO THE REAL DEAL

Do me a favor and imagine that you are Elon Musk and you have an application that handles your:

  • Rocket Launches

  • Tesla Pre-Orders

  • Twitter Stocks

But you have a challenge, you have noticed that you cannot send a log to different channels (File, Slack, Instagram DM, SMS, Email, etc) at the same time when certain events occur( for example when someone preorders 100 Tesla Model S).

So Elon, I would like to help you distribute your logs to several destinations at the same time.

THE LOG ORCHESTRATOR

So essentially, the problem here is that you need a way to receive updates about your products on different channels/destinations.

Problem —> Distribute log to different channels (File, Slack, Email, iMessage, etc)

To solve this problem, I am going to give you a folder structure to best follow my examples.

orchestrators
|___factories
|    |__init__.py
|    |log_orchestrator.py
|
|____tests
|     |test_log_orchestrator.py
|     |__init__.py
|

So as you can see in the folder structure above, the orchestrators directory is the base directory and  log_orchestrator.py file is where the log orchestrator implementation will live.

FACTORY METHOD DESIGN PATTERN AND PYTHON ABSTRACT BASE CLASSES

Before we continue, I would like to walk you through how we would go about this problem.

We will need a way to send a log to a channel or a list of channels a.k.a destinations. Ideally, the best way to go about it will be to call a method that sends this log to these channels without worrying about how it will be sent to the various destinations.

One way to go about this is to use the factory method pattern.

A Factory Pattern or Factory Method Pattern suggests that we should just define an interface or abstract class for creating an object but let the subclasses decide which class to instantiate. In other words, subclasses should be responsible for creating the instances of the class.

This will be helpful in our case because:

  • We really do not know which channels we want to send logs to per time. that is, we might want to send a log to all destinations or just Slack and Email.

  • We might want the logs to be formatted differently for the different destinations that we have. An SMS log might be text-based and an Email log might be in HTML.

It will also promote loose coupling by eliminating the need to bind application-specific classes into the code.

THE ABSTRACT LOGGER

#log_orchestrator.py

import logging
from abc import ABC, abstractmethod
from api_libraries.slack_library import SlackAPIClient
from typing import List, Dict, Callable, Optional, Any

slack_api_client = SlackAPIClient()



class Logger(ABC):
    """
    Inherits from ABC(Abstract Base Class)

    This is the base class from which all the 
    different loggers      are created
    it exposes one method send_log, which must be
    implemented by the concrete classes.
    """

    
    @abstractmethod
    def send_log(self, data, channel: str, options: Dict):
        """
        This abstract method / interface, must be 
        implemented in the concrete classes
        """
        pass

The Logger class is an abstract class (inherits Python ABC) and it provides an interface send_log which concrete classes will implement.

What this means is that if I want to create a Slack Logger, I will just need to inherit from the Logger abstract class and implement the send_log interface.

Slack Logger

#log_orchestrator.py

class SlackLogger(Logger):
    """
    This concrete class inherits from the Logger 
    abstract class and implements the send_log() interface
    It is intended to be used for Slack Logs
    """


    def send_log(self, data, channel: str, options: Dict):
        """
        this method confirms that the channel is truly
        what it expects
        and propagates data to the channel
        """
        self._log_to_slack(data, options=options)
        return


    def invalid_message_type(self):
        """
        This method is called when an invalid
        message type is detected
        """
        raise ValueError("Invalid slack message type")


    def _log_to_slack(self, data, options):
        """
        This method orchestrates and propagates
        log data with the appropriate slack
        message formatting template
        """
        if options is None:
            raise ValueError('slack channel is required')
        message_type = options.get('message_type', None)

        log_methods = {
            "rocket_launch_notifications": 
        slack_api_client.send_rocket_launch_notifications,
            "tesla_preorder_requests":
        slack_api_client.send_tesla_preorder_notifications,
            "twitter_stock_notifications":
        slack_api_client.send_twitter_stock_updates
        }
        chosen_function = log_methods.get(
        message_type, SlackLogger.invalid_message_type
        )

        return chosen_function(data)

File Logger

#log_orchestrator.py

class FileLogger(Logger):
    """
    This concrete class inherits from the Logger abstract class and implements the send_log() interface
    It is intended to be used for File Logs
    """


    def send_log(self, data, channel, options=None):
        """
        this method confirms that the channel 
        is truly what it expects
        and propagates data to the channel
        """

        self._log_to_file(data, options)
        return


    def invalid_logger_obj(self):
        raise ValueError(
           "please provide a valid logger object"
         )



    def _log_to_file(self, data, options=None):
        """
        This method is intended to orchestrate
        and propagate log data
        with the appropriate message formatting template.

        it will also deliver logs based on the log levels

        for now, we are sending the same data
        to slack and file.
        """
        chosen_log_level = options.get('log_level', 'debug')

        chosen_logger_obj = options.get(
        'logger_obj', self.invalid_logger_obj
        )

        all_log_levels = {
            "debug": chosen_logger_obj.debug,
            "info": chosen_logger_obj.info,
            "warning": chosen_logger_obj.warning,
        }

        chosen_function = all_log_levels.get(chosen_log_level)

        return chosen_function(data)

From the implementations above, you can already see a pattern. To add an Instagram DM logger, all we need to do is inherit from the abstract Logger class and implement the send_log abstract method.

We have just a few steps left. So we already have:

  • An abstract Logger class that exposes the send_log interface

An interface that all concrete loggers must implement

So what we need now is a way to decide which concrete logger class should be instantiated every time we send a log (where should we send the log? Slack? or Email?).

The Factory Method

To accomplish this, we will need a factory method (yes that is where the name of the design pattern came from). The job is to select the appropriate concrete class to instantiate.

def log_factory(channel):
    """
    This function is a factory, it's job is to
    select the appropriate concrete class based
    on the provided channel
    """
    if channel == 'file':
        return FileLogger()

    if channel == 'slack':
        return SlackLogger()

    #logging to email follows the same procedure
    # if channel == 'email':
    #     return EmailLogger()

    #logging to instagram DM follows the same procedure
    # if channel == 'instagram':
    #     return InstagramDMLogger()


The Entry Point

The final step is to define an entry point to the entire log orchestrator. This is the only thing your code will communicate with.

def log(data, channels: List[str], options=None) -> None:
    """
    This is the entrypoint for the log orchestrator
    All logs will be initiated here.
    """
    if options is None:
        options = {}
    for channel in channels:
        logger = log_factory(channel)
        if logger is None:
            raise ValueError(f"{channel} is invalid")
        logger.send_log(data, channel, options)

Params


data: Is the log data to be sent to destinations or channels

channel: the destinations for your log data

options: is a dictionary that can be used to pass additional information to the logger If the channel is 'file', the options dictionary can contain the following keys:

  • log_level: the log level to be used for the log

  • logger_obj: the logger object to be used for the log

If the channel is 'slack', the options dictionary can contain the following keys:

  • message_type: the message type to be used for the log

  • channel: the Slack channel where the log will be sent

HOW WILL YOU USE THIS?

So we have set up the log orchestrator but how will it be used?

Using the log orchestrator is as easy as importing the entry point function and passing the required arguments as specified above.

Note that the actual implementation for your channels a.k.a destinations might be different.

So Elon wants to be alerted when someone preorders more than Ten Tesla Model S

#tesla_preorders.py

from orchestrators.factories.log_orchestrator import log

def get_model_s_preorder_details(model_s_pre_order_data):
    if model_s_pre_order_data.car_count > 10:
            log_data = {
             "message": "yo! tier x customer here"
            }
            log(log_data, ['file', 'slack', 'iMessage'])
            ...

HOW SHOULD THIS BE TESTED?

Unit testing will depend on your own implementation of this orchestrator. but below is something that might guide you.

import unittest
from unittest.mock import Mock, patch

from factories import log_orchestrator


class TestLog(unittest.TestCase):


    @patch('utilities.factories.FileLogger')
    @patch('utilities.factories.log')
    def test_can_send_log(self, mock_logger, mock_send_log):
        """
        Test that a log can be sent
        """
        mock_logger.side_effect = mock_send_log.send_log
        data = "test log data"
        logger_obj = Mock()
        message_type = Mock()

        log_orchestrator.log(
        data, 
        ['file', 'slack'], 
        options={
        "message_type": message_type,
        "logger_obj": logger_obj
        })

        mock_logger.assert_called_with(
            data,
            ['file', 'slack'],
            options={"message_type": message_type,
            "logger_obj": logger_obj
            })

        mock_send_log.send_log.assert_called_with(
            data,
            ['file', 'slack'],
            options={"message_type": message_type,
            "logger_obj": logger_obj
            })

So that will be all for today. I hope you find this helpful and easy to understand.

Happy Hacking 👽

References:

Loggly, Ultimate Guide to Logging, https://www.loggly.com/ultimate-guide/python-logging-basics/

Wikipedia, Elon Musk, https://en.wikipedia.org/wiki/Elon_Musk


Wikipedia, Factory Method Pattern, https://en.wikipedia.org/wiki/Factory_method_pattern

Share this post

How To Build A Log Orchestrator in Python

onlyonepixel.substack.com
Comments
TopNewCommunity

No posts

Ready for more?

© 2023 Uchechukwu Emmanuel
Privacy ∙ Terms ∙ Collection notice
Start WritingGet the app
Substack is the home for great writing