How To Build A Log Orchestrator in Python
How To Build A Log Orchestrator in Python
Logging To Multiple Destinations In A Python Application
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
"""
passThe 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