How to write parametrized tests in Python with pytest πŸŽ₯

Hi πŸ‘‹

Welcome to another video tutorial on how to write parametrized tests in Python using pytest.

If you want to follow along, here’s the code that I’ve tested in the video.

from typing import List


class Solution:
    def move_zeroes(self, nums: List[int]) -> None:
        last_zero = 0
        index = 0
        while index < len(nums):
            if nums[index] != 0:
                nums[last_zero], nums[index] = nums[index], nums[last_zero]
                last_zero += 1
            index += 1


def main():
    solution = Solution()
    arr = [1,0,1]
    solution.move_zeroes(arr)
    print(arr)


if __name__ == '__main__':
    main()

Thanks for watching! πŸ˜„

Testing Python projects with Tox

Hi πŸ‘‹

In this article I will show you how to test your Python projects with Tox.

Introduction

Tox is a tool for automating testing in Python, their vision is to standardize the testing process. It can be used to easily test your project using multiple Python interpreters and run various commands.

Getting Started

To get started all you need to add to your project is a tox.ini file. To simplify running the tests we will make use of the following Dockerfile, which contains Python interpreters for 3.6 and 3.7

FROM ubuntu:20.04

RUN apt update && apt install -y software-properties-common \
               && add-apt-repository ppa:deadsnakes/ppa \
               && apt install -y python3.6 && apt install -y python3.7 \
               && apt install -y python3-pip && pip3 install tox

VOLUME /code

WORKDIR /code
ENTRYPOINT tox

A tox.ini file which tests using python 3.6 and python 3.7 looks like this:

# content of: tox.ini , put in same dir as setup.py
[tox]
skip_missing_interpreters = True
envlist = py36,py37

[testenv]
# install pytest in the virtualenv where commands will be executed
deps =
    pytest==6.2.1
    pytest-cov==2.11.1
    responses==0.13.3
commands =
    # NOTE: you can run any command line tool here – not just tests
    pytest

[testenv:bamboo]
commands =
  pytest β€”junitxml=results.xml \
    β€”cov=your-moduleβ€”cov-config=tox.ini β€”cov-report=xml
    coverage2clover -i coverage.xml -o clover.xml
deps =
    {[testenv]deps}
    coverage2clover

We have two environments: testenv and testenv:bamboo, the later one being used for coverage reporting in Bamboo using clover. To run Tox with a specific environment you’d type tox -e bamboo.

To run the test via the Dockerfile, first you’d build the docker container using: docker build . -f Dockerfile -t tox

Then, you’d run the container with docker run -v “$(pwd)”:”/code” tox -e bamboo to test with the Bamboo environment or just docker run -v “$(pwd)”:”/code” tox for the default env.

Practical Example

Here’s an example that you can use to follow along. We have the following files:

@denis ➜ tox_article ls
__pycache__  tests.py  tox.ini
# @denis ➜ tox_article cat tests.py
import unittest

class TestStringMethods(unittest.TestCase):

    def test_upper(self):
        self.assertEqual('foo'.upper(), 'FOO')

    def test_isupper(self):
        self.assertTrue('FOO'.isupper())
        self.assertFalse('Foo'.isupper())

    def test_split(self):
        s = 'hello world'
        self.assertEqual(s.split(), ['hello', 'world'])
        # check that s.split fails when the separator is not a string
        with self.assertRaises(TypeError):
            s.split(2)

if __name__ == '__main__':
    unittest.main()%
# @denis ➜ tox_article cat tox.ini
[tox]
skip_missing_interpreters = True
envlist = py36,py37
skipsdist = True

[testenv]
commands =
    python -m unittest%

Running Tox in our docker image will yield the following output:

@denis ➜ tox_article docker run -v "$(pwd)":"/code" tox
py36 create: /code/.tox/py36
py36 run-test-pre: PYTHONHASHSEED='520882151'
py36 run-test: commands[0] | python -m unittest
...
----------------------------------------------------------------------
Ran 3 tests in 0.000s

OK
py37 create: /code/.tox/py37
py37 run-test-pre: PYTHONHASHSEED='520882151'
py37 run-test: commands[0] | python -m unittest
...
----------------------------------------------------------------------
Ran 3 tests in 0.000s

OK
___________________________________ summary ____________________________________
  py36: commands succeeded
  py37: commands succeeded
  congratulations πŸ™‚

The same tests are run twice, first with Python 3.6 and then with Python 3.7.

Thanks for reading and happy testing! πŸ”§

FastAPI Uvicorn logging in Production

Hello πŸ™‹β€β™‚οΈ,

Running a ⏩FastAPI ⏩ application in production is very easy and fast, but along the way some Uvicorn logs are lost.

In this article I will discuss how to write a custom UvicornWorker and to centralize your logging configuration into a single file.

To keep things as simple as possible I’ve put all my code in a single Python file.

main.py

import uvicorn as uvicorn
from fastapi import FastAPI, APIRouter

router = APIRouter(prefix="")


def create_app():
    fast_app = FastAPI()
    fast_app.include_router(router)
    return fast_app


@router.get("/")
def read_root():
    return {"Hello": "World"}


if __name__ == '__main__':
    app = create_app()
    uvicorn.run(app=app)

Running the code will return a {"Hello": "World"} json when you visit the root endpoint / at http://127.0.0.1:8000. 😁

When you check the console window, the following log lines are printed:

INFO:     Started server process [10276]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:53491 - "GET / HTTP/1.1" 200 OK

Notice the Uvicorn log GET / HTTP/1.1″ 200 OK.

According to Uvicorn’s deployment docs we should run Uvicorn in a production settings with the following command: gunicorn -k uvicorn.workers.UvicornWorker main:create_app.

(venv2) ➜  FastAPILogging gunicorn -k uvicorn.workers.UvicornWorker main:create_app
[2021-05-17 22:10:44 +0300] [6250] [INFO] Starting gunicorn 20.1.0
[2021-05-17 22:10:44 +0300] [6250] [INFO] Listening at: http://127.0.0.1:8000 (6250)
[2021-05-17 22:10:44 +0300] [6250] [INFO] Using worker: uvicorn.workers.UvicornWorker
[2021-05-17 22:10:44 +0300] [6252] [INFO] Booting worker with pid: 6252
[2021-05-17 22:10:45 +0300] [6252] [WARNING] ASGI app factory detected. Using it, but please consider setting the --factory flag explicitly.
[2021-05-17 22:10:45 +0300] [6252] [INFO] Started server process [6252]
[2021-05-17 22:10:45 +0300] [6252] [INFO] Waiting for application startup.
[2021-05-17 22:10:45 +0300] [6252] [INFO] Application startup complete.

Now, if we visit the root endpoint, the console won’t print “GET / HTTP/1.1” 200 OK anymore/ πŸ€¦β€β™‚οΈ.

To fix it we need a custom UvicornWorker βš™ and a logging configuration file πŸ—ƒ.

Create a new file and name it logging.yaml, then paste the following contents in it:

version: 1
disable_existing_loggers: false

formatters:
standard:
format: "%(asctime)s - %(levelname)s - %(message)s"

handlers:
console:
class: logging.StreamHandler
formatter: standard
stream: ext://sys.stdout

loggers:
uvicorn:
error:
propagate: true

root:
level: INFO
handlers: [console]
propagate: no

This file will configure our root logger and our Uvicorn logger. To read more on the topic please see Python logging configuration.

Next, we will create a custom UvicornWorker class that will set log_config to the path of our logging.yaml file, to pass the logging configuration that we’ve just made to Uvicorn. πŸ¦„

I added the following code in main.py:

class MyUvicornWorker(UvicornWorker):
    CONFIG_KWARGS = {
        "log_config": "/mnt/c/Users/denis/PycharmProjects/FastAPILogging/logging.yaml",
    }

β–Ά If we run the application with:

gunicorn -k main.MyUvicornWorker main:create_app

We should see the Uvicorn access logs printed in the console πŸ¦„

(venv2) ➜  FastAPILogging gunicorn -k main.MyUvicornWorker main:create_app
[2021-05-17 22:31:28 +0300] [6278] [INFO] Starting gunicorn 20.1.0
[2021-05-17 22:31:28 +0300] [6278] [INFO] Listening at: http://127.0.0.1:8000 (6278)
[2021-05-17 22:31:28 +0300] [6278] [INFO] Using worker: main.MyUvicornWorker
[2021-05-17 22:31:28 +0300] [6280] [INFO] Booting worker with pid: 6280
2021-05-17 22:31:28,185 - WARNING - ASGI app factory detected. Using it, but please consider setting the --factory flag explicitly.
2021-05-17 22:31:28,185 - INFO - Started server process [6280]
2021-05-17 22:31:28,185 - INFO - Waiting for application startup.
2021-05-17 22:31:28,185 - INFO - Application startup complete.
2021-05-17 22:31:30,129 - INFO - 127.0.0.1:54004 - "GET / HTTP/1.1" 200

Thanks for reading! πŸ“š

requirements.txt

click==7.1.2
fastapi==0.65.1
gunicorn==20.1.0
h11==0.12.0
httptools==0.2.0
pydantic==1.8.2
PyYAML==5.4.1
starlette==0.14.2
typing-extensions==3.10.0.0
uvicorn==0.13.4
uvloop==0.15.2

Context Managers and Cross Cutting concerns in Python

Hello,

In this short article I would like to talk about context managers. I personally consider that at the core they are just a form of decorators. If you don’t know what a decorator is check the Decorator Pattern Wikipedia article.

Decorators can be used to implement cross-cutting concerns. We have componentA and we need logging and security, we could write the logic for logging and security handling in componentA but some people consider component a should be componentA not componentAthatAlsoKnowsAboutSecurityAndOtherStuff. Since it’s not the component’s responsibility to authorize requests or log calls to a external logging service, we can wrap the componentA into a decorator that does just that.

A formal definition for cross-cutting concerns as taken from Wikipedia is the following:

In aspect-oriented software development, cross-cutting concerns are aspects of a program that affect other concerns. These concerns often cannot be cleanly decomposed from the rest of the system in both the design and implementation, and can result in either scattering (code duplication), tangling (significant dependencies between systems), or both.

And some examples of cross cutting concerns include:

Since the context managers are sort of similar to decorators you can use them to implement cross cutting concerns. Let’s explore.

Simple Example

In Python you can have two types of context managers: a function and a class. In order for the function to behave like a context manager it will need to be decorated with the @contextmanager decorator, and in order for a class behave like a context manager it needs to implement __enter__ and __exit__.

Context managers can be called using the with statement. The following code snippet demonstrates two context managers:

  • One that logs when the function is called and when it exits.
  • One that intercepts the function arguments.
from contextlib import contextmanager

@contextmanager
def simple_context_manager(function):
    try:
        print("calling function")
        yield function
    finally:
        print("function call has ended")

class SimpleContextManager:
    def __init__(self, cb):
        self.cb = cb

    def _intercept(self, *args, **kwargs):
        print(f"calling with {args} {kwargs}")
        return print(*args, **kwargs)

    def __enter__(self):
        print("intercept start")
        return self._intercept

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("intercept end")

def main():
    with simple_context_manager(print) as print_func:
        print_func("hi")

    with SimpleContextManager(print) as print_func:
        print_func("hi")
        print_func("hi", end="\n\n", sep=",")
        print_func("hi")

if __name__ == '__main__':
    main()

Caching

What is caching? In short..

Caching is used to store the result of an expensive computation somewhere in memory or on a persistent storage device in order to optimize the program.

We have the compute_fibonacci function, which is quite slow. A version that uses cache has been implementing in the CachedComputeFibonacci class. Notice how the code takes some time to output the result for the first call of print(cached_compute_fibonacci(35)) statement but the second print in instant.

def compute_fibonacci(number):
    if number <= 1:
        return number
    return compute_fibonacci(number-1) + compute_fibonacci(number-2)


class CachedComputeFibonacci:
    def __init__(self):
        self._cache = {}

    def __call__(self, *args, **kwargs):
        number = args[0]
        if number in self._cache:
            return self._cache[number]
        result = compute_fibonacci(number)
        self._cache[number] = result
        return result

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

def main():
    # Non cached
    print(compute_fibonacci(10))

    # Cached
    with CachedComputeFibonacci() as cached_compute_fibonacci:
        print(cached_compute_fibonacci(35))
        print(cached_compute_fibonacci(35))



if __name__ == '__main__':
    main()

Logging

Logging can be useful for debugging and auditing purposes.

def compute_fibonacci(number):
    if number <= 1:
        return number
    return compute_fibonacci(number-1) + compute_fibonacci(number-2)


class LoggedComputeFibonacci:
    def __init__(self):
        pass

    def __call__(self, *args, **kwargs):
        print(f"calling compute_fibonacci with args={args} kwargs={kwargs}")
        result = compute_fibonacci(args[0])
        print(f"compute_fibonacci={result}")
        return result

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

def main():
    # Logging
    with LoggedComputeFibonacci() as cached_compute_fibonacci:
        print(cached_compute_fibonacci(35))
        print(cached_compute_fibonacci(36))



if __name__ == '__main__':
    main()

Error detection and correction

If you find yourself duplicating the same try/catch logic in multiple places of your code perhaps you can extract it into a context manager for handling errors:

from contextlib import contextmanager

@contextmanager
def my_error_handler():
    try:
        yield
    except ZeroDivisionError:
        print("abort abort")

def main():
    # error handling
    with my_error_handler():
        print("0 / 0 =", 0 / 0)



if __name__ == '__main__':
    main()

The code is definitely more cleaner this way, in my opinion.

Thanks for reading and I hope that you’ve learnt something!