Category: Programming

I already explained what is SOLID Principles in a previous article and here are examples in Python for each of the SOLID principles to make it more clear:

Single Responsibility Principle (SRP)

The SRP states that a class should have only one reason to change. In other words, a class should have only one responsibility. Here’s an example of a class that violates the SRP:

class Employee:
    def calculate_pay(self):
        # Calculate the employee's pay
        pass

    def save_to_database(self):
        # Save the employee's data to the database
        pass

In this example, the Employee class has two responsibilities: calculating the employee’s pay and saving the employee’s data to the database. To follow the SRP, we could split this class into two separate classes:

class Employee:
    def calculate_pay(self):
        # Calculate the employee's pay
        pass

class EmployeeRepository:
    def save_to_database(self, employee):
        # Save the employee's data to the database
        pass

Now we have one class responsible for calculating the employee’s pay and another class responsible for saving the employee’s data to the database.

Open/Closed Principle (OCP)

The OCP states that classes should be open for extension but closed for modification. In other words, you should be able to add new functionality to a class without changing its existing code. Here’s an example of a class that violates the OCP:

class Employee:
    def __init__(self, name, salary):
        self.name = name
        self.salary = salary

    def calculate_bonus(self):
        return self.salary * 0.1

In this example, the Employee class calculates a bonus based on the employee’s salary. If we want to calculate a bonus based on other factors, such as the employee’s performance or the company’s revenue, we would need to modify the Employee class. To follow the OCP, we could use inheritance to create a new class for calculating bonuses:

class BonusCalculator:
    def calculate_bonus(self, employee):
        pass

class SalaryBonusCalculator(BonusCalculator):
    def calculate_bonus(self, employee):
        return employee.salary * 0.1

class PerformanceBonusCalculator(BonusCalculator):
    def calculate_bonus(self, employee):
        # Calculate bonus based on performance
        pass

class RevenueBonusCalculator(BonusCalculator):
    def calculate_bonus(self, employee):
        # Calculate bonus based on revenue
        pass

Now we have one class responsible for calculating bonuses and multiple subclasses that can calculate bonuses based on different criteria.

Liskov Substitution Principle (LSP)

The LSP states that objects of a superclass should be replaceable with objects of a subclass without changing the correctness of the program. In other words, subclasses should be able to be used in place of their parent classes without causing unexpected behavior. Here’s an example of a class hierarchy that violates the LSP:

class Bird:
    def fly(self):
        pass

class Ostrich(Bird):
    def fly(self):
        raise NotImplementedError()

In this example, the Ostrich class is a subclass of Bird, but it cannot fly. If we try to use an Ostrich object in place of a Bird object, we will get an unexpected NotImplementedError. To follow the LSP, we could split the Bird class into two separate classes:

class Bird:
    def fly(self):
        pass

class FlyingBird(Bird):
    def fly(self):
        pass

class Ostrich(Bird):
    pass

Now we have one class for birds that can fly and another class for birds that cannot fly.

Interface Segregation Principle (ISP)

The ISP states that clients should not be forced to depend on methods they do not use. In other words, interfaces should be tailored to the needs of their clients. Here’s an example of a class that violates the ISP:

class Printer:
    def print(self, document):
        pass

    def scan(self, document):
        pass

In this example, the Printer class provides both printing and scanning functionality. If a client only needs printing functionality, they will be forced to depend on the scan method. To follow the ISP, we could split the Printer class into two separate interfaces:

class Printer:
    def print(self, document):
        pass

class Scanner:
    def scan(self, document):
        pass

Now we have two separate interfaces, one for printing and one for scanning, that clients can use independently.

Dependency Inversion Principle (DIP)

The DIP states that high-level modules should not depend on low-level modules. Instead, both should depend on abstractions. In addition, abstractions should not depend on details. Instead, details should depend on abstractions. Here’s an example of a class that violates the DIP:

class EmployeeService:
    def __init__(self, employee_repository):
        self.employee_repository = employee_repository

    def get_employee_by_id(self, employee_id):
        return self.employee_repository.get_by_id(employee_id)

In this example, the EmployeeService class depends on the EmployeeRepository class, which is a low-level module. To follow the DIP, we could introduce an abstraction that both the EmployeeService and EmployeeRepository classes depend on:

class EmployeeRepository:
    def get_by_id(self, employee_id):
        pass

class EmployeeService:
    def __init__(self, employee_repository):
        self.employee_repository = employee_repository

    def get_employee_by_id(self, employee_id):
        return self.employee_repository.get_by_id(employee_id)

class SqlEmployeeRepository(EmployeeRepository):
    def get_by_id(self, employee_id):
        # Retrieve employee from SQL database
        pass

class MongoEmployeeRepository(EmployeeRepository):
    def get_by_id(self, employee_id):
        # Retrieve employee from MongoDB
        pass

Now we have an abstraction, the EmployeeRepository class, that both the EmployeeService and SqlEmployeeRepository and MongoEmployeeRepository classes depend on. This allows us to easily switch between different database implementations without changing the EmployeeService class.

These examples show how each of the SOLID principles can be applied in Python code to create more maintainable, flexible, and reusable software.

Conclusion

The SOLID principles provide a set of guidelines for writing clean and maintainable code. By following these principles, we can write code that is more modular, flexible, and reusable, making it easier to maintain and extend over time.

In Python, we can apply the SOLID principles in a variety of ways, such as using inheritance and polymorphism to adhere to the LSP, creating small and focused classes that adhere to the SRP, using dependency injection to adhere to the DIP, and splitting interfaces to adhere to the ISP. By applying these principles, we can write code that is easier to understand, modify, and extend, ultimately leading to more robust and maintainable software.

It’s worth noting that while the SOLID principles are generally considered to be good coding practices, they are not always applicable in every situation. As with any programming guideline or best practice, it’s important to use good judgement and apply the principles in a way that makes sense for the specific requirements and constraints of your project.

In conclusion, the SOLID principles are an important set of guidelines for writing clean and maintainable code. By following these principles in our Python code, we can create software that is more flexible, reusable, and easier to maintain and extend over time.

This is a step by step tutorial for beginners on how to write a python command line tool to fetch data from a remote API. This will teach you how to write clean, maintainable, testable and easy to use code in python.

The point of this tutorial is mainly to learn how to structure the code in a way that it’s easy to extend in the future and also easy to understand and test.

The requirements

Implement a CLI that fetches public holidays from the existing API, https://date.nager.at, and shows the next n (5 by default) occurring holidays.

• In order to avoid fetching the data too frequently, the endpoint shouldn’t be called more than once a day. To achieve that, we need to implement caching. There are many different ways to do that. I chose to use Redis because it’s really powerful, fast, reliable and widely used in many applications.
• The country code should be passed as a cli argument.
• The output should contain the following information: Date, name, counties, types.

Building CLI

Argparse is the default python module for creating command line tools. It provides all the features you need to build a simple CLI. It can work for our example but i find it a little bit complicated and not very pythonic that’s why i chose another one called click. It’s a Python package for creating beautiful command line interfaces in a composable way with as little code as necessary. It uses decorators to define commands.

You can install it using pip:

$ python -m pip install -U click

To define a command using click, the command needs to be a function wrapped with  the decorator @click.command():

import click


@click.command()
def main():
    click.echo("It works.")


if __name__ == "__main__":
    main()

initial implementation of public holidays CLI

We need to add two arguments to this command, the country code which is a string and the number of upcoming vacation days to display, which is an integer and the default value is five. to do this, we need to use the decorator @click.option as follows:

import click


@click.command()
@click.option(
    "--country-code",
    prompt="Country Code",
    help="Country code. complete list is here https://date.nager.at/Country."
)
@click.option(
    "--max-num",
    prompt="Max num of holidays returned",
    help="Max num of holidays returned.",
    default=5,
)
def main(country_code, max_num=5):
    """
    Simple CLI for getting the Public holidays of a country by country code.
    """
    click.echo(f"country_code: {country_code}, max_num: {max_num}")
    click.echo("It works.")


if __name__ == "__main__":
    main()

Now our command is ready. Let’s start with the second step, using docker.

Using docker

Docker is a software platform that allows you to build, test, and deploy applications quickly. Docker packages software into containers that have everything the software needs to run. Using Docker helps deploy and scale applications into any environment and be sure that your code will run without any installation problem.

To dockerize our command we need to create a Dockerfile. It’s a text document containing the instructions that docker should follow to build the docker image (using docker build) which is used to run the docker container later.

here is our Dockerfile:

# use python3.9 base image
FROM python:3.9

# defining the working directory
WORKDIR /code

# copy the requirements.txt from the local directory to the docker image
ADD ./requirements.txt /code/requirements.txt

# upgrade pip setuptools
RUN pip install --upgrade setuptools wheel

# install all dependencies from requirements.txt file
RUN pip install -r requirements.txt

Dockerfile

Using docker-compose

Docker compose is a tool that uses yaml files to define multiple container applications (like this one). Docker compose can be used in all environments but it’s not recommended to use it in production. It’s recommended to use docker swarm which is very similar to docker-compose or even better use kubernetes.

We need to run multiple containers because we need to have a caching layer in our command line tool to cache the holidays for one day. we will use Redis for that but let’s first start by setting up the redis container and make it available to our command line tool.

Here is the docker-compose.yaml file for redis. it defines one service called redis and uses redis:alpine image which is available in docker hub

version: '3.7'

services:
  redis:
    image: "redis:alpine"

We also need to add our command line tool to the docker-compose file to be able to run the command and start the redis container with only one command. here is how can we do it

version: '3.7'

services:
  command:
    # build the local Dockerfile image and use it here
    build: .
    
    # command to start the container
    command: python main.py --country-code DE --max-num 5
    
    # volume to mount the code inside the docker-container
    volumes:
      - .:/code
    
    # this container should not start until redis container is up
    depends_on:
      - redis
  redis:
    image: "redis:alpine"

Here we added another service called command which uses the docker image from the docker file we already implemented previously and the command to start this container is just python main.py. it also depends on redis which means that the command container should not start until the redis container is up.

Now you should be able to run the command and start the redis server by just running:

docker-compose up

Configuring the cache

The next step is to configure the caching layer and start using it in our command line tool. The first step is to install redis using pip. also make sure to add it to the requirements.txt.

$ python -m pip install redis

Then create a new file called cache.py. This should contain the following code

from redis import StrictRedis


# settings from url redis://{host}:{port}
# the host is the name of redis service we defined in the docker-compose file
REDIS_URL = 'redis://redis:6379'


redis_conn = StrictRedis.from_url(REDIS_URL)

cache.py

This will create a connection to the redis server using the singleton design pattern because we need to make sure that we only have one connection and we are re-using this connection instead of creating a new connection each time we want to communicate to the redis server.

The next step is to create some helper functions to handle the common use cases of the cache like caching some data, retrieving some data from cache and also invalidating the cache if needed.

import json
from datetime import datetime
from typing import Optional

from redis import StrictRedis


# settings
REDIS_URL = 'redis://redis:6379'
LAST_UPDATED_AT_KEY_POSTFIX = '_last_updated_at'
DEFAULT_CACHE_IN_SECONDS = 86400  # 1 day


redis_conn = StrictRedis.from_url(REDIS_URL)


def get_data_from_cache(key: str) -> Optional[dict]:
    """
    retrieve data from cache for the given key
    """
    data = redis_conn.get(key)
    if data:
        return json.loads(data)


def save_data_to_cache(key: str, data: dict, expire_in_sec: int = DEFAULT_CACHE_IN_SECONDS) -> None:
    """
    Save data to cache
    """
    redis_conn.set(key, json.dumps(data), ex=expire_in_sec)
    redis_conn.set(key + LAST_UPDATED_AT_KEY_POSTFIX, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), ex=expire_in_sec)


def invalidate_cache_for_key(key: str) -> None:
    """
    invalidate cache for the given key
    """
    redis_conn.delete(key)
    redis_conn.delete(key + LAST_UPDATED_AT_KEY_POSTFIX)

cache.py

Now the caching layer is ready for us to use in the command line. As you can see, all we need to do to save some data to the cache is to import and use the save_data_to_cache function without worrying much about which caching solution we use or how to connect to it. Also, if we decided for some reason to change the caching backend to use memcache for example, all we need to do is to change the cache.py file and make sure our helper functions work. no need to change anything in the application.

Holidays Client

The next step is to implement the Holidays Client that will allow us to fetch holidays from an external API.

First we need to install requests. It’s a very common HTTP library. Requests allows you to send HTTP/1.1 requests extremely easily. There’s no need to manually add query strings to your URLs, or to form-encode your PUT & POST data.

To install it,

$ python -m pip install requests

Then create a new file called client.py. We will first implement the base class HolidaysClient that will define all attributes and methods we need to implement the client. It does not matter which backend we use to fetch the holidays, it will always use this class as a parent class and implement the get_holidays method.

import abc


class HolidaysClient(abc.ABC):
    """
    Abstract class to be used as a base class to any endpoint for getting the public holidays.
    """
    # Base url for the external endpoint
    BASE_URL = None

    def get_holidays(self, country_code: str, year: int) -> dict:
        """
        getting the holidays from external API by country code and year
        """
        raise NotImplemented()

Then we need to implement the NagerHolidaysClient that inherits from HolidaysClient and implement the get_holidays method using date.nager.at PublicHolidays API.

import abc

import requests


class HolidaysClient(abc.ABC):
    """
    Abstract class to be used as a base class to any endpoint for getting the public holidays.
    """
    # Base url for the external endpoint
    BASE_URL = None

    def get_holidays(self, country_code: str, year: int) -> dict:
        """
        getting the holidays from external API by country code and year
        """
        raise NotImplemented()


class NagerHolidaysClient(HolidaysClient):
    """
    Nager client to get holidays from date.nager.at
    """
    # base url of nager client
    BASE_URL = 'https://date.nager.at'

    def get_holidays(self, country_code: str, year: int) -> dict:
        """
        fetch holidays from date.nager.at using PublicHolidays API
        """
        url = f'{self.BASE_URL}/api/v3/PublicHolidays/{year}/{country_code}'
        response = requests.get(url)
        response.raise_for_status()
        response_data = response.json()
        return response_data

Now the client should be ready. The next step will be adding some usecases and utils to help us connecting everything together.

Implementing the core logic of Holidays CLI

First, We need to implement a function that will display the result to the console in a human readable way. It’s better to implement this functionality in a separate function because we expect the output format to be changed often and in the future we might implement some other ways to show the results like for example, write them to a file or send emails to customers with the upcoming holidays.

For now, We will keep it very simple. Just a simple print to the console will be enough. The code should be something like this:

from typing import List

import click


def show_results(results: List[dict]) -> None:
    """
    Given a list of objects, it will print these data to the console in a human-readable way. 
    """
    click.echo('result:')
    click.echo('----------------')
    for idx, item in enumerate(results):
        click.echo(f'{idx + 1}- {item}')
    click.echo('----------------')

The output should be something like this:

result:
----------------
1- {'date': '2022-08-15', 'name': 'Assumption Day', 'counties': ['DE-SL'], 'types': ['Public']}
2- {'date': '2022-09-20', 'name': "World Children's Day", 'counties': ['DE-TH'], 'types': ['Public']}
3- {'date': '2022-10-03', 'name': 'German Unity Day', 'counties': None, 'types': ['Public']}
| 4- {'date': '2022-10-31', 'name': 'Reformation Day', 'counties': ['DE-BB', 'DE-MV', 'DE-SN', 'DE-ST', 'DE-TH', 'DE-HB', 'DE-HH', 'DE-NI', 'DE-SH'], 'types': ['Public']}
5- {'date': '2022-11-01', 'name': "All Saints' Day", 'counties': ['DE-BW', 'DE-BY', 'DE-NW', 'DE-RP', 'DE-SL'], 'types': ['Public']}

Then, We will need to implement the caching logic. if there is no data available in the cache, we should call the HolidaysClient to fetch the data and save them to the cache. here is the code to do this:

from typing import List, Optional

import requests
from datetime import datetime

from cache import get_data_from_cache, save_data_to_cache
from client import NagerHolidaysClient


def get_next_occurring_holidays(data: List[dict], max_num: int = 5) -> List[dict]:
    """
    parse Holidays API response and get next n holidays
    :param data: Holidays API response
    :param max_num: number of holidays in the response
    :return: list of holidays
    """
    # get today's date
    today_date = datetime.now().date()
    
    # init the results
    result = []
    
    # for each holiday in the holiday api response
    for holiday in data:
        
        # break if we already reached the required number of holidays in the result
        if len(result) >= max_num:
            break
        
        # get the date of the current holiday
        holiday_date = datetime.strptime(holiday['date'], '%Y-%m-%d').date()
        
        # skip if the holiday date is in the past
        if today_date > holiday_date:
            continue
        
        # save the result
        result.append({
            'date': holiday['date'],
            'name': holiday['name'],
            'counties': holiday['counties'],
            'types': holiday['types'],
        })

    return result


def get_next_holidays_by_country_code(country_code, max_num=5, year=None) -> (Optional[str], Optional[List[dict]]):
    """
    given a country code and a year, it gets holidays from external API (or cache).
    :param country_code: 2 letters country code. case-insensitive
    :param max_num: number of holidays we want to get
    :param year: the year we want to get holidays for
    :return: error string if any error happens and list of results if there is no error
    """
    # caching key should be something like this `2022;DE`
    cache_key = f'{year};{country_code}'
    
    # check if the data is already cached
    data_from_cache = get_data_from_cache(cache_key)
    if data_from_cache:
        # if the data is in the cache then we don't need to call the external API
        print(f'Getting data from cache for country: {country_code} and year: {year}')
        result = get_next_occurring_holidays(data_from_cache, max_num)
        return None, result

    try:
        # getting the holidays from Nager Holidays API
        response_data = NagerHolidaysClient().get_holidays(country_code, year)
    except requests.exceptions.HTTPError:
        return 'HTTPError', None
    except requests.exceptions.JSONDecodeError:
        return 'JSONDecodeError', None

    print(f'saving data to cache for country: {country_code} and year: {year}')
    save_data_to_cache(cache_key, response_data)
    result = get_next_occurring_holidays(response_data, max_num)
    return None, result

Finally, We need to use this in the command main.py file so the final version of it should be like this

from datetime import datetime

import click

from usecases import get_next_holidays_by_country_code


__author__ = "Ramadan Khalifa"

from utils import show_results


@click.command()
@click.option(
    "--country-code",
    prompt="Country Code",
    help="Country code. complete list is here https://date.nager.at/Country."
)
@click.option(
    "--max-num",
    prompt="Max num of holidays returned",
    help="Max num of holidays returned.",
    default=5,
)
def main(country_code, max_num=5):
    """
    Simple CLI for getting the Public holidays of a country by country code.
    """
    # initialize the target year with the current year
    target_year = datetime.now().year
    results = []
    
    # loop until we reach our target number of holidays
    while len(results) < max_num:
        error, next_result = get_next_holidays_by_country_code(
            country_code, max_num=max_num - len(results), year=target_year
        )
        
        # show the error if there is any
        if error:
            click.echo(error)
            return
        
        # next
        results += next_result
        target_year += 1

    # print results to the user
    show_results(results)


if __name__ == "__main__":
    main()

The whole project is available in github.

Conclusion

This was a simple and straightforward project about holidays. We learned how to have independent modules for our project to help us extend our project in the future if needed. We also learned about docker, docker-compose and redis. We will dive deeper into those tools and will explore more in the upcoming tutorials. Please hit subscribe to get updates and new tutorials once they are available.