
Managing a monorepo with multiple Python packages presents unique challenges, particularly in ensuring efficient CI/CD. In this article, I will share how I created CI/CD workflows for our monorepo of Python packages using GitHub Actions.

We will delve into the workflows, including matrix complexity, and scripts for checking changed paths, alongside other custom scripts and composite actions that facilitate this process.

Our primary CI workflow (`on-pr-build-push.yaml`) triggers on pull requests, ensuring that all changes are validated before merging. This workflow comprises several jobs: `ChangedPaths`, `Lint`, `Typecheck`, `Test`, and `Build-Push`.

**Job: ChangedPaths**

This job determines which packages have been modified and outputs a matrix that drives the subsequent jobs. So for example, if a PR modifies `package1` and `package2`, the matrix will contain two rows: `package1` and `package2`.

```yaml
name: Build & Push Python Package

on:
  pull_request:

jobs:
  ChangedPaths:
    runs-on: ubuntu-latest
    outputs:
      matrix: ${{ steps.get_changed_paths.outputs.matrix }}
    timeout-minutes: 10
    permissions:
      id-token: write
      contents: read

    steps:
      - name: Checkout
        uses: actions/checkout@v3
        with:
          token: ${{ secrets.GLOBAL_GITHUB_TOKEN }}
          fetch-depth: 0

      - name: Run init
        uses: ./.github/actions/init # Custom composite action for initializing the environment

      - name: Get Changed Paths
        id: get_changed_paths
        shell: bash
        run: |
          python $GITHUB_WORKSPACE/.github/scripts/internal/get_changed_paths.py
```

The `get_changed_paths.py` python script identifies the changed paths and filters them to include only those within the packages directory.
This script is quite straightforward and uses the `git diff` command by using the built-in `subprocess` module in order to identify the changed paths.

```python
import subprocess
import os

output_file = os.getenv('GITHUB_OUTPUT')


def get_changed_paths():
    cmd = 'git diff --name-only $(git rev-parse HEAD) HEAD~1'
    result = subprocess.run(cmd, shell=True, capture_output=True)
    if result.returncode != 0:
        raise Exception("Error getting changed paths")

    changed_paths = result.stdout.decode("utf-8").strip().split("\n")
    print(f"Changed paths: {changed_paths}")
    return changed_paths


def get_filtered_paths(changed_paths):
    filtered_paths = set()
    for path in changed_paths:
        if path.startswith("packages/"):
            filtered_paths.add(path.split("/")[1])
    print(f"Filtered paths: {filtered_paths}")
    return list(filtered_paths)


if __name__ == "__main__":
    changed_paths = get_changed_paths()
    filtered_paths = get_filtered_paths(changed_paths)
    print(filtered_paths)
    with open(output_file, "a") as file:
        file.write(f"matrix={filtered_paths}")
```

Note the `capture_output=True` argument allows us to capture the command's output. We then decode and split the output to get a list of changed paths. The `get_filtered_paths` function filters these paths to include only those that start with `packages/` which is my path in my repos case.

Also note that GitHub Actions uses environment files to pass data between steps and jobs. By writing the output matrix to a file (GITHUB_OUTPUT), we can ensure that the list of changed paths is available as an output in the ChangedPaths job. This allows subsequent jobs to access and iterate over the paths efficiently.

**Job: Lint, Typecheck, and Test**

These jobs runs composite actions for linting (flake8), type checking (mypy), and tests (pytest) on the changed packages using a matrix strategy. They depend on the ChangedPaths job.

```yaml
Lint:
  needs: ChangedPaths
  if: ${{ needs.ChangedPaths.outputs.matrix != '[]' }}
  strategy:
    matrix:
      path: ${{ fromJson(needs.ChangedPaths.outputs.matrix) }}
  runs-on: ubuntu-latest
  timeout-minutes: 10
  permissions:
    id-token: write
    contents: read

  steps:
    - name: Checkout
      uses: actions/checkout@v3

    - name: Get path
      run: echo ${{ matrix.path }}

    - name: Run init
      uses: ./.github/actions/init
      with:
        change_path: ${{ matrix.path }}

    - name: Run Lint action
      uses: ./.github/actions/lint
      env:
        paths: ${{ matrix.path }}
```

**Job: Build-Push**

The final job in the CI pipeline builds and pushes the packages if all previous jobs succeed.

```yaml
Build-Push:
  needs: [ChangedPaths, Lint, Typecheck, Test]
  if: ${{ needs.ChangedPaths.outputs.matrix != '[]' }}
  strategy:
    matrix:
      path: ${{ fromJson(needs.ChangedPaths.outputs.matrix) }}
  runs-on: ubuntu-latest
  timeout-minutes: 15
  permissions:
    id-token: write
    contents: read

  steps:
    - name: Checkout
      uses: actions/checkout@v3
      id: checkout

    - name: Run init
      id: run-init
      uses: ./.github/actions/init
      with:
        change_path: ${{ matrix.path }}

    - name: Build & Publish Python Packages
      uses: ./.github/actions/build-push-package
      env:
        paths: ${{ matrix.path }}
```

This job calls a custom composite action (`build-push-package`) that builds and upload the package to GitHub cache, so the user can install it from there.

```yaml
name: Build & Release Package
description: 'Build & Push Python Package'
author: 'Or Kazaz'

runs:
  using: composite

  steps:
    - name: Checkout
      uses: actions/checkout@v3
      with:
        fetch-depth: 0

    - name: Setup Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'

    - name: Install Python Dependencies
      shell: bash
      run: |
        pip install --upgrade pip
        pip install poetry twine check-wheel-contents

    - name: Run Poetry Check for pyproject.toml
      working-directory: ./packages/${{ env.paths }}
      shell: bash
      run: poetry check

    - name: Build Package Distribution
      working-directory: ./packages/${{ env.paths }}
      shell: bash
      run: poetry build

    - name: Check Distribution Descriptions
      working-directory: ./packages/${{ env.paths }}
      shell: bash
      run: |
        twine check dist/*

    - name: Check Wheel Contents
      working-directory: ./packages/${{ env.paths }}
      shell: bash
      run: |
        check-wheel-contents dist/*.whl

    - name: Cache Package Distribution
      uses: actions/upload-artifact@v3
      with:
        name: ${{ env.paths }}
        path: ./packages/${{ env.paths }}/dist/
```

The secondary workflow (`tag-and-release.yaml`) is triggered on a manually on demand by using `workflow_dispatch`. It tags the package, updates its version is pyproject.toml.toml, and publish the packages into our internal PyPI repository (on Jfrog Artifactory). It consists of multiple jobs to ensure a smooth release process.

**Job: Tag and Release**

This job tags the repository and creates a release on GitHub by zipping it and using an action to create a release to be available for download.

```yaml
jobs:
  tag-and-release:
    runs-on: ubuntu-latest
    timeout-minutes: 15

    outputs:
      package_version: ${{ steps.dynamic_version.outputs.package_version }}

    steps:
      - name: Checkout
        uses: actions/checkout@v3
        with:
          ref: ${{ github.event.inputs.commit_hash }}

      - name: Run Dynamic-Versioning
        id: dynamic_version
        uses: ./.github/actions/dynamic-versioning
        with:
          package_name: ${{ github.event.inputs.package_name }}

      - name: Archive package directory
        working-directory: ./packages
        run: |
          zip -r ${{ github.event.inputs.package_name }}.zip ./${{ github.event.inputs.package_name }}
          echo "ARTIFACT_FILE_PATH=./packages/${{ github.event.inputs.package_name }}.zip" >> $GITHUB_ENV
        env:
          ARTIFACT_FILE_PATH: ${{ env.ARTIFACT_FILE_PATH }}

      - name: Create Release
        uses: ncipollo/release-action@v1
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        with:
          artifacts: '${{ env.ARTIFACT_FILE_PATH }}'
          commit: ${{ github.event.inputs.commit_hash }}
          tag: ${{ github.event.inputs.package_name }}-v${{ steps.dynamic_version.outputs.package_version }}
          name: ${{ github.event.inputs.package_name }}-v${{ steps.dynamic_version.outputs.package_version }}
          token: ${{ secrets.GLOBAL_GITHUB_TOKEN }}
          skipIfReleaseExists: true
```

Note that the `dynamic-versioning` action is a custom action that updates the version in the `pyproject.toml` file.

```yaml
name: 'Dynamic Versioning'
description: 'Bump version in python package'
author: 'Or Kazaz'

inputs:
  package_name:
    description: 'Package name'
    required: true

outputs:
  package_version:
    description: 'bumped package version'
    value: ${{ steps.bump_version.outputs.package_version }}

runs:
  using: composite

  steps:
    - name: Checkout
      uses: actions/checkout@v3
      with:
        fetch-depth: 0

    - name: Bump version
      id: bump_version
      shell: bash
      run: |
        PACKAGE_VERSION=$($GITHUB_WORKSPACE/.github/scripts/internal/bump_version.sh ${{ inputs.package_name }})
        echo "${{ inputs.package_name }} package will be bumped to version: ${PACKAGE_VERSION}"
        echo "package_version=$PACKAGE_VERSION" >> $GITHUB_OUTPUT
```

Bump version is a very simple bash script the increments the package version in the `pyproject.toml` file.

```bash
#!/bin/bash

PACKAGE_NAME=$1

VERSION_LINE=$(grep version ./packages/$PACKAGE_NAME/pyproject.toml)
OLD_VERSION=$(echo "$VERSION_LINE" | awk -F'"' '{print $2}')
NEW_VERSION=$(echo "$OLD_VERSION" | awk -F. -v OFS=. '{$3=$3+1}1')

sed -i "s/$VERSION_LINE/version = \"$NEW_VERSION\"/g" ./packages/$PACKAGE_NAME/pyproject.toml

VERSION=$(grep version ./packages/$PACKAGE_NAME/pyproject.toml | awk -F'\"' '{print $2}')
echo $VERSION
```

**Job: Build**

This job builds the package distribution using Poetry.

```yaml
build:
  needs: tag-and-release
  name: Build Python Package
  runs-on: ubuntu-latest
  timeout-minutes: 30

  steps:
    - name: Checkout
      uses: actions/checkout@v3
      with:
        ref: ${{ github.event.inputs.commit_hash }}

    - name: Setup Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'

    - name: Install Python Dependencies
      working-directory: ./packages/${{ github.event.inputs.package_name }}
      run: |
        pip install --upgrade pip
        pip install --upgrade poetry twine check-wheel-contents

    - name: Run Dynamic-Versioning
      id: dynamic_version
      uses: ./.github/actions/dynamic-versioning
      with:
        package_name: ${{ github.event.inputs.package_name }}

    - name: Build Package Distribution
      working-directory: ./packages/${{ github.event.inputs.package_name }}
      run: |
        poetry build
```

**Job: Publish to JFrog**

This job publishes the built package to JFrog by retrieving the cached package distribution and using the official `gh-action-pypi-publish` action.

```yaml
publish-jfrog:
  needs: build
  name: Publish Package to JFrog
  runs-on: ubuntu-latest
  timeout-minutes: 15

  steps:
    - name: Retrieve Cached Package Distribution
      uses: actions/download-artifact@v3
      with:
        name: ${{ github.event.inputs.package_name }}
        path: ./packages/${{ github.event.inputs.package_name }}/dist

    - name: Publish to JFrog
      uses: pypa/gh-action-pypi-publish@release/v1
      with:
        packages-dir: ./packages/${{ github.event.inputs.package_name }}/dist
        user: ${{ secret.secret_user }}
        password: ${{ secret.secret_password }}
        repository-url: https://the.internal.repo.url/artifactory/api/pypi/pypi
```

**Job: Commit Updated Version**

Finally, this job commits the updated version to the repository while skipping CI.

```yaml
commit-version:
  needs: [tag-and-release, build, publish-jfrog]
  name: Commit updated version
  runs-on: ubuntu-latest
  timeout-minutes: 15

  steps:
    - name: Checkout
      uses: actions/checkout@v3

    - name: Commit updated version
      shell: bash
      run: |
        git config --global user.name $GITHUB_USER
        git config --global user.email $GITHUB_EMAIL
        git pull
        bash $GITHUB_WORKSPACE/.github/scripts/internal/bump_version.sh ${{ github.event.inputs.package_name }}
        git add ./packages/${{ github.event.inputs.package_name }}/pyproject.toml
        git commit -m "Bump ${{ github.event.inputs.package_name }} to version ${{ needs.tag-and-release.outputs.package_version }} [skip ci]"
        git push origin ${GITHUB_REF#refs/heads/}
```

![CI-ON-PR](https://res.cloudinary.com/dtt2usukl/image/upload/v1718703406/Image_18-06-2024_at_12.36_ge6oku.jpg)_On Pull request CI of 3 packages modified_

This setup ensures that only the affected packages are processed, reducing build times and resource usage. The use of matrix strategies, custom scripts, and composite actions provides a flexible, generic and scalable approach to managing CI/CD workflows of such use case.


In this article i'll share how I handheld the CI/CD in a mono repo with multiple python packages using GitHub Actions.

Handling CI/CD in a Mono Repo With Multiple Python Packages


We've implemented extensive improvements, including expanding device capacity, enhancing monitoring systems, introducing features like automated recovery mechanisms, chaging regions using internal network VPN and more.
These advancements will streamline testing processes, optimize resource utilization, and provide a seamless experience for developers and testers across global locations.
With ~250 users, supporting 25 different platforms and 150 testing devices, we are well-positioned for continued success.

For those of you who are not familiar with DAZN, here's a brief overview:

> _DAZN is the world’s leading live sports BTC streaming service. With DAZN, fans in any country can watch sport in any language, on any device, live and on-demand. Purpose built to stream live sports globally, DAZN is the market leader in Italy, Spain, Germany, and Japan, where it holds the top-tier domestic football rights — Serie A, La Liga, Bundesliga and J League, respectively — as well as the richest catalogue of live and on-demand content. DAZN is also available in over 200 other territories, where it boasts the most extensive library of elite and emerging sports content._

DAZN develops both HTML-based and native applications designed to run on a variety of platforms, including TVs, set-top boxes, gaming consoles, and native devices. Ensuring high quality across these diverse platforms necessitates thorough testing at every stage of development, ranging from unit tests and component tests to emulator tests and real-device testing.

The global pandemic in 2020 prompted a paradigm shift in work practices, underscoring the importance of enabling engineers to access diverse resources and conduct testing remotely. Quality remains paramount, necessitating comprehensive testing methodologies despite remote work environments.

To address these challenges, we adopted a proactive “Lab as a Service” approach, culminating in the establishment of an in-house lab facility in Katowice, Poland. This strategic decision aimed to centralize resources and provide engineers with the necessary tools and infrastructure to conduct testing efficiently.

However, establishing and operating an in-house lab posed unique challenges that required careful consideration and resolution. These challenges encompassed logistical, technical, and organizational aspects, underscoring the need for adaptive solutions and collaborative efforts.

In response to these challenges, ongoing initiatives are focused on optimizing the efficiency and effectiveness of the in-house lab, streamlining workflows, enhancing collaboration among teams, and leveraging technology to overcome geographical barriers. By embracing innovation and agility, DAZN aims to continuously improve its testing capabilities and deliver exceptional quality across its diverse range of applications and platforms.

## Challenges

DAZN employs a diverse team of developers and test engineers spread across different global locations such as Poland, India, Israel, and the UK. Each team comes with its unique set of needs and requirements, necessitating a robust, highly available, and cost-effective platform to accommodate them effectively.

Key points include:

- Diverse Platform Support:
  The platform must ensure comprehensive testing across all platforms supported by DAZN, emphasizing thoroughness and quality assurance in our testing procedures.
- Tailored Solutions:
  Teams may require different tools and resources for their testing processes; for instance, some teams may rely on manual UI/UX testing devices, while others prioritize automation solutions.
- Integration Challenges:
  Incorporating in-house tools and services into our platform is essential to ensure seamless continuity for engineers, requiring careful integration efforts.
- Scalability and Maintenance:
  The platform needs to support a large number of engineers on a daily basis across multiple global sites, balancing scalability with efficient maintenance to avoid operational bottlenecks.

- Cost Considerations:
  Maintaining devices, infrastructure, and manpower across various locations can incur significant expenses, necessitating a cost-effective approach to resource allocation and management.

- Ownership and Accountability:
  Clear ownership and accountability structures are crucial to prevent issues such as maintenance gaps or device mismanagement.

To address these challenges, ongoing efforts are focused on streamlining processes, enhancing collaboration between teams, and leveraging automation wherever possible to optimize resource utilization and improve efficiency in our testing and development workflows.

## Solution — Lab as a Service

Our team, composed of seasoned experts in DevOps, Automation Infrastructure, and IT, recognized the need for a novel solution to address the challenges faced in managing Device Under Test (DUT) usage at DAZN. In the first quarter of 2023, we conducted a comprehensive assessment to identify and map out the major pain points associated with DUT utilization within the organization. This assessment culminated in the development of a pioneering approach coined “Lab as a Service,” aimed at alleviating the burden on developers and testers by handling all DUT-related tasks and physical infrastructure management.

During the second quarter of 2023, we introduced the Minimum Viable Product (MVP) setup, leveraging the Suitest framework — a vendor offering remote control and automation capabilities tailored for both HTML-based and native platforms. Our team meticulously orchestrated the deployment of all requisite physical infrastructure, assembling a dedicated lab housing 50 test devices within a spacious room at the DAZN Katowice office.

![DUT lab v1.0 (QA room — May 2023)](https://miro.medium.com/v2/resize:fit:640/format:webp/1*MDbJo7nq77qwWrRCpNvsrA.jpeg)_DUT lab v1.0 (QA room — May 2023)_

To bolster reliability and resilience, we implemented a comprehensive monitoring and alerting system, with Grafana serving as the frontend interface. Additionally, we integrated DUT recovery mechanisms to enhance the robustness of our testing environment.

![Grafana monitoring dashboard](https://miro.medium.com/v2/resize:fit:640/format:webp/1*8a9dpHjMLH1FZCQptCNplw.png)_Grafana monitoring dashboard_

The efficacy of these enhancements became apparent swiftly, as an increasing number of engineers gravitated towards utilizing the centralized lab infrastructure instead of configuring local devices. Leveraging the valuable feedback provided by users, we iteratively refined and enhanced the lab environment, ensuring that it continues to meet the evolving needs of our development and testing teams.

## Lab as a service 2.0

As the user base and the number of devices in the DAZN lab continued to surge, spurred by the establishment of a new DAZN site in India and the expansion of supported platforms throughout the latter half of 2023, we undertook further enhancements to the Katowice lab. These adjustments were aimed at accommodating the organization’s growth and fortifying its capacity to scale even further. Throughout the first quarter, we initiated a series of improvements targeting both the physical infrastructure and supporting features:

### Infrastructure enhancements

- **Relocation and Expansion:**

The DUT lab underwent relocation to a larger space, facilitating increased scalability and the seamless onboarding of additional platforms and devices. Leveraging existing racks more efficiently, we enabled the mounting of devices on both sides of each rack, resulting in a 50% boost in the number of TVs and a 40% increase in the number of set-top boxes (STBs), totaling 120 DUTs overall, with ample room for further expansion.

![DUT lab v2.0 (open space — January 2024)](https://miro.medium.com/v2/resize:fit:640/format:webp/1*IKUuw6aU2wQMzTBKCKS_bA.jpeg)_DUT lab v2.0 (open space — January 2024)_

- **Streaming Solutions:**

Collaborating with Suitest, we identified a cost-effective solution enabling the streaming of device HDMI output for STBs and gaming consoles. Additionally, we enhanced the TV streaming setup by refining the phone streaming app.

![Live DUT Streaming Phone camera setup](https://miro.medium.com/v2/resize:fit:720/format:webp/1*K460jjXRPqXUVQf8PVg0ng.jpeg)_Live DUT streaming mobile phone camera setup_

![HDMI grabber/splitter combo](https://miro.medium.com/v2/resize:fit:640/format:webp/1*s-cKLE8Oh1h5mliovKc91Q.jpeg)_HDMI grabber/splitter combo_

### Features enhancements

- **Collaboration with Suitest:**

Over the course of the year, we actively collaborated with Suitest to influence their roadmap, contributing to improvements in user experience and interface. A significant milestone was the introduction of a Unified UI integrating video streaming, remote control, and logs within the same browser window, enhancing productivity for lab users.

![Suitest Unified UI](https://miro.medium.com/v2/resize:fit:640/format:webp/1*Wg96k859tBYZwz13vVXG3Q.png)_Suitest Unified UI_

- **LR (Living Room) Launcher**:

In collaboration with the DAZN LivingRoom development team, we introduced LR Launcher, simplifying the workflow for manual testers and streamlining the onboarding process for partner platforms such as Sky and Movistar for the sake of ease of use when testing different development environments.

![LR Launcher](https://miro.medium.com/v2/resize:fit:640/format:webp/1*5AObQs1_Vdyd9cXhf5qF3w.png)_LR Launcher_

- **Enhanced Monitoring**:

With the growing user base and device count, robust monitoring became paramount. We revamped our monitoring service, streamlining processes and transitioning from an API-based solution to WSS notifications. Furthermore, we expanded monitoring coverage and implemented alerting for control units and streaming devices.

- **Automated Recovery**:

Leveraging smart plugs across all TV platforms, we implemented remote manual and automated recovery mechanisms for malfunctioning DUTs. Efforts are underway to develop similar mechanisms for cameras and control units, bolstering the resilience and efficiency of our lab environment.

- **Local Tunnel**:

A local tunnel solution providing exposure of a local development server allows developers and testers to seamlessly execute their code on a real physical lab device. This capability not only streamlines the testing process but also provides a practical and authentic environment for validating local code during development.

- **Lab Insights Report (BI)**:

An automated report detailing lab usage and testing trends. This report will provide valuable insights into usage patterns, enabling us to further optimize our services based on real data.

![BI Report](https://res.cloudinary.com/dtt2usukl/image/upload/v1716458627/Image_27-08-2023_at_11.58_sas9dh.jpg)_BI Report_

**VPN Change Region & VPN Console**:

A VPN solution that will empower lab users to seamlessly transfer DUTs between regions remotely. Using firewalls controlled by OPNSense, a lambda backend and a UI portal written in React (NextJS) users are able to testing of region-dependent content across all lab devices by switching regions as needed.

![VPN Console](https://res.cloudinary.com/dtt2usukl/image/upload/v1716458833/Image_23-05-2024_at_13.06_oxbech.jpg)_VPN Console_

## Service & Support

- **Continuous Improvement:**

Since the launch of the DUT lab in May 2023, we have continually bolstered our service and support infrastructure. This includes expanding our support engineering team, enhancing documentation and guides, and maintaining accessible support channels during working hours. Our commitment to swift issue resolution is underscored by our expanded remote resolution capabilities, facilitated by proactive monitoring and device error-based recovery processes.

- **Onboarding Optimization:**

Through the introduction of “Lab as a Service 2.0” and tailored training sessions, we have streamlined the onboarding process for new users, ensuring a seamless transition and maximizing productivity from the outset using official docs portal (using github pages and backstage).

These enhancements collectively represent our ongoing commitment to optimizing the DAZN lab environment, fostering innovation, and empowering our engineering teams to deliver exceptional quality across all platforms and devices.

## Acknowledgements

Special thanks to the amazing AP Nexus team for their dedication and hard work, it was a pleasure working with each one of you!

[Dennis Groisman](https://www.linkedin.com/in/dennis-groisman-9811b335/), [Daniel-Shahar Cohen](https://www.linkedin.com/in/danielshaharcohen/), [Rafal Muszalik](https://www.linkedin.com/in/rafa%C5%82muszalik/), [Blazej Milewski](https://www.linkedin.com/in/blazej-milewski/), [Alex Rivkin](https://www.linkedin.com/in/alex-rivkin-b037a6a1/), [Yoav Brown](https://www.linkedin.com/in/yoav-brown-970b55265/) and the one and only [Gal Moran](https://www.linkedin.com/in/gal-moran-a3454095/).

[![DAZN Lab as a Service](https://res.cloudinary.com/dtt2usukl/image/upload/v1716460903/play_video_lab_fnkajn.jpg)](https://res.cloudinary.com/dtt2usukl/video/upload/v1716460249/tvlab_v2_withmusic_twifar.mp4 'DAZN Lab as a Service - Nexus Team')_My team - Lab as a Service 2.0_

Article on [Medium (DAZN Engineering Blog)](https://medium.com/dazn-tech/lab-as-a-service-in-dazn-9bcc9d549970)


I'm incredibly proud to share the outstanding work our team has accomplished recently in upgrading our testing lab infrastructure.

Lab as a Service in DAZN


Makefiles are powerful tools in a developer's arsenal, capable of significantly improving efficiency and consistency in the build process. By understanding and utilizing Makefiles for Docker and Terraform, developers can streamline their workflows, reduce repetitive tasks, and focus more on the creative aspects of software development.

In this article, we'll explore the basics of Makefiles and how they can be used to automate the build process with examples of Docker and Terraform commands usage.

### What is a Makefile?

Incorporating a Makefile into your project can greatly streamline various development tasks, such as managing dependencies, running tests, building documentation,formatting code, performing static analysis, build/compile/transpile, cleaning up temporary files and more.

The flexibility of Makefiles allows you to automate routine processes in a structured and efficient manner. Here's a more detailed look at how you can leverage Makefiles in your projects:

### Basic Structure of a Makefile

A Makefile typically consists of targets, prerequisites, and recipes. Targets represent the goals you want to achieve, which could be actions like running your app or installing dependencies. Prerequisites are files or other targets needed before a target can be executed. Recipes are shell commands that make will execute.

For instance, you might have a `run` target to execute your Python application, a `setup` target for installing dependencies listed in `requirements.txt`, and a `clean` target for removing Python bytecode files or the `__pycache__` directory.

### Advanced Techniques

**Default Goals and Help Messages:** Setting a default goal, like `.DEFAULT_GOAL := help`, allows you to display a helpful message when make is run without a target.

**Virtual Environments:** Integrating virtual environments in your Makefile ensures that the right Python environment and dependencies are used. This can be crucial in projects with specific version requirements or multiple dependencies.

**Injecting Paths into `PYTHONPATH:`** This advanced trick allows Python to search for packages in specified file system paths without needing to install them, which can be particularly useful in projects with heavy dependencies or multiple branches.

### Some Practical Examples

**Running Tests with Coverage:** You can create a coverage target to run tests with coverage, specifying the commands for running the tests and generating a coverage report.

**Linting and Static Analysis:** A lint target can be set up to run various linting and static analysis tools like flake8, pylint, and mypy on your code.

**Task Dependencies and Parameters:** Makefiles allow you to specify dependencies between tasks. For example, a test target might depend on a lint target. You can also pass parameters to tasks, like specifying an IP and port for a serve task.

**Makefile for Docker Commands:**

By leveraging Makefiles in your Python projects, you not only save time but also create a more robust and maintainable development environment. These examples and advanced features illustrate just how versatile and powerful Makefiles can be in automating and streamlining your Python development workflow.

```makefile
export TOP := $(shell git rev-parse --show-toplevel)
export IMAGE_NAME := my-docker-image
export CONTAINER_NAME := my-container-image

build:
	docker build --build-arg JFROG_USERNAME=${JFROG_USERNAME} --build-arg JFROG_ARTIFACTORY_TOKEN=${JFROG_ARTIFACTORY_TOKEN} -t $(IMAGE_NAME) .

run:
	$(TOP)/scripts/run_docker.sh &

exec:
	docker exec -it $(CONTAINER_NAME) /bin/sh

delete:
	docker stop $(CONTAINER_NAME)
	docker rm $(CONTAINER_NAME)
```

In this example the `build` command creates a Docker image with build arguments (`make build`).

The `run` command runs a script to start the container in the background (`make run`). It runs a bash script that starts the container in the background and saves the container ID in a file.

The `exec` provides an interactive shell within the container for debugging proposes (`make exec`).

The `delete` stops and removes the container (`make delete`).

**Makefile for Terraform Commands:**

Terraform, an infrastructure as code software tool, also benefits from Makefiles, its commands can become quite complex. Here's an example on why it is useful to use Makefiles for Terraform:

```makefile
account := dev
namespace ?= $(shell echo $(USER) | awk '{print tolower($0)}')
service ?= $(shell basename `git rev-parse --show-toplevel`)

ifeq ($(account), dev)
    AWS_PROFILE := 123456
else ifeq ($(account), stage)
    AWS_PROFILE := 654321
else
    $(error Invalid account specified: $(account))
endif

init:
	terraform init -reconfigure \
		-backend-config=config/$(account)/backend.tfvars -backend-config=key=services/$(namespace)/$(service).tfstate

plan:
	terraform plan -out /tmp/$(service).tfplan \
		-var-file=config/$(account)/terraform.tfvars -var-file=config/$(account)/service.tfvars \
		-var "namespace=$(namespace)" -var "service_name=$(service)"

apply:
	terraform apply /tmp/$(service).tfplan

validate:
	@terraform validate

destroy:
	terraform destroy \
		-var-file=config/$(account)/terraform.tfvars -var-file=config/$(account)/service.tfvars \
		-var "namespace=$(namespace)" -var "service_name=$(service)"
```

This Makefile automates various Terraform commands like `init` which initializes a new or existing Terraform configuration (`make init`).
This command is particularly used to initialize the working directory containing Terraform configuration files, with -reconfigure to force re-initialization. It specifies backend configuration variables and the state file location.

The `plan` command creates an execution plan. This command creates a Terraform plan and outputs it to a temporary file. It specifies variable files and sets additional variables like namespace and service_name (`make plan`).

The `apply` command applies the changes required to reach the desired state of the configuration (`make apply`).

The `validate` command is used to validate the syntax and configuration of the Terraform files (`make validate`).

The `destroy` command is used to destroy all resources managed by the current Terraform configuration (`make destroy`).

In conclusion, Makefiles stand as a fundamental and powerful tool in software development, offering a versatile approach to automating and streamlining various aspects of the build process. Whether you're working with Docker containers or managing infrastructure with Terraform, Makefiles bring a level of efficiency and consistency that can significantly enhance your workflow. By effectively leveraging Makefiles, developers not only save valuable time but also ensure a more robust and maintainable development process, making them an essential component in any modern software project.


Makefiles are a cornerstone tool in the software development process, particularly useful for streamlining and automating the build process.

Why Makefile is One of My Essentials Tools in Every Software Project


In continuation to my [previous post](./hosted-machines-monitoring), I took the OpenTelemetry Collector Contrib project and wrapped it in a minimal CLI using the [Appeal library](https://github.com/larryhastings/appeal/). This CLI is also compatible for all the common OS's (Windows, MacOS and Linux).

The goal was to make the OTEL binary a little bit more focused for our team use case, and easy for installation and usage.

## So What is Appeal?

Appeal is a command-line argument processing library for Python, like argparse, optparse, getopt, docopt, Typer, and click. But Appeal takes a refreshing new approach - the easiest approach I've seen so far. You're basically write Python functions, and Appeal translates them into command-line options and arguments.

Although its official documentation declares it wasn't tested on Windows yet, I found it to be working just fine (on Windows 10).

## My Use Case

I wanted to create a CLI that will be able to download the latest OTEL binary from the official GitHub repository, install it, run it, stop it and uninstall it without the user to worry about anything.
So in my use case at work I created the following commands: Download, Install, Run, Stop, Uninstall and Help is being generated for us automatically due to appeal's nature.

To start the CLI, I created `otel_cli` with a shebang so it'll be able to run as a script.
In addition to that i've added the imports used, initialization of the Appeal object and some global variables that will be used later.

```python
#!/usr/bin/env python3

import os
import shutil
import signal
import socket
import tarfile
import subprocess
import appeal
import ssl
import platform
import sys

from urllib.request import urlretrieve

app = appeal.Appeal()

ssl._create_default_https_context = ssl._create_unverified_context

CORALOGIX_DOMAIN = 'somedomain.coralogix.com'
CORALOGIX_APP_NAME = 'CI-Machine'
OTEL_PID = 'otelcol.pid'
_OTEL_CONTRIB_DIR = 'otel-contrib-dir'

#...commands...#

if __name__ == '__main__':
    app.main()

```

Now let's dig in into the commands themselves.

**download command**

```python
def get_machine_type():
    arch = platform.machine()
    if arch in ['AMD64', 'x86_64']:
        return "amd64"
    elif arch in ['arm64', 'aarch64']:
        return "arm64"
    else:
        print("Failed to detect architecture")
        raise ValueError(f"Unsupported architecture: {arch}")


def get_os_type():
    if platform.system() == 'Darwin':
        print("Detected MacOS")
        return "darwin"
    elif platform.system() == 'Linux':
        print("Detected Linux OS")
        return "linux"
    elif platform.system() == 'Windows':
        print("Detected Windows OS")
        return "windows"
    else:
        raise ValueError(f"Unsupported OS: {platform.system()}")

@app.command()
def download(version="0.89.0", ignore_cache=True):
    '''Download the OpenTelemetry Collector binary for the current machine.'''
    machine_type = get_machine_type()
    os_type = get_os_type()
    filename = f"otelcol-contrib_{version}_{os_type}_{machine_type}.tar.gz"
    url = f"https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v{version}/{filename}"
    print(f"Downloading: {filename}...")
    if not os.path.exists(_OTEL_CONTRIB_DIR):
        os.mkdir(_OTEL_CONTRIB_DIR)
    urlretrieve(url, f'./{_OTEL_CONTRIB_DIR}/{filename}')
    print("Download completed.")

```

The `download` command is designed to fetch the OpenTelemetry Collector binary that matches the user's system architecture and operating system. This command leverages Python's built-in `platform` and `urllib` libraries to determine the system specifics and retrieve the necessary files.

`get_machine_type`: This function uses `platform.machine()` to identify the system's architecture. It supports both amd64 and arm64 architectures. If the architecture is not recognized, it raises a ValueError.

`get_os_type`: This function uses `platform.system()` to determine the operating system (Darwin for macOS, Linux, and Windows). Like `get_machine_type`, it raises a ValueError if the OS is not supported.

The download function first retrieves the machine type and OS type using the above functions. It constructs the filename for the binary to download, using the version number, OS type, and machine type.

The binary's URL is generated based on the filename. It checks if the directory `_OTEL_CONTRIB_DIR` exists and creates it if it doesn't. The binary is downloaded using `urlretrieve` and stored in `_OTEL_CONTRIB_DIR`. And eventually the function prints a message to indicate the completion of the download.

**install command**

```python
@app.command()
def install():
    '''Install the OpenTelemetry Collector binary for the current machine.'''
    files = [f for f in os.listdir(f'./{_OTEL_CONTRIB_DIR}') if f.endswith('.tar.gz')]
    if not files:
        raise FileNotFoundError("No tar.gz file found for installation.")
    tar_file = f'./{_OTEL_CONTRIB_DIR}/{files[0]}'
    print(tar_file)

    print(f"Installing {tar_file}...")
    with tarfile.open(tar_file, 'r:gz') as tar:
        tar.extractall(path=f'./{_OTEL_CONTRIB_DIR}')
    print("Installation completed. You can now run the application by running: otel_cli run")
```

The `install` command is responsible for setting up the OpenTelemetry Collector binary on the user's system. This command assumes that the binary has been downloaded using the download command and is present in a specific directory.

The function begins by searching for `.tar.gz` file within the `_OTEL_CONTRIB_DIR` directory. This is where the downloaded binary should be located. If no such file is found, it raises a FileNotFoundError.

Once the file is located, the script prints the path of the tar file, indicating the binary to be installed. Using Python's `tarfile` module, the script opens the .tar.gz file in read mode and extracts its contents to \_OTEL_CONTRIB_DIR.

After successfully extracting the files, the script prints a message indicating that the installation is complete and that the user can now run the application.

The use of `tarfile.open` in the with statement ensures that the tar file is correctly opened and closed after the operation, following Python's recommended practice for file handling. The `extractall` method is used to extract all contents of the tar file into the specified directory. This approach abstracts the extraction process from the user, making the installation process straightforward and user-friendly.

**run command**

```python
@app.command()
def run(config_file="config.yaml"):
    '''Run the OpenTelemetry Collector binary for the current machine.'''
    binary_name = 'otelcol-contrib'
    if sys.platform == 'win32':
        binary_name += '.exe'

    otel_col_path = os.path.join(os.getcwd(), _OTEL_CONTRIB_DIR, binary_name)
    if not os.path.isfile(otel_col_path):
        raise FileNotFoundError(f"{otel_col_path} binary not found. Please install first.")

    os.environ['CORALOGIX_SUBSYS_NAME'] = socket.gethostname()

    if os.path.isfile('.env'):
        with open('.env') as f:
            for line in f:
                key, value = line.strip().split('=', 1)
                os.environ[key] = value

    os.environ['CORALOGIX_DOMAIN'] = CORALOGIX_DOMAIN
    os.environ['CORALOGIX_APP_NAME'] = CORALOGIX_APP_NAME

    print("Starting OpenTelemetry Collector...")
    process = subprocess.Popen([os.path.join(os.getcwd(), _OTEL_CONTRIB_DIR, 'otelcol-contrib'), '--config', config_file])
    with open(OTEL_PID, 'w') as otel_pid:
        otel_pid.write(str(process.pid))
    print(
        f"OpenTelemetry Collector is running in the background with APP_NAME: "
        f"{os.environ['CORALOGIX_APP_NAME']} and Subsystem: "
        f"{os.environ['CORALOGIX_SUBSYS_NAME']}")
```

The `run` command in the CLI tool is designed to start the OpenTelemetry Collector with a specified configuration. This command is crucial as it activates the collector, allowing it to begin its telemetry data collection and processing.

The function starts by setting the default binary name to `otelcol-contrib`.
It then checks if the platform is Windows `(sys.platform == 'win32')`. For Windows, it appends .exe to the binary name, accommodating the typical executable format in Windows.

The script constructs the full path to the binary using `os.path.join`, combining the current working directory, the \_OTEL_CONTRIB_DIR, and the binary name.
If the binary does not exist at the constructed path, a FileNotFoundError is raised, prompting the user to install the binary first.

The script sets the `CORALOGIX_SUBSYS_NAME` environment variable to the hostname of the current machine, using `socket.gethostname()` (this is how the machine your'e monitoring will displayed in your future Coralogix dashboards).
It then reads environment variables from a `.env` file, if present, and sets them in the current environment.
The `CORALOGIX_DOMAIN` and `CORALOGIX_APP_NAME` are set to a predefined values (global variables above).

The script prints a message indicating the start of the OpenTelemetry Collector.
It uses `subprocess.Popen` to start the OpenTelemetry Collector in the background, passing the path to the binary and the configuration file as arguments.
The process ID (PID) of the started collector is written to a file named `OTEL_PID`. This PID is used later for stopping the collector.

After successfully starting the collector, the script prints a message with details about the running instance, including the APP_NAME and Subsystem, derived from the environment variables.

**stop command**

```python
@app.command()
def stop():
    '''Stop the OpenTelemetry Collector.'''
    if not os.path.exists(OTEL_PID):
        print("OpenTelemetry Collector is not running or pid file is missing.")
        return

    with open(OTEL_PID, 'r') as otel_pid:
        pid = int(otel_pid.read().strip())

    try:
        if sys.platform == 'win32':
            subprocess.run(['taskkill', '/F', '/PID', str(pid)], check=True)
        else:
            os.kill(pid, signal.SIGTERM)

        os.remove(OTEL_PID)
        print("OpenTelemetry Collector has been stopped.")
    except ProcessLookupError:
        print("Process not found. It may have already been stopped.")
    except Exception as e:
        print(f"Error stopping the process: {e}")
```

The `stop` command in the CLI tool is designed to safely terminate the OpenTelemetry Collector that's running in the background. This command is crucial for stopping the collector's processes and ensuring that resources are properly released.

The function first checks if the `OTEL_PID` file exists. This file contains the process ID (PID) of the running OpenTelemetry Collector. If the file does not exist, it indicates that the collector is not running, and the function returns after printing an appropriate message.

To read the file with the PID in it, for Windows systems `(sys.platform == 'win32')`, it uses the `subprocess.run` method to execute the `taskkill` command, forcibly terminating the process with the specified PID.
For other operating systems, it uses the `os.kill` function with the SIGTERM signal to terminate the process. This is a more Unix/Linux traditional way of handling process termination.

The script then removes the `OTEL_PID` file, signifying that the collector has been stopped.

The script also includes exception handling for situations where the process may not be found (ProcessLookupError) or other unexpected errors occur. This ensures that any issues during the stopping process are gracefully caught and reported to the user.

Eventually, the script prints a message to inform the user that the OpenTelemetry Collector has been stopped successfully, providing clear feedback on the action's outcome.

**uninstall command**

```python
@app.command()
def uninstall():
    '''Uninstalls the OpenTelemetry Collector files and binaries.'''
    if os.path.exists(_OTEL_CONTRIB_DIR):
        print(f"Uninstalling OpenTelemetry Collector from {_OTEL_CONTRIB_DIR}...")
        shutil.rmtree(_OTEL_CONTRIB_DIR)
        print("Un-installation complete. OpenTelemetry Collector has been removed.")
    else:
        print("OpenTelemetry Collector is not installed or has already been uninstalled.")
```

The `uninstall` command is a crucial component of the CLI tool, enabling users to remove the OpenTelemetry Collector and its associated files from their system. This command ensures that users can cleanly and easily reverse the installation process.

The command starts by checking if the `_OTEL_CONTRIB_DIR` directory exists. This directory is where the OpenTelemetry Collector's binaries and related files are stored post-installation. If the directory exists, indicating that the collector is installed, the script prints a message to inform the user that the uninstallation process has begun.

It then uses the `shutil.rmtree` function to recursively delete the \_OTEL_CONTRIB_DIR directory, thereby removing all the files and binaries related to the OpenTelemetry Collector. This function is a powerful tool in Python for directory removal, capable of deleting a directory and all its contents.
After successful removal of the directory, the script prints a confirmation message indicating the completion of the uninstallation process.

If the \_OTEL_CONTRIB_DIR directory does not exist, the script assumes that the OpenTelemetry Collector was never installed or has already been uninstalled. In such cases, it prints a message stating that the OpenTelemetry Collector is not installed or has already been removed.
This check prevents the script from attempting to delete non-existent directories, which could lead to errors.

## Usage

For convenience, add the `otel_cli` executable to your PATH and update its symlink to point the python binary. You can do that by running the following command (MacOS):

```bash
ln -s $PWD/otel_cli /usr/local/bin/otel_cli
```

To download the OpenTelemetry collector, run the following command:

```bash
otel_cli download
```

Then you can install the OpenTelemetry collector as a service by running the following command:

```bash
otel_cli install
```

You can start the OpenTelemetry collector by running the following command:

```bash
otel_cli run
```

You can stop it by running the following command:

```bash
otel_cli stop
```

You can uninstall it by running the following command:

```bash
otel_cli uninstall
```

For further help simply run:

```bash
otel_cli help
```

And it'll output all the available commands and their descriptions (taken from the functions docstring).


In continuation to my previous post, I'll show you how to create a minimal and compatible CLI using the Appeal library.

Easy Minimal CLI Using Appeal Library


I absolutely love using pytest as my go-to testing tool whenever I'm working on a Python project. Among its many features, pytest's parametrization is a standout and has been crucial in simplifying my test cases and maximizing code reusability instead of duplicating tests.
What's more, if it's not clear, parameterized tests are being presented as different tests on each test execution and report.

I have been using it ever since I started using pytest for many different test types, such as unit tests, end-to-end tests, and integration tests.

Let's take a look at some real examples (although I had to make them a little bit more generic) of how to use pytest parametrization.

**Basic Parametrization**

The most straightforward use case is to feed multiple inputs to a single test function. This is useful for testing different edge cases or multiple data points without writing separate test cases.

In the example below we wrote a test function that asserts on the length of a string, while with parametrization we can test multiple strings with a single test function.
We provide test input (for example "hello") and the expected result (in this case 5).

```python
import pytest

@pytest.mark.parametrize("input_string, expected_length", [
    ("hello", 5),
    ("world", 5),
    ("pytest", 6)
])
def test_string_length(input_string, expected_length):
    assert len(input_string) == expected_length
```

**Indirect Parametrization**

With indirect, rather than passing the value directly to the test function, pytest will treat the given input as a fixture name. This allows us to use the fixture's logic to transform or process the input value before passing it to the test. Here, each number is doubled using a fixture before being tested.

```python
import pytest

@pytest.mark.parametrize("input, expected", [
    (2, 4),
    (3, 6),
    (4, 8)
], indirect=["input"], ids=["double-2", "double-3", "double-4"])
def test_double_number(input, expected):
    assert double(input) == expected

@pytest.fixture
def double(input):
    return input * 2
```

**Custom ID for Test Cases**

Sometimes it's beneficial to have custom names for our tests, especially when the default names become unclear or too lengthy. The ids parameter lets us define custom names for our tests. In this example, we use custom IDs to quickly identify tests for even and odd numbers using the `ids` argument that expects a list.

```python
import pytest


@pytest.mark.parametrize("number", [2, 3, 4, 5, 6],
                         ids=["ev-2", "odd-3", "ev-4", "odd-5", "ev-6"])
def test_even_numbers(number):
    is_even = number % 2 == 0
    if "ev" in pytest.current_test().name:
        assert is_even
    else:
        assert not is_even
```

**Parametrization with Multiple Arguments**

Here we demonstrate how to parameterize a test with multiple input arguments. We're testing the addition of two numbers. This is a powerful feature when we want to test a function with various combinations of inputs.

```python
import pytest


@pytest.mark.parametrize("a, b, expected_sum", [
    (1, 2, 3),
    (3, 3, 6),
    (5, 5, 10)
])
def test_addition(a, b, expected_sum):
    assert a + b == expected_sum
```

**Utilizing Parametrization in Asynchronous Tests**

With pytest-asyncio plugin (which is kind of standard these days), we can also parameterize asynchronous tests. In this example, we test an asynchronous function that returns a string after a built-in delay just for the sake of the example.

```python
import pytest
import asyncio

@pytest.mark.parametrize("input_str", ["hello", "world", "pytest"],
                         ids=["hello-async", "world-async", "pytest-async"])
@pytest.mark.asyncio
async def test_string_return_async(input_str):
    returned_str = await async_string_return(input_str)
    assert returned_str == input_str

async def async_string_return(s):
    await asyncio.sleep(0.1)
    return s
```

**Parametrization with Dynamic Fixtures**

The following test demonstrates a more complex usage of pytest's parametrization. This is especially the case when you have a dynamic fixture (like `dynamic_execution_payload`) which depends on the parameter passed (`path_arg` in this example).

```python
import pytest


BASE_URL = "http://example.com/api"


@pytest.fixture(scope="function")
def payload_for_execution(file_path):
    return {
        'repo': "demo-repo",
        'ref': "demo-branch",
        "paths": [file_path]
    }


@pytest.mark.parametrize('file_path', [
    'mock/testDir/file4.js',
    'mock/testDir/file3.js',
    'mock/testDir/file1.js',
    'mock/testDir/file2.js'
])
def test_file_processing_with_dynamic_payload(file_path, payload_for_execution):
    execution = requests.post(url=f"{BASE_URL}/startExecution",
                              headers={"Content-Type": "application/json"},
                              json=payload_for_execution)
    execution_id = execution.json()['executionId']

    response = requests.get(url=f"{BASE_URL}/getExecution/{execution_id}",
                            headers={"Content-Type": "application/json"})
    response_json = response.json()

    assert 'result' in response_json, "Response JSON does not have the 'result' key"
```

First we specified different file paths as inputs for the test function. For each of these paths.
With it we utilize the fixture `payload_for_execution` that generates a payload dynamically based on the `file_path` given and this payload is then used in our test. This demonstrates how you can use the parameters of your test in your fixtures, making your tests more dynamic and flexible.

Within the test, an HTTP POST request is made to presumably start some kind of processing for the file specified in the payload. This request returns an execution ID. A subsequent GET request fetches the result of this execution. Finally, assertions are performed on this result.

**Parametrization with Using Dates**

Dates in every language can be a true pain, therefore its important we tests them thoroughly.

```python
import pytest

from datetime import datetime, timedelta


@pytest.mark.parametrize(
    "input_date, expected_query",
    [
        (datetime(2023, 3, 15), "?start=2023-03-01&end=2023-04-01&category=premium"),
        (datetime(2023, 12, 31), "?start=2023-12-01&end=2024-01-01&category=premium"),
        (datetime(2023, 1, 1), "?start=2023-01-01&end=2023-02-01&category=premium"),
        (datetime(2023, 2, 28), "?start=2023-02-01&end=2023-03-01&category=premium"),
    ],
    ids=["mid-March", "end-of-year", "start-of-year", "end-of-February"]
)
def test_generate_query_based_on_date(mocker, input_date, expected_query):
    mocker.patch("module_under_test.datetime", MockDateTime)
    MockDateTime.now = mocker.Mock(return_value=input_date)
    assert generate_query_for_service(category='premium') == expected_query

def generate_query_for_service(category):
    now = datetime.now()
    start_of_month = now.replace(day=1)
    end_of_month = (start_of_month + timedelta(days=32)).replace(day=1)
    return f"?start={start_of_month.strftime('%Y-%m-%d')}&end={end_of_month.strftime('%Y-%m-%d')}&category={category}"
```

First we provide different datetime objects as inputs. For each datetime, there's a corresponding expected output. This is a classic example of table-driven testing. But in many scenarios, functions might fetch the current date and time using something like `datetime.now()` which can become tricky for testing, thats why you want to control what "now" means so you can predict the output. The `mocker.patch` (using pytest-mock plugin) line mocks the return value of `datetime.now()` to be the input date, allowing for predictable and testable behavior.

The hypothetical function `generate_query_for_service` is responsible for generating a query based on the current date and a given category. It's logic is simplified for this demonstration.

With this structure, you're dynamically testing the function with multiple dates, ensuring it works across various boundary conditions like month ends, year starts, etc.

**Parametrization for Testing Common Web Components**

There are some cases where you have components that are used in multiple places in your application. For example, you might have a navigation bar component that displays on all of your website pages. So you might want to test that component in isolation, but also test it in the context of different pages, this can be done easily with parametrization.

```python
import pytest


test_data = [
    ('https://www.kazis.dev/', "homepage"),
    ('https://www.kazis.dev/about', "about page"),
    ('https://www.kazis.dev/blogs', "blogs page"),
    ('https://www.kazis.dev/blogs/python-type-checking', "blog page")
]

@pytest.mark.parametrize("test_url, test_id", test_data, ids=[item[1] for item in test_data])
def test_logo_element_exists(sync_page, test_url, test_id):
    home_page = NavBar(sync_page)
    home_page.open(test_url)
    link = sync_page.locator('role=link >> text=Kazis Dev Blog Kazi\'s Dev Blog')
    assert link, "Link not found."
```

So in this example we're using Playwright test framework, but the same concept can be applied to any other framework. We have a list of tuples, each tuple contains a URL and a test ID. The test ID is used to identify the test in the report and display it in every test execution.
We then use the `parametrize` decorator to pass the URL and test ID to the test function. We also use the `ids` argument to specify the test ID for each test case.

And in the test function itself we assert that the logo element exists on each page that we're providing as an input. This test will run for each URL in the list and it's a classic example how we can test simple functionality of a web component that doesn't change on different pages.

To conclude,
Pytest parametrization is a powerful feature that reduces redundancy, improves test clarity, and enhances the reusability of test code. These are just a few examples showcasing its capabilities. Using parametrization, you can write more concise and robust tests, and that's without mentioning other out of the box plugins which support pytest.


I absolutely love using pytest as my go-to testing tool whenever i'm working on a Python project. Pytest Parametrization is one of reasons why.

Powerful Pytest Parametrization


I'll start with a disclaimer - I'm fairly new to Python typing system while you actually need to enforce it, and I'm still learning. I'm writing this post to share my experience and what I've learned so far.

MyPy is a static type checker for Python. It combines the benefits of dynamic typing and static typing. As it is optional, you don't have to use this in your codebase, but once you decide to use it, it can significantly increase code readability and detect certain types of bugs before runtime.

So at first, I've started with a simple Python service, and I've added MyPy to it while mypy runs in a non-strict mode. I've added a few type hints and return values to the code, and it was pretty straightforward.
Then I went on and ran it in strict mode, which I think requires some explanation for those who are not familiar with it.

**Strict mode vs Non-strict mode**

MyPy's "strict mode" is an optional feature that enforces stricter type checking rules in your code. This is intended to catch more potential bugs and make your code more robust, at the cost of needing to be more explicit about your types.

In non-strict (or normal) mode, MyPy uses a certain set of rules to check your types. This is a balance between flexibility and safety. It's designed to catch the most obvious type errors while not requiring too much type annotation overhead.

When you enable strict mode, you're telling MyPy to use a more rigorous set of rules. These rules catch more potential issues but also require more complex type annotations.

Here are some of the things that change when you enable strict mode:

- All function parameters and return values must be explicitly typed. Without strict mode, MyPy allows functions without type annotations.

- Variables won't be implicitly typed as `Any`. Without strict mode, MyPy will often infer the type of a variable as Any if it can't determine its type.

- MyPy will check the body of every function, regardless of whether it has type annotations. Without strict mode, MyPy might skip the bodies of functions without type annotations.

- Optional types must be explicit. In normal mode, MyPy will infer an implicit Optional for arguments with a default value of None. Strict mode requires Optional to be explicit.

And I assume they are way more to discover.

Strict mode is enabled by adding `--strict` to the command line when running MyPy, or by adding `strict = True` to the MyPy configuration file. It's often a good idea to enable strict mode if you're starting a new project, as it encourages good type annotation habits from the beginning. However, for existing projects, turning on strict mode could lead to a large number of errors that need to be fixed.
Whether or not to use strict mode in such cases can be a team decision based on the specific requirements of the project and the team's capacity to handle the additional type annotation work.

In general, strict mode can be a very powerful tool to ensure type safety and can lead to higher quality, more maintainable code. However, it comes with an overhead in terms of the additional type annotations required, so it's a trade-off that each project team needs to evaluate for themselves.

**On to the PoC**

So I started with a PoC on one of our small Python services (~30 src files).

At first, mypy was installed as a dev dependency and ran in a non strict mode. After a batch of fixes I enabled strict mode which had quite significant amount of errors that discovered two potential bugs, and noted that in the code I was using a pattern which is considered as a risky one (Liskov Substitution Principle (LSP)). In simple terms, states that if a program is using a base class, it should be able to use any of its subclasses without the program knowing it.

After a day of fixes, mypy ran perfectly on my local dev machine.

I've implemented both pre-commit hooks (for staged changes) and CI workflow (for all src files) to run mypy on the codebase. Configuration file (mypy.ini) was added to the root of the project, and it's being used by both pre-commit and CI workflow.
In High level, the workflow runs mypy in strict mode but ignores import checks for 3rd party packages.

**Issues Encountered**

- I had a known type, but there’s a slight option that the type will be different. In such case you can use the built-in cast method from the typing module. It'll tell the static type checker that we're expecting a specific type even if it could be something else.

For example:

```python
from typing import cast

def check_if_attribute_exists(argument: str) -> str | None:
    # [...]
    device = SomeClass(argument).execute()
    if 'attribute' in device:
        return cast(str, device['attribute'])
    else:
        return None
```

`execute` method returns the following type - `dict[str, Any]` - meaning a dictionary with keys as string, while the value can be anything.
Then if there’s a attribute key in the dictionary, we fetch it’s value and treats it’s like a string, although it can be anything else per our return value defined for `execute` method.

In such case, mypy throws the following error:
`src/api/device.py:28: error: Returning Any from function declared to return "dict[str, Any]" [no-any-return]`

In the following case, we know that if attribute exists, it’ll be a string, therefore we can use the cast method and it’ll solve the mypy error.
Of course, in any case like this or similar to an error that you want to remove, you can use the following comment next to the problematic code line `# type: ignore`, although I'm not a fan of it.

- I couldn’t find a way the set a global exclusion of import checks across all 3rd packages. Current solution was to define it in the config file per package such as in this case for the `requests` library.

```ini
[mypy-requests]
ignore_missing_imports = True
```

- Custom Types - There are scenarios where you’d like to create your own custom type. Cases like where you have complicated nested type, or a case where you want to return something specific like in this example:

I have a method in a class that should return a response object from a 3rd party API response. There are cases where i’d like to get some of its attributes such as status_code, content, reason, etc. There are many options to define a custom type, but one of the easiest ways is to use a `TypeAlias` as suggested by mypy’s docs. [Kinds of types - mypy 1.5.1 documentation](https://mypy.readthedocs.io/en/stable/kinds_of_types.html)

I can save the response object in a variable wrapped in TypeAlias and use it as a return value of a method.

```python
import requests

from typing import TypeAlias

Response: TypeAlias = requests.models.Response

def execute(self) -> Response:
    response_json = self._execute(self._base_url + self._path() + self. _query_params())
    return response_json
```

- What type should I've used for \*args and \*\*kwargs - If you don’t know what type you should expect, it is common practice to use the `Any` type? This is what I did in my case, but I know it's not considered as best practice (Will get to it later).

- Environment variables default values - Due to the fact the that env vars can be something and can be `None` (if not passing anything), mypy raises an error saying that for example env var called `SERVER_PORT` can be int or None, and when using this env var to start the server, mypy complains you’ll have an issue in case the port is None obviously. This can be solved in two practices; set default value (which is more common practice per mypy’s docs), or when using the env var, wrap it with cast.

For example, let’s say I have a server using this `SERVER_PORT` env var for startup

```python
import uvicorn

uvicorn.run(app, host='0.0.0.0', port=int(os.environ.get('SERVER_PORT', '8080')))
```

Server runs without passing anything to this env var, due to default port.

By the way, my team members are against default values in case like this as it can hide issues from us, so instead of assigning a default port, I used a clear string as default so the problem (if there is one) will be right there in our faces.

```python
import uvicorn

uvicorn.run(app, host='0.0.0.0', port=int(os.environ.get('SERVER_PORT', 'MissingPort')))
```

The error message will be very clear what went wrong:

`Error: 'MissingPort' is not a valid port number.`

- Remember the `Any` type question? well, I got to know Generics after some research. I Should have used Generics in many cases which can provide us much better solution for type Annotating even unknown types. Awesome and detailed explanation can be found here - [Python Typing Generics and Python 3.11 Variadic Generics](https://www.youtube.com/watch?v=7Chd5gPHlDg&ab_channel=AnthonyShaw). Seriously, it's one of the best detailed explanation I've seen so far.

- Our Internal Packages needs to be marked as type safe -

An internal Python library we're using in some of our services was flagged by mypy as not safe although I already implemented types in the library.
After some research I found out that I need to add a `py.typed` file to the package. [PEP 561 -- Distributing and Packaging Type Information](https://www.python.org/dev/peps/pep-0561/).

After reading PEP-561 I realized I should have included a py.typed file, or the type checker won't use the type hints provided by the package.
It's fairly simple to include this file: just touch a `py.typed` file in your package root directory and include it in your distribution.

I'm using poetry, so I added

```toml
packages = [
  {include = "mypackage"},
  {include = "mypackage/py.typed"},
]
```

If you're using `setup.py` (which you shouldn't), you can add package_data to setup call:

```python
setup(
    package_data={"mypackage": ["py.typed"]},
)
```

**Common Issues**

- Missing Stub Packages:

If you see errors about missing library stubs for a package like `requests` for example, you can use the `--install-types` command line option with MyPy to automatically install the missing type stubs packages. But using it in CI is a bit hacky in my point of view, because it requires a user input. To workaround it you can run the command like this:

```bash
yes | mypy --install-types.
```

Honestly, this doesn’t seem as a ideal solution to me, because not all packages has `py.typed` defined in their `pyproject.toml`. I preferred using the ignore imports flags instead due to this.

- Virtual Environment Issues:

MyPy type checking might behave differently inside and outside of a Python virtual environment. If you're encountering errors when running outside a virtual environment, try running MyPy inside the virtual environment where the package dependencies are installed.

- Dealing with Dynamic Attributes:

When dealing with classes that have dynamic attributes (like classes using `__getattr__` or `__setattr__`), you might encounter errors about attributes not being defined. This is because MyPy can't know about these attributes at the static type checking phase. You might need to add type annotations or comments to help MyPy understand these.

That's about it. Would love to get some feedback and hear about your experience with MyPy as I'm just starting to learn how to work with type annotations in Python.


I've assigned with a task to integrate MyPy into our Python services. Here's what I've learned so far (and still learning).

Integration of MyPy into Python Our Services


One of the many features of FastAPI is support for background tasks. I first encountered this feature when I was working on a project at work that required processing large datasets triggered from an API endpoint.

In This article I'll share what i've learned about FastAPI's background tasks, with some insights and examples, especially in the context of data manipulation using the popular Pandas library.

**So What are Background Tasks?**
In web development, many times, you'd come across operations that are time-consuming. These could be sending out emails, handling database migrations, or processing large datasets. Running such operations (even when using async operations) can block your API from responding to other requests, or worst - hang the response from an HTTP request, leading to a degraded user experience.
That's where background tasks come in play. They allow these time-consuming operations to run "in the background", ensuring that your API remains responsive.

FastAPI provides a simple way to execute functions in the background using the `BackgroundTasks` class. This lets you execute functions asynchronously while sending the response to the client.

Let's create an example wheres an endpoint in FastAPI receives some data, triggers a background task to process it with Pandas, and instantly sends a response back to the client.

Here's a simple FastAPI application:

```python
from fastapi import FastAPI, BackgroundTasks
import pandas as pd
import time

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "OK"}
```

Now let's add an endpoint that will receives a list, convert it to a Pandas DataFrame, process it in the background, and return a response to the client immediately:

```python
def process_data(data: list):
    df = pd.DataFrame(data)
    # Just for demonstration, we'll simulate some processing time
    time.sleep(5)
    print(df)

@app.post("/process/")
def process_endpoint(data: list, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_data, data)
    return {"message": "Data is being processed"}
```

In this example, when you send a POST request with a list of data to the "/process/" endpoint, FastAPI will trigger the `process_data` function as a background task and immediately respond with `{"message": "Data is being processed"}` to the client while the task is being processed in the background.

Things to take into account (learned the hard way), we can basically say background tasks are silent workers, meaning that if they fail, you might not know. Thus, error handling is crucial:

```python
def process_data(data: list):
    try:
        df = pd.DataFrame(data)
        time.sleep(5)
        print(df)
    except Exception as e:
        print(f"Error occurred: {e}")
```

Another thing to take into account is that background tasks are not guaranteed to run. If your application crashes or restarts, the background tasks will be lost. If you need to ensure that a task is executed, you should consider using something like a message queue or an async task handler like Celery, which is considered as a better practice than these lightweight background tasks.


FastAPI feature for lightweight background tasks (functions) that will run in parallel with your main application.

FastAPI Background Tasks - Introduction Guide


When it comes to testing in Python, pytest stands out as an excellent framework with its rich feature-set that includes detailed info on failing assert statements, auto-discovery of test modules, and advanced fixture management. While working on unit testing for one of our python services at work I wanted to test some scenarios that involves some more critical logging for debugging at real time.

Taking a look at [pytest official docs](https://docs.pytest.org/en/7.1.x/how-to/logging.html) I got to know the `caplog` fixture. I had many cases before where I actually found better solutions from built-in fixtures, but this time I was amazed by how easy it was to use and how much it can do. And it's safe to say I never looked back when it's comes to logging capturing in pytest.

**So what is caplog?**

In pytest, caplog is a fixture provided by pytest, which captures log messages emitted by the logging module. This allows you to validate that your application is logging the right messages or to debug a test by capturing and examining the log output and the log level.

Let's start with a concrete test case to illustrate caplog's usage:

```python
import pytest
import logging

def test_logging(caplog):
    logging.getLogger().info("Test logging")
    assert "Test logging" in caplog.text
```

In the test above, we import the caplog fixture from pytest. We then log an info message and assert that the message is present in the captured log text.

While caplog is great for checking that a specific log message was output, it can do a lot more. You can use it to validate the level of logging messages, check the logger that issued them, and even change the log level for the duration of the test.

Here's an example where we're doing more advanced checks:

```python
def test_logging_advanced(caplog):
    caplog.set_level(logging.INFO)
    logger = logging.getLogger("TestLogger")
    logger.warning("This is a warning")

    assert "This is a warning" in caplog.text
    assert caplog.records[0].levelname == "WARNING"
    assert caplog.records[0].name == "TestLogger"
```

In the example above, we first set the logging level to INFO, then log a warning message. Afterwards, we assert that the message is in the captured log, but also that the level of the message is WARNING and that it came from the logger "TestLogger".

You can also assert messages from specific loggers:

```python
def test_logging_from_specific_logger(caplog):
    logger = logging.getLogger('MyLogger')
    logger.error("An error message")
    assert "An error message" in caplog.text
    assert caplog.records[0].name == "MyLogger"
```

In this test, we create a logger with the name 'MyLogger', emit an error message, and then confirm that the message was logged by that specific logger.

In some cases, you may want to clear the messages captured by caplog. This is particularly useful in parameterized tests or when a single test function is checking multiple scenarios.

```python
def test_clearing_log_messages(caplog):
    logger = logging.getLogger()
    logger.info("First message")
    caplog.clear()

    logger.info("Second message")
    assert "First message" not in caplog.text
    assert "Second message" in caplog.text
```

Here, after logging the "First message", we clear the caplog. Then we log the "Second message" and check that the first message is not in the captured log, while the second message is.

There are some cases where the tested function or method may produce multiple logs. In such cases, you can use the caplog.records list to check the log messages. This list contains all the log records captured by caplog, and you can use it to check the log messages, levels, and loggers.

```python
import pytest
import logging

def function_that_logs():
    logger = logging.getLogger()
    logger.warning("This is a warning message")
    logger.error("An error has occurred!")
    logger.info("Completed operation successfully.")

def test_log_capturing(caplog):
    function_that_logs()
    log_messages = [record.message for record in caplog.records]
    assert "This is a warning message" in log_messages
    assert "An error has occurred!" in log_messages
    assert "Completed operation successfully." in log_messages
```

In this example, the function_that_logs() generates multiple log messages of different levels: WARNING, ERROR, and INFO.

Our test test_log_capturing() invokes function_that_logs(), captures the log messages into a list, and then verifies each of the expected log messages is present in the captured logs.

Note that caplog by default captures log messages of WARNING level or higher. So if you need to capture lower-level messages (e.g. INFO or DEBUG), you should set the logging level in caplog to a lower value before the function call.

It's important to note, that such tests as the example above, you are asserting that the messages are present somewhere in the log, regardless of their position. This is done by checking that the expected message is in the list of log messages, which will return True if the expected message is found anywhere in the list.

If you want to assert on the order of the log messages, then you would need to check the messages based on their index in the log_messages list. For example, if you want to check that the first logged message is "This is a warning message", you could do:

```python
assert log_messages[0] == "This is a warning message"
```

This will only pass if "This is a warning message" is the first message in the log. However, remember that relying on the order of log messages can make your test more brittle, as changes to the order of logging in the code would cause the test to fail. In general, it's better to check that the expected messages are present somewhere in the log, unless the order is particularly important.

That's about it, easy, right?
By using pytest's caplog fixture, you can create robust tests that not only verify your application's behavior, but also confirm it's producing the right log messages. This can be invaluable when diagnosing issues in production, as it ensures your logging is accurate.


Pytest's built-in caplog fixture to capture log messages is nothing but amazing. Capture logs for unit testing can't get easier

Capturing Logs with Pytest Can't Be Easier


I'm not really sure why, but for some reason I couldn't find many references on troubleshooting and setup NewRelic with FastAPI. I'm not sure if it's because FastAPI is still fairly new, or maybe it's because it's so simple to setup that no one really bothered to write about it, so I thought there's no harm on doing so.

When I started this task at work, I couldn't believe how intuitive and easy it was to integrate the NewRelic agent and getting amazing set of metrics and insights on my application without much effort.
Monitoring applications is a critical part of any software development cycle, and NewRelic provides an amazing Application Performance Monitoring (APM) tool for this purpose.

So here we're going to integrate NewRelic with FastAPI and Docker, allowing us to collect valuable data and monitor our applications more efficiently:

1. To start off, we need to install the NewRelic Python agent by running the following command:

```bash
pip install newrelic
```

2. Next, we need to import the NewRelic agent in our application file:

```python
import newrelic.agent
```

3. Set Up NewRelic Configuration;

There are two options here - Either download the `newrelic.ini` template file from NewRelic and fill in your valid New Relic license key.
You can find details about the configuration file [here](https://docs.newrelic.com/docs/apm/agents/python-agent/configuration/python-agent-configuration/#agent-configuration-file).

For Docker users, environment variables can be used as well, and this is actually my recommended way to do this. You just need to set `NEW_RELIC_APP_NAME` and `NEW_RELIC_LICENSE_KEY`.
Then pass in as an environment variable to your Docker container (either via Dockerfile or docker-compose.yml file).
On our case, we're fetching the license key from Vault using Terraform, which will be covered in a different blog post at some point.

4. Continue on and add an ENTRYPOINT for the NewRelic agent in your Dockerfile above the CMD command:

```dockerfile
ENTRYPOINT ["newrelic-admin", "run-program"]
```

If you're running the app without Docker, you can run your application with with additional command of newrelic agent, for example, here's how you can use it with gunicorn start command:

```bash
newrelic-admin run-program gunicorn app:app --config 'gunicorn.conf.py' --preload $*
```

5. To set up NewRelic, we initialize the agent and register our application:

I'd rather to wrap the two init agents commands with my own setup function, so I can pass in the application name as an argument:

```python
def setup_newrelic(app_name: str):
    newrelic.agent.initialize()
    newrelic.agent.register_application(name=app_name, timeout=10.0)
```

Then, we initialize the NewRelic agent in our gunicorn/Uuvicorn application:

```python
if __name__ == '__main__':
    setup_newrelic(app_name=f'{os.environ.get("NEW_RELIC_APP_NAME")}')
    uvicorn.run(app, host='0.0.0.0', port=int(os.environ.get('SERVER_PORT')))
```

By doing this, you will have enabled APM, which provides a wealth of useful metrics.

![APM-example](/images/apm.png)

6. Creating a Custom Event

Finally, to create a custom event in NewRelic, we can do the following:

```python
newrelic.agent.record_custom_event(event_name='ExampleEventName',
                value={'status_code': response.status_code},
                application=newrelic.agent.register_application())
```

The event_name must be a string, and value requires a dictionary with as many attributes as you wish to include.

Here's an example of a custom metric I created that helped me debug a large application where I wanted to get all exceptions types:

![custom-metric](/images/custom_metric_nr.png)

That's it! You now have NewRelic integrated with your FastAPI application running in a Docker environment. The possibilities for monitoring and analyzing your application are now significantly broader and more robust, helping you ensure that your application is always running at its best.


Enhance your FastAPI application's monitoring with NewRelic. Gain insights and improve performance with very minimal and simple setup steps.

Integrating NewRelic with FastAPI & Docker


I recently encountered an intriguing issue at my workplace, which inspired me to pen down this blog post. I was utilizing a Python package built by a team different from mine. This particular scenario arose when I was employing this package in one of my functions, and in order to catch any exceptions raised by the package, I wrote a try-except block. To augment the code readability, I inserted a logging statement within this block, as illustrated below:

```python
try:
    # do something
    log.info(f"Looking for faulty devices, found: {len(faulty_devices)} devices")
    for device in faulty_devices:
        log.info(f"Device: {device['deviceId']} has faulty status: {device['status']}, sending a message to MQ")
        await messages.send_message_to_queue(message=device) #one of the internal package methods
except Exception as err:
    log.error(f"An error occurred while getting data from devices or while sending message to MQ: {err}")
```

However, upon testing an edge case that would steer the flow of control into the except block, the resultant log message was surprising to me from the expected error details.
The log just printed something like:

`An error occurred while getting data from devices or while sending message to MQ:` The exception was missing from the log message.

Looking at the source code of the package, you could see the issue immediately (below code is merely a representative example and not the actual code):

```python
from http import HTTPStatus


class APError(Exception):
    """ Base class for all AP errors """
    CODE: int
    DESCRIPTION: str
    REASON: int

    def __init__(self, *args, **kwargs):
        self.extra = kwargs
        super().__init__(*args)

    def __repr__(self) -> str:
        params = [repr(arg) for arg in self.args]
        for key, value in self.extra.items():
            params.append(f'{key}={repr(value)}')
        return f'{self.__class__.__name__}({", ".join(params)})'


class GeneralSQSError(APError):
    """ General error related to SQS """

    CODE = HTTPStatus.BAD_REQUEST
    DESCRIPTION = 'Error returned from SQS service'


class SQSQueueDoesNotExistError(APError):
    """ The SQS queue requested does not exist """

    CODE = HTTPStatus.NOT_FOUND
    DESCRIPTION = 'Queue does not exist'
```

So why it's related the log output?

When an exception is raised, you can directly print the exception `err` to get its string representation. By default, this calls the `__str__` method of the exception object. If this method is not overridden in the exception class, Python uses the `__str__` method of the base Exception class, which by default returns the arguments passed to the exception when it was raised.

In this case of custom exceptions, we have overridden the `__repr__` method but not the `__str__` method.
`__repr__` is used to provide a "string representation" of the object for debugging purposes and is not the same as `__str__`. While `__repr__` is used when you call repr(object), `__str__` is used when you call str(object) or when you use the object in a string context (like f-string in our example).

If you want your exceptions to provide a useful message when used in a string context, you should override the `__str__` method.

So here are a few common practices for creating custom exceptions in Python:

## Inheritance from Python's built-in exceptions

Whenever you create a custom exception, you should make it inherit from Python's built-in exceptions. The most common practice is to inherit from the Exception class or from a more specific exception class if it fits your case.

```python
class CustomError(Exception):
    """Base class for other exceptions"""
    pass
```

## Use meaningful names

Like all Python identifiers, exception names should be as descriptive as possible.

## Add useful error messages

The constructor of your custom exception should be defined to accept a custom error message as a parameter. This message can then be passed to the base class constructor using the `super()` function. This way, you can provide more specific details about what went wrong.

```python
class CustomError(Exception):
    """Base class for other exceptions"""
    def __init__(self, message="A custom exception occurred"):
        self.message = message
        super().__init__(self.message)
```

## Add additional context

Sometimes, an error message alone is not enough. If there are important variables or state information that would help diagnose the problem, consider adding these to your exception.

```python
class CustomError(Exception):
    """Base class for other exceptions"""
    def __init__(self, message="A custom exception occurred", context=None):
        self.message = message
        self.context = context
        super().__init__(self.message)
```

## Document your exceptions

If other developers (or future you) are going to use your code, you should document your exceptions.
What conditions cause them to be raised? What do they signify?

Here's a full example of a custom exception in Python:

```python
class InsufficientBalanceError(Exception):
    """Raised when the balance is insufficient for a transaction"""
    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount
        message = f"Your balance of {balance} is insufficient for a transaction of {amount}."
        super().__init__(message)
```

You can use it like this:

```python
def withdraw(balance, amount):
    if balance < amount:
        raise InsufficientBalanceError(balance, amount)
    return balance - amount
try:
    withdraw(50, 100)
except InsufficientBalanceError as e:
    print(str(e)) # Your balance of 50 is insufficient for a transaction of 100.
```

## Define special builtin methods

Defining special methods like `__str__` for your exceptions can be very useful in some cases.

```python
class InsufficientBalanceError(Exception):
    """Raised when the balance is insufficient for a transaction"""
    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount

    def __str__(self):
        return f"Your balance of {self.balance} is insufficient for a transaction of {self.amount}."
```

Now, the `__str__` method is defined inside the InsufficientBalanceError class, so when you print the exception object, Python will print whatever the `__str__` method returns.

Remember that the `__str__` method should always return a string. It is called by built-in functions and operators that convert the object into a string, such as `print()` and `str()`.

There additional methods that are sometime being overridden in custom exceptions, like `__repr__` for example.

if you want to customize how your exceptions are displayed when using the `repr()` function, you'll need to define `__repr__` method as well in your exception class.

for example:

```python
class InsufficientBalanceError(Exception):
    """Raised when the balance is insufficient for a transaction"""
    def __init__(self, balance, amount):
        self.balance = balance
        self.amount = amount

    def __str__(self):
        return f"Your balance of {self.balance} is insufficient for a transaction of {self.amount}."

    def __repr__(self):
        return f"InsufficientBalanceError(balance={self.balance}, amount={self.amount})"
```

In this example, `__repr__` returns a string that shows how you would create the object, which is a common convention for **repr**.

To sum up,

When creating custom exceptions in Python, there are several common practices to consider. It's crucial to inherit from Python's built-in exceptions, usually from the Exception class.
Naming your exceptions clearly and descriptively can make your code easier to understand.
Moreover, it's a good idea to add useful error messages to your exceptions and pass any relevant context that could help diagnose the problem.
Documenting what conditions cause your exceptions to be raised and what they signify is another beneficial practice.

Custom exceptions can be further enhanced with the implementation of Python's special methods. The `__str__` method, for example, can be used to provide a human-readable string representation of the exception, which is returned when calling `str()` or `print()` on the exception object. The `__repr__` method provides a string that could be used to recreate the object, typically used for debugging and development. This is invoked when calling `repr()` on the object.

By following these guidelines, you can make your custom exceptions more meaningful and informative, aiding in the debugging process and making your code easier to use and understand.


Boost your Python code clarity and debuggability with custom exceptions. Explore best practices for crafting and handling them effectively

Crafting Meaningful Custom Exceptions in Python: Best Practices and Common Techniques


During development of a monitoring project at work, I encountered a need to regularly sample APIs from a third-party tool we use, then reformat the responses to suit our requirements before sending the data to our monitoring infrastructure tools (VictoriaMetrics & Grafana). As the project progressed, it became apparent that I needed to schedule each function based on its specific scope and at varying intervals. The `schedule` library in Python offered a straightforward solution to this challenge.

Python's `schedule` library is an in-process job scheduler that uses the builder pattern for configuration. It allows you to run Python functions (or any other callable) periodically at pre-determined intervals using very a simple and a human-friendly syntax.

This post aims to discuss the advantages and disadvantages of this library, provide practical examples, and share a real-world use case from a recent project.

I should state early on, that this is not a perfect solution for all type of "tasks" in Python. I believe that this covers quite well the more simpler tasks.
Moving on from the overview, let's take a look a little bit on the advantages and disadvantages found during the research of the development process of our project.

## Advantages

### Simplicity

One of the main benefits of the schedule library is its simplicity. It uses an easy-to-understand syntax that makes it accessible even to Python beginners. With a few lines of code, you can set up tasks to be run at any desired interval.

### Flexibility

With schedule, you can set up jobs to run at any interval using any Python function. This flexibility allows you to tailor how and when your jobs run to best suit your needs. Whether you want a function to run every five minutes, once a day, or every Monday at noon, schedule has you covered.

### In-process Scheduling

Another key advantage of schedule is that it enables you to manage all of your scheduled tasks within your Python process. This feature can simplify both your application architecture and deployment process, as there is no need for an external system to handle your tasks.

## Disadvantages

### Not Suitable for Large-scale/Distributed Systems

While schedule is a powerful tool for managing tasks in a single Python process, it does have limitations. For instance, it does not support features such as distributed scheduling or persistence of schedules across restarts. Consequently, it may not be the best fit for larger, distributed systems, or for situations where you need your schedules to persist across restarts.

### No Built-in Error Handling

The schedule library also lacks built-in error handling or job failure notifications. While it does allow error handling within jobs, it's up to the developer to ensure proper error handling is in place.

## Real-world Example: Our Monitoring Service

In a recent project, I used schedule to build a monitoring service that sends metrics to a time-series database. The service obtains most of its information from a third-party tool and, therefore, requires functions to run at scheduled intervals.

Here's how `schedule` was employed in the service:

```python
@repeat(every(config.DEVICE_STATUS_SAMPLING_INTERVAL).seconds)
def metric_data_devices():
    # function implementation...
```

In this function, `schedule` is used to run the function every X seconds, according to an environment variable comes from our deployment infrastructure code (Terraform).

Of course, there are many other built-in intervals at different formats interval (hours, minutes, days, etc.)

Here's an example from the [schedule library official docs](https://schedule.readthedocs.io/en/stable/), but I recommend exploring it further as I encountered many cool features that might be helpful for such task.

```python
import schedule
import time

def job():
    print("I'm working...")

schedule.every(10).minutes.do(job) # every 10 minutes
schedule.every().hour.do(job) # every hour
schedule.every().day.at("10:30").do(job) # once a day at a specific hour
schedule.every().monday.do(job) # every specific day in a week
schedule.every().wednesday.at("13:15").do(job) # every specific day in a week in a specific time
schedule.every().day.at("12:42", "Europe/Amsterdam").do(job) # every day in a specific time of a specific timezone

while True:
    schedule.run_pending()
    time.sleep(1)
```


Dive into the schedule library through practical examples & a real-life project scenario. Discover how to schedule your Python methods effectively

Schedule Your Python Methods with Schedule Library


This all started when I got a task at work to do a research about a the best way to move on from the deprecated method fo python packaging (setup.py, manifest.ini, etc) to the new and standard way using `pyproject.toml`.
From this simple task - I was able to convince my team to move on also from pip-tools and move on to [Poetry](https://python-poetry.org/).

## So What is Poetry?

In short, Poetry is a relatively new dependency management and packaging tool for Python.
It aims to simplify and streamline the process of managing dependencies, packaging, and publishing Python projects.
Poetry focuses on simplicity, ease of use, and reproducibility of environments.

Here's some of its capabilities:

- Poetry uses a `pyproject.toml` file to manage project metadata and dependencies, following the [PEP 517](https://peps.python.org/pep-0517/)/[PEP 518](https://peps.python.org/pep-0518/) standards.

- It provides features like dependency resolution, virtual environments, lock files, build systems, and publishing capabilities.

- It supports direct installation of dependencies from various sources like PyPI, Git, HTTPS, and local files.

- Poetry integrates well with the packaging and distribution tools like twine.

- Poetry generates a lock file for dependencies (poetry.lock).

- Supports extra-index-url similar to pip (secondary artifactory source for python packages).

- Supports dependency installation per “groups” such as dev-dependencies.

## Why Poetry Then?

Poetry provides a streamlined and user-friendly experience for managing dependencies, packaging, and publishing Python projects.
Poetry makes it straightforward to define and manage dependencies using the pyproject.toml file, which is the standard method to handle python packages at the moment.
It handles dependency resolution efficiently, ensuring that you have reproducible environments for both development and deployment.
Poetry also offers seamless integration with popular packaging and distribution tools like twine, making it convenient to publish your projects to PyPI or other repositories.

Additionally, Poetry's virtual environments and lock file mechanism help ensure that your project dependencies are isolated, consistent, and reproducible across different environments. The lock file, poetry.lock, guarantees that everyone working on the project will have the same versions of the dependencies installed, avoiding version conflicts (very similar to package-lock.json in JS).

With its comprehensive set of features, extensive documentation, and an active community, Poetry provides a robust and user-friendly solution for managing Python dependencies, making it an excellent choice for modern Python development projects.

So let's move to some usage example (that can be also be found easily it the great Poetry documentation btw).

To start a new python package/project -

```bash
poetry new <project-name>
```

This will generate some basic files and directories for your project:

```bash
poetry-demo
├── pyproject.toml
├── README.md
├── poetry_demo
│   └── __init__.py
└── tests
    └── __init__.py
```

The `pyproject.toml` file is what is the most important here. This will orchestrate your project and its dependencies. For now, it looks like this (example from one of my repos):

```toml
[tool.poetry]
name = "poetry-demo"
version = "0.1.0"
description = ""
authors = ["Sébastien Eustace <sebastien@eustace.io>"]
readme = "README.md"
packages = [{include = "poetry_demo"}]

[tool.poetry.dependencies]
python = "^3.11"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
```

Poetry assumes your package contains a package with the same name as tool.poetry.name located in the root of your project. If this is not the case, populate tool.poetry.packages to specify your packages and their locations.

Similarly, the traditional MANIFEST.in file is replaced by the tool.poetry.readme, tool.poetry.include, and tool.poetry.exclude sections. tool.poetry.exclude is additionally implicitly populated by your .gitignore.

Poetry will require you to explicitly specify what versions of Python you intend to support, and its universal locking will guarantee that your project is installable (and all dependencies claim support for) all supported Python versions.

To initialize an already existing project -

```bash
poetry init
```

To add a new dependency -

```bash
poetry add requests
```

Or with options just like with pip -

```bash
poetry add "fastapi[all]"
```

this for example will add these libraries to the pyproject.toml

```toml
[tool.poetry.dependencies]
python = "^3.11"
fastapi = {extras = ["all"], version = "^0.95.1"}
requests = "^2.30.0"
```

To remove a dependency -

```bash
poetry remove pytest-asyncio
```

Add a new dependency to “dev dependencies” -

```bash
poetry add pytest --group dev
```

This will add it under dev group deps -

```toml
[tool.poetry.group.dev.dependencies]
pytest = "^7.3.1"
flake8 = "^6.0.0"
pytest-mock = "^3.10.0"
```

Add another source for internal packages -

```bash
poetry source add --secondary <name> <url>
poetry config http-basic.<name> <username> <password/token>
```

Activate virtual environment -

```bash
poetry shell
```

Install project’s dependencies -

```bash
poetry install
```

And of course many other options you can find in [Poetry's documentation](https://python-poetry.org/docs/).


Poetry: Simplify Python dependency management with an intuitive CLI, virtual environments, and seamless packaging & publishing

Python Packaging & Dependency Management with Poetry


When working with JSON in Python, you'll likely come across the built-in json module, which provides easy-to-use methods for encoding and decoding JSON data.

In this blog post, we'll discuss the differences between json.load/json.loads and json.dump/json.dumps methods, and provide examples of how to use each method.

Why? It's indeed quite basic, but it can be a bit confusing when getting started with Python and while using the json module.

### json.load and json.loads

`json.load` and `json.loads` are used to decode JSON data. While both methods serve the same purpose, they differ in the type of input they accept:

- json.load: Decodes a JSON object from a file-like object (e.g., a file or a StringIO object).
- json.loads: Decodes a JSON object from a string.

Let's look at examples for each method:

```python
import json

with open("example.json", "r") as file:
    data = json.load(file)

print(data)
```

Here, example.json is a file containing a JSON object, and `json.load` is used to read and decode the JSON data into a Python object (e.g., dictionary or list).

```python
import json

json_string = '{"name": "Alice", "age": 30, "city": "New York"}'
data = json.loads(json_string)

print(data)
```

In this example, `json.loads` is used to decode a JSON-formatted string into a Python dictionary.

### json.dump and json.dumps

`json.dump` and `json.dumps` are used to encode Python objects into JSON format. Similar to `json.load` and `json.loads`, these methods differ in the type of output they produce:

- json.dump: Encodes a Python object and writes it to a file-like object.
- json.dumps: Encodes a Python object and returns a JSON-formatted string.

Examples for each method:

```python
import json

data = {"name": "Alice", "age": 30, "city": "New York"}

with open("output.json", "w") as file:
    json.dump(data, file)
```

In this example, `json.dump` is used to write the JSON representation of a Python dictionary to a file named output.json.

```python
import json

data = {"name": "Alice", "age": 30, "city": "New York"}

json_string = json.dumps(data)
print(json_string)
```

Here, `json.dumps` is used to convert a Python dictionary into a JSON-formatted string.

Now that we've covered the basic usage of `json.load`, `json.loads`, `json.dump`, and `json.dumps`, let's take a look at a "real-world production" example using these methods. In this example, we'll read JSON data from a RESTful API, modify the data, and save it to a file.

Imagine you're working with a weather API that returns a JSON object containing weather data for a given location. Here's how you might use `requests` library to get the data, `json.loads` to parse it, and `json.dump` to save it:

```python
import json
import requests

# Fetch weather data from a RESTful API
url = "https://api.example.com/weather?location=New%20York"
response = requests.get(url)

# Parse JSON data using json.loads
weather_data = json.loads(response.text)

# Modify the data (e.g., convert temperature to Celsius)
weather_data['temperature'] = (weather_data['temperature'] - 32) * 5/9

# Save modified data to a file using json.dump
with open("weather_data.json", "w") as file:
    json.dump(weather_data, file)
```

In this example, we first fetch weather data from a RESTful API using the `requests` library. The API returns a JSON-formatted string, which we parse into a Python dictionary using `json.loads`. Next, we modify the temperature value by converting it from Fahrenheit to Celsius. Finally, we save the modified weather data to a file named weather_data.json using `json.dump`.

This example demonstrates how you might use json.load, json.loads, json.dump, and json.dumps in a production environment when working with JSON data from various sources, such as APIs, files, or strings.


Dive into the differences between Python's json module methods: json.load/json.loads & json.dump/json.dumps

Exploring JSON in Python: load/loads and dump/dumps


Ok, I know it's kind of basic stuff right here, but this was one of the first bugs I event created and I felt I have to share it.

In Python, there are several ways to check whether a variable is a specific type. In my example i'll use the dictionary object.

While these ways may seem interchangeable, they actually have some important differences that can affect your code's performance and maintainability. Let's take a closer look at each of these methods.

### Using `isinstance()`

The `isinstance()` function is a built-in Python function that takes two arguments: an object and a class (or tuple of classes). It returns True if the object is an instance of the specified class, and False otherwise.

For example, to check whether a variable my_dict is a dictionary, you can use the following code:

```python
my_dict = {}

if isinstance(my_dict, dict):
    # do something with my_dict
```

One key difference between `isinstance()` and other methods is their behavior with subclassing. `isinstance()` returns True not only for instances of the specified class, but also for instances of any subclass of that class.

For example, consider the following code:

```python
class MyDict(dict):
    pass

my_dict = MyDict()

print(isinstance(my_dict, dict))  # True
print(type(my_dict) is dict)      # False
```

Here, my_dict is an instance of a subclass of dict called MyDict. `isinstance()` returns True because MyDict is a subclass of dict, whereas `type()` returns False because MyDict is not the same as dict.

### Using `type()`

The `type()` function is also a built-in Python function that takes one argument: an object.
It returns the type of the object as a Python class.

For example, to check whether a variable my_dict is a dictionary, you can use the following code:

```python
my_dict = {}

if type(my_dict) is dict:
    # do something with my_dict
```

Another important difference between these methods is their performance. In general, `isinstance()` is slower than `type()` because it has to check the object's class hierarchy to determine whether it is an instance of the specified class or subclass.
Therefore, if you know that an object is a dictionary and you just need to check its type, using `type()` may be faster.

### Using `is` dict

In Python, `is` dict is a keyword that checks whether an object is a dictionary object. It works similarly to the `type()` function, but it is more efficient because it does not have to look up the object's class hierarchy. If an object is a dictionary, `is` dict will return True, otherwise, it will return False.

For example, consider the following code:

```python
my_dict = {}

print(my_dict is dict)  # True
```

But there comes the old nostalgic bug I had and talked about before:

```python
my_dict = {'key1': 'value1', 'key2': 'value2'}

print(my_dict is dict)  # False
```

In this code, my_dict is an instance of the dict class, so my_dict is dict returns False.

When we use the `is` operator to compare my_dict with dict, Python checks if my_dict and dict refer to the same object in memory. However, my_dict and dict are not the same object in memory, even though they have the same type.

In other words, dict `is` a built-in Python class, while my_dict is an instance of that class. The `is` operator checks for identity, which means it checks whether two objects are the same object in memory. In this case, my_dict and dict are not the same object, so the expression my_dict is dict returns False.

### Conclusion

There are several ways to check whether a variable is a dictionary object in Python, including `isinstance()`, `type()`, and `is` dict. Each of these methods has some important differences that can affect your code's behavior and performance.

If you want to check whether an object is an instance of a certain class or any of its subclasses, you should use `isinstance()`. However, keep in mind that `isinstance()` may be slower than `type()` because it has to check the object's class hierarchy.

If you want to check whether an object is exactly of a certain type, use `type()`. This method is generally faster than `isinstance()` because it doesn't have to look up the object's class hierarchy.

If you just need to check whether an object is a dictionary, you can use `is` dict. This keyword is more efficient than using `type()` and can quickly determine the type of an object in your code. However, keep in mind that `is` dict checks specifically for dictionary objects, and not for subclasses of dict.

In summary, the best method to use depends on your specific use case. Choose the method that best fits your needs and keep in mind the differences between them. Don't create stupid bugs like I did :)


Type checking is a thing you do a lot in Python as it can help prevent simple yet frustrating errors that may arise from incorrect variable types. In fact, it was one of the earliest and most lame bug I encountered in my programming journey

Different Ways to Check a Variable Type in Python


Pytest is a powerful and flexible testing framework for Python that allows developers to easily write and execute test cases for their code. One of the key features of pytest is the ability to use custom markers to organize and categorize tests in efficient way.

Custom markers in pytest are user-defined labels that can be attached to individual tests, test classes, or entire test suites. These markers can be used to provide additional context and information about the purpose and intent of each test, which can help developers better understand the test results and identify issues more quickly.
For those of us who had the chance to write down manual tests - You should be familiar with the above.

There are several advantages of using pytest custom markers in your testing workflow:

**Improved organization and readability:** Custom markers can help organize tests into logical categories, making it easier to navigate and understand complex test suites. For example, you might use markers to group tests by feature, priority, or level of complexity.

**Flexible test selection:** With custom markers, you can selectively run subsets of tests based on their markers. This can be especially useful when you have a large test suite and want to focus on a specific set of tests, such as those related to a particular feature or module.

In a small Test Automation project I had a chance to take part in, at some point we wanted to differentiate between tests that will be executed on each uploaded Pull request, and for heavier test to run in a nightly scheduled job.
Pytest custom markers was such an easy solution.

In your pytest.ini file just use the markers tag and create as many markers as you like.

```python
[pytest]
markers =
    sanity:Run only tests marked as sanity
```

From there you can mark the tests you'd like to tag with the pytest mark decorator.
This is a small test from my project as an example:

```python
@pytest.mark.sanity
def test_leave_room_sanity(self, get_default_parameters, room_id_private_after_join):
    ec_api.leave_room(room_id=room_id_private_after_join, **get_default_parameters)
```

Now if you want to run only the tests with the custom marker you created, simply run pytest command with -m flag and the marker name.

```bash
pytest -m sanity
```

You can also run all the tests that aren't marked with the marker you created by running a reverse command.

```bash
pytest -m "not sanity"
```

There are many more advantages we can leverage using custom markers, such as:

**Customizable reporting**: Pytest's reporting system can be customized to display information based on the markers attached to each test. This allows you to generate more detailed reports that provide additional context and insights into the test results.

**Integration with other tools**: Custom markers can be used to integrate pytest with other tools and systems in your development workflow. For example, you might use markers to trigger automated deployments or trigger additional testing based on the results of certain tests.

Overall, custom markers are a powerful feature of pytest that can help developers write more organized, readable, and flexible tests. By using markers to categorize and label tests, developers can better understand their test results and identify issues more quickly, leading to a more efficient and effective testing process.


This is the easiest way to categorize and maximize the flexibility and readability of your testing workflow and results

Organizing Your Tests with Custom Markers in Pytest


Let's begin with a brief, high-level explanation of middleware. In software development, middleware refers to a helper component that connects different parts of an application. It commonly acts as a mediator between a client and server, facilitating communication and data exchange.

Additionally, middleware can provide various services, including authentication, logging, caching, and more.

![middleware](/images/middleware.png)

Middleware can simplify many tasks in web development, such as:

- Handling authentication and authorization
- Parsing and validating user input
- Serving static files and compressing responses
- Logging and error handling
- Adding security measures to protect against attacks
- Caching
  And much more.

In a specific use case, a microservice had a single endpoint for AWS health checks. The aim was to add logging and custom metrics reported to [New Relic](https://newrelic.com/) without cluttering the route method.
To achieve this, middleware was utilized instead of embedding the additional functionality in the route method.

Let's take a look:

**Route method (HEAD method)**

```python
from fastapi import FastAPI

app = FastAPI()


@app.head("/api/v1/health") # This will return 200 in case of successful request
async def health_check():
    return ""
```

**Middleware**

```python
from fastapi import Request


@app.middleware("http")
async def response_middleware(request: Request, call_next):
    response = await call_next(request)
    log.info(f"Healthcheck endpoint response: {response.status_code}") # Log the result
    newrelic_custom_event('HealthCheckEndpoint', {'status_code': response.status_code}) # Method that will send a metric with the status code response
    return response
```

Using middleware instead of adding all the extra logging and New Relic metric in the route method is a better approach because it helps to keep the route method clean and focused on its main purpose of handling HTTP requests and responses.
By using middleware, the additional functionality such as logging and metrics can be added to the request/response pipeline in a centralized way, without cluttering the route method with extraneous code. This example might be very minimal but I hope you get the point regardless.

In addition to the above reason, this approach can also improve code maintainability and reusability since the middleware can be reused across multiple routes, rather than having to duplicate the logging and metric code in every route method.


I try to keep my routes as "clean" as i can, additional logic will be executed via middleware - Here's a small example I did with FastAPI

Python Articles

All Python Articles