charlesreid1.com blog

A Singleton Configuration Class in Python

Posted in Python

permalink

Overview

In this post we cover a strategy for managing configurations for programs using a Singleton pattern to create a static Config class.

This allows the user to create an instance of the Config class, pointing it to a specific config file, which it loads into memory.

The Config class provides several static methods for accessing configuration options from the config file. Here's an example of its usage:

Config('/path/to/config.json')

if Config.get_foo() == "bar":
    do_stuff()

The principle is to define one configuration file location, and be done with it. The configuration file is JSON formatted, but the pattern can be adapted to use any format.

The Config class can also be used to wrap and process both variables in the configuration file and environment variables.

The Config class implements a separation of concerns by only processing top-level configuration variable options, and leaving more detailed configuration file parsing to the classes that need it. This allows for more flexible config files.

The Singleton Pattern

The Singleton pattern involves the use of instance variables, which the variables _CONFIG_FILE and _CONFIG are. These are shared across all instances of class Config and can be accessed via Config._CONFIG_FILE, etc.

The location of the config file can be set in the constructor, or can be provided via the CONFIG_FILE environment variable. The config class also provides a method for accessing environment variables that are required by the Config class, and raising a custom exception if it is not present.

The constructor starts by checking that the configuration file exists, then loads the configuration file into memory (at Config._CONFIG as a dictionary):

class Config(object):

    #########################
    # Begin Singleton Section
    #########################

    _CONFIG_FILE: typing.Optional[str] = None
    _CONFIG: typing.Optional[dict] = None

    def __init__(self, config_file = None):
        if config_file is None:
            config_file = Config.get_required_env_var("CONFIG_FILE")

        # Check that specified config file exists
        assert os.path.exists(config_file)

        # Use singleton pattern to store config file location/load config once
        Config._CONFIG_FILE = config_file
        with open(config_file, 'r') as f:
            Config._CONFIG = json.load(f)

    @staticmethod
    def get_config_file() -> str:
        return Config._CONFIG_FILE

    @staticmethod
    def get_required_env_var(envvar: str) -> str:
        if envvar not in os.environ:
            raise ConfigException(f"Please set the {envvar} environment variable")
        return os.environ[envvar]

Aside from the constructor, every method in the Config class is a @staticmethod or a @classmethod.

Get Variable Functions

We add two additional methods to get configuration variables: one to get variables from the config file, one to get environment variables. Here they are:

class Config(object):

    ...

    @staticmethod
    def get_required_env_var(envvar: str) -> str:
        if envvar not in os.environ:
            raise Exception("Please set the {envvar} environment variable")
        return os.environ[envvar]

    @staticmethod
    def get_required_config_var(configvar: str) -> str:
        assert Config._CONFIG
        if configvar not in Config._CONFIG:
            raise Exception(f"Please set the {configvar} variable in the config file {Config._CONFIG_FILE}")
        return Config._CONFIG[configvar]

We saw the get_required_env_var()function in action in the constructor. Theget_required_config_var()` can be useful for config variables that are dependent on other config variables.

Config Functions

Continuing with the Config class defined above, we now define methods that implement logic for specific configuration variables.

Here are two example config variables.

The variable foo is set using the configuration file. The configuration file is a dictionary, meaning it consists of key-value pairs, so the variable foo is set by the value corresponding to the foo key in the config file.

For example, using the following simple configuration file:

{
    "foo": "hello world"
}

if the Config class is initialized with that configuration file, Config.get_foo_var() will return the string hello world.

Similarly, the barvariable is set using the environment variable BAR. If the BAR variable is not set, the program will raise an exception when Config.get_bar_var() is called.

class Config(object):

    ...see singleton section above...

    #############################
    # Begin Configuration Section
    #############################

    _FOO: typing.Optional[str] = None
    _BAR: typing.Optional[str] = None

    @classmethod
    def get_foo_var(cls) -> str:
        """Example variable that is set in the config file (preferred)"""
        if cls._FOO is None:
            cls._FOO = Config.get_required_config_var('foo')
        return cls._FOO

    @classmethod
    def get_bar_var(cls) -> str:
        """Example variable that is set via env var (not preferred)"""
        if cls._BAR is None:
            cls._BAR = Config.get_required_env_var('BAR')
        return cls._BAR

    @classmethod
    def get_wuz(cls) -> str:
        if cls._WUZ is None:
            if 'wuz' not in cls._CONFIG:
                cls._WUZ = Config.get_required_env_var('WUZ')
            else:
                cls._WUZ = cls._CONFIG['wuz']
        if not os.path.isdir(cls._WUZ):
            raise Exception(f"Error: Path {cls._WUZ} is not a directory")
        return cls._WUZ

The wuz variable, in this example, is a variable that can be set with a config file variable, or (if it is not present in the config file) with an environment variable. The wuz variable msut also be a path, so there is logic for checking whether the path exists.

Reset method

It can be useful to clear out an existing config file in order to load a new config file - specifically, when testing. Here we define a reset() method that clears out variable values. We will show an example of how to use the reset() method below.

    @classmethod
    def reset(cls) -> None:
        cls._CONFIG_FILE = None
        cls._CONFIG = None
        cls._FOO = None
        cls._BAR = None
        cls._WUZ = None

This could be done more gracefully by iterating over each attribute of the Config class and only nullifying those attributes whose variable name matches the given pattern (start with an underscore, only contain capital letters and underscores) using a regular expression.

Creating a configuration context manager

To make tests more convenient, we define a context manager that takes a dictionary as an input. The context manager creates a temporary file with the contents of that dictionary, and resets the Config class using the temporary file as the new config file. This allows tests to be written using different configurations on the fly, very useful when testing different configuration options:

cfg = {"peanut_butter": "jelly"}
with TempConfig(cfg) as config_file:
    print(f"Temporary configuration file is at {config_file}")
    val = Config.get_required_config_var("peanut_butter")
    assert val=="jelly"

Here is the context manager class to temporarily replace the configuration wrapped by the Config class:

class TempConfig(object):
    """
    Temporarily patch the Config class to use the config
    dictionary specified in the constructor.
    """

    def __init__(self, config_dict, *args, **kwargs):
        """This is the step that's run when object constructed"""
        super().__init__()
        # This is the temp configuration the user specified
        self.config_dict = config_dict
        # Make a temp dir for our temp config file
        self.temp_dir = tempfile.mkdtemp()
        # Make a temp config file
        _, self.temp_json = tempfile.mkstemp(suffix=".json", dir=self.temp_dir)
        # Set the wuz variable to the temporary directory
        self.config_dict['wuz'] = self.temp_dir

    def __enter__(self, *args, **kwargs):
        """This is what's returned to the "as X" portion of the context manager"""
        self._write_config(self.temp_json, json.dumps(self.config_dict))
        # Re-init Config with new config file
        Config(self.temp_json)
        return self.temp_json

    def __exit__(self, *args, **kwargs):
        """
        Close the context and clean up; the *args are needed in case there is
        an exception (we don't deal with those here)
        """
        # Delete temp file
        os.unlink(self.temp_json)
        # Delete temp dir
        shutil.rmtree(self.temp_dir)
        # Reset all config variables
        Config.reset()

    def _write_config(self, target: str, contents: str):
        """Utility method: write string contents to config file"""
        with open(target, "w") as f:
            f.write(contents)

Next steps

That's it for now. This singleton configuration class is being written into a new version of centillion, which will be centillion version 2.0. This is still a pull request in a centillion fork, though, so it's a work in progress. Stay tuned!

Tags:    python    programming    patterns    design patterns    registry    computer science   

Using Mock API Servers

Posted in Python

permalink

Summary

In a prior post, we covered how to write a mock API server that stored a thread as a class attribute and used it to run the server in the background by starting a thread.

However, we neglected to cover how to actually use the mock API server. So here we include some examples of how you can use the mock API server to write better tests for components that require interacting with APIs.

The MockAPIServer Class

Let's start with a recap of the mock API server class. Major features included:

  • Inheriting from the base HTTP server class in Python, to take advantage of the methods available through it

  • Using a singleton design pattern to start and stop the fake API server

Basically we create the server, call start_serving(), and that starts the server on a thread in the background.

Here is the source code:

class MockAPIServer(BaseHTTPRequestHandler):
    _server = None
    _thread = None

    @staticmethod
    def get_addr_port():
        addr = "127.0.0.1"
        port = "9876"
        return addr, port

    @classmethod
    def start_serving(cls):
        # Get the bind address and port
        cls._addr, cls._port = cls.get_addr_port()

        # Create an HTTP server
        cls._server = HTTPServer((cls._addr, cls._port), cls)

        # Create a thread to run the server
        cls._thread = threading.Thread(target=cls._server.serve_forever)

        # Start the server
        cls._thread.start()

    @classmethod
    def stop_serving(cls):
        # Shut down the server
        if cls._server is not None:
            cls._server.shutdown()

        # Let the thread rejoin the worker pool
        cls._thread.join(timeout=10)
        assert not cls._thread.is_alive()

    def do_POST(self):
        ctype, pdict = cgi.parse_header(self.headers.get("content-type"))
        # Enforce rule: JSON only
        if ctype != "application/json":
            self.send_response(400)
            self.end_headers()
            return
        # Convert received JSON to dict
        length = int(self.headers.get("content-length"))
        message = json.loads(self.rfile.read(length))

        # Process the json

        # Send a response
        response = bytes(json.dumps(message), "utf8")
        self._set_headers()
        self.wfile.write(response)

    def _set_headers(self):
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()

A Basic Unit Test with MockAPIServer

Let's make a basic test that uses the MockAPIServer class. We'll use unittest for simplicity, other testing frameworks offer similar functionality.

Before testing our code, we'll need to make sure the API URL is configurable, sicne we will need to get the mock API server's bind address and port and use those to instruct our code where to find the API server.

Here is a short example function that we'll test:

foobar.py:

import urllib.parse

def make_api_call(api_url)
    """
    A simple function that gets an API endpoint
    and returns if no problems raised.
    """
    # Assemble our API call
    endpoint = '/hello/world'
    url = urllib.parse.urljoin(api_url, endpoint)
    params = dict(message='hello world')

    # The basic mock server will just echo our request back
    data = requests.get(url, params=params)
    data = resp.json()
    return

Now we write a short test for our foobar script:

test_foobar.py:

from foobar import make_api_call
import unittest

class TestAPICalls(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.app = MockAPIServer()
        cls.app.start_serving()

    def test_api_call():
        addr, port = self.app
        api_url = urllib.parse.urljoin(
            'http://', addr, port
        )
        make_api_cal(api_url)

    @classmethod
    def tearDownClass(cls):
        cls.app.stop_serving()

We can run the test like so:

python test_foobar.py

Here's what happens when we run the test:

  • The setUpClass() method is called, which creates a mock API server that starts on localhost on port 9876 and runs the server on a thread.
  • The test_api_call() method is run, which makes the API call to the mock API server. (Nothing interesting is happening on either end, right now, but stay tuned for more examples.)
  • The tearDownClass() method is called, which stops the mock API server and returns the thread worker to the pool of workers.

Stay tuned for more complicated examples in the future - we are currently working on extending this mock API server to mock calls to the Github API.

Tags:    http    server    python    mock    mocking    api    flask    web server   

Creating Mock API Servers

Posted in Python

permalink

Overview

In this post we discuss a way of mocking an API server during tests. This technique will let you create a fake API server that can respond to API calls however you want.

The technique is twofold:

  • First, we create a mock API handler that extends BaseHTTPRequestHandler, which is the built-in HTTP server class in Python. We can extend the server class to control how it responds to requests - to implement a method to respond to POST requests, we implement a do_POST() method, to respond to GET requests we implement a do_GET() method, and so on. (In the example below, we restrict the types of requests to JSON content only.)

  • Second, we use the Singleton design pattern, by implementing two class methods, start_serving() and stop_serving(), that we can call before and after our tests to set up and tear down the fake API server. This method will take care of starting the HTTP server on a separate thread, so that it does not block execution.

Mock API Server Class

Let's start with the mock server class. This is going to extend the BaseHTTPRequestHandler class from the http.server module, and extend it.

We implement a stub method for the POST response behavior; this is the only type of request that our mock API server will respond to.

We also have two stub class methods to start and stop the server.

class MockAPIServer(BaseHTTPRequestHandler):
    _server = None
    _thread = None

    def do_POST(self):
        pass

    @classmethod
    def start_serving(cls):
        pass

    @classmethod
    def stop_serving(cls):
        pass

Start/Stop Serving

We start with the two class methods to start and stop the server.

Getting Bind Address/Port

Define another static method to get the address to bind to, and the port to use; in this case we'll hard code values, but this function could also find unused networking ports, etc.

    @staticmethod
    def get_addr_port():
        addr = "127.0.0.1"
        port = "9876"
        return addr, port

Start Serving

Next, the start_serving() method should start a thread (using the cls._thread attribute to store it for later) and create an underlying HTTP server (and using the self._server attribute to store it for later):

    @classmethod
    def start_serving(cls):
        # Get the bind address and port
        cls._addr, cls._port = cls.get_addr_port()

        # Create an HTTP server
        cls._server = HTTPServer((cls._addr, cls._port), cls)

        # Create a thread to run the server
        cls._thread = threading.Thread(target=cls._server.serve_forever)

        # Start the server
        cls._thread.start()

Stop Serving

The stop_serving() method stops the thread

    @classmethod
    def stop_serving(cls):
        # Shut down the server
        if cls._server is not None:
            cls._server.shutdown()

        # Let the thread rejoin the worker pool
        cls._thread.join(timeout=10)
        assert not cls._thread.is_alive()

Handling Requests

The mock API server should only process POST requests, and should only accept JSON-formatted requests. We can implement those checks and have the server return a 500 error if clients do not send a properly formatted JSON request.

Defining POST Response Method

To define a response to POST requests made to the API we are mocking, we start by validating the JSON request that is received.

Note: this utilizes several built-in methods of the HTTP server class.

    def do_POST(self):
        ctype, pdict = cgi.parse_header(self.headers.get("content-type"))
        # Enforce rule: JSON only
        if ctype != "application/json":
            self.send_response(400)
            self.end_headers()
            return
        # Convert received JSON to dict
        length = int(self.headers.get("content-length"))
        message = json.loads(self.rfile.read(length))

        # Process the json
        ...

Now, the JSON can be processed using a validate function, for example, or generic success/failure responses returned based on the contents of a request.

Let's do something very simple: have the API server return whatever was sent in the request.

We can turn the dictionary message (a dictionary containing the original request) back into a string, and the string into a stream of bytes. Then we can write headers and the stream of bytes into the response.

        # Send a response
        response = bytes(json.dumps(message), "utf8")
        self._set_headers()
        self.wfile.write(response)

The _set_headers() method is a short method that just sends (writes) the correct headers:

    def _set_headers(self):
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()

Note: the send_headers() and end_headers() methods are built-in to the HTTP server base class we are using.

Putting it all together

Putting it all together, we get one final mock API server class:

class MockAPIServer(BaseHTTPRequestHandler):
    _server = None
    _thread = None

    @staticmethod
    def get_addr_port():
        addr = "127.0.0.1"
        port = "9876"
        return addr, port

    @classmethod
    def start_serving(cls):
        # Get the bind address and port
        cls._addr, cls._port = cls.get_addr_port()

        # Create an HTTP server
        cls._server = HTTPServer((cls._addr, cls._port), cls)

        # Create a thread to run the server
        cls._thread = threading.Thread(target=cls._server.serve_forever)

        # Start the server
        cls._thread.start()

    @classmethod
    def stop_serving(cls):
        # Shut down the server
        if cls._server is not None:
            cls._server.shutdown()

        # Let the thread rejoin the worker pool
        cls._thread.join(timeout=10)
        assert not cls._thread.is_alive()

    def do_POST(self):
        ctype, pdict = cgi.parse_header(self.headers.get("content-type"))
        # Enforce rule: JSON only
        if ctype != "application/json":
            self.send_response(400)
            self.end_headers()
            return
        # Convert received JSON to dict
        length = int(self.headers.get("content-length"))
        message = json.loads(self.rfile.read(length))

        # Process the json

        # Send a response
        response = bytes(json.dumps(message), "utf8")
        self._set_headers()
        self.wfile.write(response)

    def _set_headers(self):
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()

Tags:    http    server    python    mock    mocking    api    flask    web server   

March 2022

How to Read Ulysses

July 2020

Applied Gitflow

September 2019

Mocking AWS in Unit Tests

May 2018

Current Projects