DEV Community: Mahmoud Shawara

TDD joyful coding

Mahmoud Shawara — Wed, 21 Jul 2021 15:02:31 +0000

Why don't you just write unit tests and stop yelling?

Because you find it boring to test your code since you think it is already working fine. Also, you are most likely to write tests that will make your code pass, because your code wasn't designed to be testable.

Isn't it stupid to write tests for code that doesn't exist?

I thought so, but since I watched a demo about how it works, I started to understand how it can help you have more fun while coding

TDD steps

Write test to make it fail
write code to make it pass
Refactor the code
Repeat the steps 1,2,3 until you can't make it fail

This article will be a demo for Queue implementation:

First iteration

Step 1

#tests.py
class TestQueue(unittest.TestCase):
    def test_queue(self):
        q = MyQueue()

Step 2

#queue.py
class MyQueue:
    pass

Step 3

#tests.py
class TestQueue(unittest.TestCase):
    def test_queue(self):
        q = Queue()

#queue.py
class Queue:
    pass

Second iteration

Step 1

#tests.py
class TestQueue(unittest.TestCase):
    def test_new_queue_is_empty(self):
        q = Queue()
        self.assertTrue(q.is_empty())

Step 2

#queue.py
class Queue:
    def is_empty(self) -> bool:
        return True

Third iteration

Step 1

#tests.py
class TestQueue(unittest.TestCase):
    ...
    def test_queue_after_one_enqueue_not_empty(self):
        q = Queue()
        q.enqueue(3)
        self.assertFalse(q.is_empty())

Step 2

#queue.py
class Queue:
    __empty = True

    def is_empty(self) -> bool:
        return self.__empty

    def enqueue(self, element: int):
        self.__empty = False

Step 3

#tests.py
class TestQueue(unittest.TestCase):
    def setUp(self) -> None:
        self.q = Queue()

    def test_new_queue_is_empty(self):
        self.assertTrue(self.q.is_empty())

    def test_queue_after_one_enqueue_not_empty(self):
        self.q.enqueue(3)
        self.assertFalse(self.q.is_empty())

Fourth iteration

Step 1

#tests.py
class TestQueue(unittest.TestCase):
    ...
    def test_dequeue_from_empty_queue(self):
        self.assertRaises(UnderFlowException, self.q.dequeue)

Step 2

#queue.py
class UnderFlowException(Exception):
    def __str__(self):
        return "Can't dequeue from empty queue"

class Queue:
    ...
    def dequeue(self):
        if self.is_empty():
            raise UnderFlowException()
        return -1

Fifth iteration

Step 1

#tests.py
class TestQueue(unittest.TestCase):
    ...
    def test_is_empty_after_one_enqueue_one_dequeue(self):
        self.q.enqueue(10)
        self.q.dequeue()
        self.assertTrue(self.q.is_empty())

Step 2

#queue.py
    ...
    def dequeue(self):
        if self.is_empty():
            raise UnderFlowException()
        self.__empty = True
        return -1

Sixth iteration

Step 1

#tests.py
class TestQueue(unittest.TestCase):
    ...
    def test_is_not_empty_two_enqueue_one_dequeue(self):
        self.q.enqueue(10)
        self.q.enqueue(20)
        self.q.dequeue()
        self.assertFalse(self.q.is_empty())

Step 2

#queue.py
class Queue:
    __size = 0

    def is_empty(self) -> bool:
        return self.__size == 0

    def enqueue(self, element: int):
        self.__size += 1

    def dequeue(self):
        if self.is_empty():
            raise UnderFlowException()
        self.__size -= 1
        return -1

Seventh iteration

Step 1

#tests.py
class TestQueue(unittest.TestCase):
    ...
    def test_dequeue_value(self):
        self.q.enqueue(10)
        self.assertEqual(self.q.dequeue(), 10)
        self.q.enqueue(20)
        self.assertEqual(self.q.dequeue(), 20)

Step 2

#queue.py
class Queue:
    __size = 0
    __element = -1

    def is_empty(self) -> bool:
        return self.__size == 0

    def enqueue(self, element: int):
        self.__size += 1
        self.__element = element

    def dequeue(self) -> int:
        if self.is_empty():
            raise UnderFlowException()
        self.__size -= 1
        return self.__element

Eighth iteration

Step 1

#tests.py
class TestQueue(unittest.TestCase):
    ...
    def test_dequeue_value_after_two_enqueues(self):
        self.q.enqueue(10)
        self.q.enqueue(20)
        self.assertEqual(self.q.dequeue(), 10)

Step 2

#queue.py
class Queue:
    __size = 0
    __elements = []

    def is_empty(self) -> bool:
        return self.__size == 0

    def enqueue(self, element: int):
        self.__size += 1
        self.__elements.append(element)

    def dequeue(self) -> int:
        if self.is_empty():
            raise UnderFlowException()
        self.__size -= 1
        return self.__elements.pop(0)

Step 3

#queue.py
class Queue:
    __elements = []

    def is_empty(self) -> bool:
        return len(self.__elements) == 0

    def enqueue(self, element: int):
        self.__elements.append(element)

    def dequeue(self) -> int:
        if self.is_empty():
            raise UnderFlowException()
        return self.__elements.pop(0)

Django ORM Optimization Tips #4 recursive query

Mahmoud Shawara — Fri, 02 Oct 2020 23:13:55 +0000

Recursive quires is frequently needed when you have a self referencing table

Our example today will be items and categories user wants to get item details with item category path to root category.

Root-Category > Category > Sub-Category > leaf-Category



#models.py
class Category(models.Model):
    name = models.CharField(max_length=127)
    parent = models.ForeignKey('self', null=True)

We need to implement query that get the category path to root:

The naive solution:



def path_to_root_category(category_id):
    if category_id is None:
        return []
    category = Category.objects.get(pk=category_id)
    result = path_to_root_category(category.parent_id)
    result.append(category)
    return result

path_to_root_category(4)
[<Category: root>, <Category: category>, <Category: sub-category>, <Category: leaf-category>]

The cost of this function is 4 queries (number of nodes in the path)
If we have a path of large number of nodes this will be so bad.

The optimized solution:



def path_to_root_category(category_id):
    query = '''
    WITH RECURSIVE parents AS (
        SELECT category.*, 0 AS relative_depth
        FROM category
        WHERE id = %s

        UNION ALL

        SELECT category.*, parents.relative_depth - 1
        FROM category,parents
        WHERE category.id = parents.parent_id
    )
    SELECT id, name, parent_id, relative_depth
    FROM parents
    ORDER BY relative_depth;
    '''
    return Category.objects.raw(query, [category_id])

Result:

Query explanation

First find the target category element and make it the first element of the parents set
Follow parent of all categories in parents set and add them to the same set
Finally select all categories in the parents set ordered by their depth

Django ORM Optimization Tips #3 aggregation

Mahmoud Shawara — Wed, 23 Sep 2020 19:20:44 +0000

Today we are going to discuss GROUP BY with aggregation(SUM)

Suppose we have Tables of users , playlists and songs.

Each user can have one or more playlists.
Each playlist can have one or more songs

# models
class User(models.Model):
    name = models.CharField(max_length=127)

    @property
    def playlists_total_length(self) -> int:
        return sum([p.songs_total_length for p in self.playlists.all()])


class Playlist(models.Model):
    user = models.ForeignKey(User, related_name='playlists')
    name = models.CharField(max_length=255)

    @property
    def songs_total_length(self) -> int:
        return sum([song.length for song in self.songs.all()])


class Song(models.Model):
    playlist = models.ForeignKey(Playlist, related_name='songs')
    title = models.CharField(max_length=255)
    length = models.PositiveIntegerField(help_text=_('In seconds'))

So if our goal is to list all users with playlists_total_length

How bad is this code?

If we had for example:

10 Users
10 Playlists per user
10 songs per playlist

Total objects to fetch = 10 (Users) + 10*10 (Playlists) + 10*10*10 (Songs)

1110 DB rows in 3 different tables for just 10 users

Solution: GROUP BY with aggregation(SUM)

Use aggregation to calculate total length of all songs: Song.objects.aggregate(Sum('length'))

SELECT SUM("song"."length") AS "length__sum" FROM "song";

Result: {'length__sum': 490}

Use annotate for Group by: Song.objects.annotate(Sum('length'))

SELECT * , SUM("song"."length") AS "length__sum" 
FROM "song" 
GROUP BY "song"."id", 
         "song"."playlist_id", 
         "song"."title",
         "song"."length

Result: <QuerySet [Song(...),Song(...),...]>

Use values before annotate to group by specific columns: Song.objects.values('playlist_id').annotate(Sum('length'))

SELECT "song"."playlist_id", SUM("song"."length") AS "length__sum"
FROM "song" 
GROUP BY "song"."playlist_id"

Result: <QuerySet [{'playlist_id': 2, 'length__sum': 310}, ...]>

Use values_list to select sum values only not dict

Song.objects.values('playlist_id').annotate(Sum('length')).values_list('length__sum')

We can compress this query and still getting the same generated sql query:

Song.objects.values('playlist_id').values_list(Sum('length'))

SELECT SUM("song"."length") AS "length__sum" 
FROM "song" 
GROUP BY "song"."playlist_id"

Result: <QuerySet [(310,), (180,)]>

List all Playlists with total songs length:

songs_subquery = Subquery(
    queryset=Song.objects
        .values('playlist_id') \
        .filter(playlist__id=OuterRef('id')) \
        .values_list(Sum('length')),
    output_field=IntegerField())

queryset = Playlist.objects.annotate(
    _songs_total_length=songs_subquery)

OuterRef will make it possible to access fields from parent query when you are in subquery. OuterRef('id') -> "playlist"."id"

SELECT "playlist"."id",
       "playlist"."user_id",
       "playlist"."name",
       (
           SELECT SUM("song"."length")
               FROM "song"
               WHERE "song"."playlist_id" = ("playlist"."id")
               GROUP BY "song"."playlist_id" LIMIT 1
        ) AS "songs_total_length"
    FROM "playlist";

List all Users with total songs length:

songs_subquery = Subquery(
    queryset=Song.objects
        .values('playlist__id') \
        .filter(playlist__id=OuterRef('id')) \
        .values_list(Sum('length')),
    output_field=IntegerField(default=0))

playlists_subquery = Subquery(
    queryset=Playlist.objects
        .values('user__id') \
        .filter(user__id=OuterRef('id')) \
        .annotate(_songs_total_length=Sum(songs_subquery)) \
        .values_list('_songs_total_length'),
    output_field=IntegerField(default=0))

queryset = User.objects.annotate(_playlists_total_length=playlists_subquery)

SELECT "user"."id",
       "user"."name",
       (SELECT SUM((SELECT SUM(U0."length") AS "length__sum"
                    FROM "song" U0
                    WHERE U0."playlist_id" = V0."id"
                    GROUP BY U0."playlist_id")) AS "_songs_total_length"
        FROM "playlist" V0
        WHERE V0."user_id" = "user"."id"
        GROUP BY V0."user_id") AS "_playlists_total_length"
FROM "user";

It will be so hard to write this complex query in everywhere you want

So here is the final clean code:

class UserQuerySet(models.QuerySet):
    @classmethod
    def playlists_total_length(cls):
        playlist_annotation = PlaylistQuerySet.songs_total_length()

        queryset = Playlist.objects \
            .values('user__id') \
            .filter(user__id=OuterRef('id')) \
            .values_list(Sum(playlist_annotation))

        return Subquery(
            queryset=queryset,
            output_field=IntegerField()
        )

    def collect(self):
        return self.annotate(_playlists_total_length=self.playlists_total_length())


class User(models.Model):
    objects = UserQuerySet.as_manager()
    name = models.CharField(max_length=255)

    @property
    def playlists_total_length(self):
        if hasattr(self, '_playlists_total_length'):
            return self._playlists_total_length
        return sum([p.songs_total_length for p in self.playlists.all()])



class PlaylistQuerySet(models.QuerySet):
    @classmethod
    def songs_total_length(cls):
        queryset = Song.objects \
            .values('playlist__id') \
            .filter(playlist__id=OuterRef('id')) \
            .values_list(Sum('length'))  # The SUM aggregation

        return Subquery(
            queryset=queryset,
            output_field=models.IntegerField()
        )

    def collect(self):
        return self.annotate(_songs_total_length=self.songs_total_length())


class Playlist(models.Model):
    objects = PlaylistQuerySet.as_manager()
    user = models.ForeignKey(User, related_name='playlists')
    name = models.CharField(max_length=255)

    @property
    def songs_total_length(self):
        if hasattr(self, '_songs_total_length'):
            return self._songs_total_length
        return sum([song.length for song in self.songs.all()])



class Song(models.Model):
    playlist = models.ForeignKey(Playlist, related_name='songs')
    title = models.CharField(max_length=255)
    length = models.PositiveIntegerField(help_text='In seconds')

Thanks, Ivo Donchev

Django ORM Optimization Tips #2 prefetch_related

Mahmoud Shawara — Mon, 21 Sep 2020 20:16:48 +0000

This has a similar purpose to select_related, in that both are designed to stop the deluge of database queries that is caused by accessing related objects, but the strategy is quite different.

select_related works by creating an SQL join and including the fields of the related object in the SELECT statement. For this reason, select_related gets the related objects in the same database query. However, to avoid the much larger result set that would result from joining across a ‘many’ relationship, select_related is limited to single-valued relationships - foreign key and one-to-one.

prefetch_related, on the other hand, does a separate lookup for each relationship, and does the ‘joining’ in Python. This allows it to prefetch many-to-many and many-to-one objects, which cannot be done using select_related

lets go to our favourite part examples:

In this we have users and languages tables, each user knows more than one language and we want to list users with their languages

# models.py
class Language(models.Model):
    name = models.CharField(max_length=63)
    code = models.CharField(max_length=7)


class User(models.Model):
    fullname = models.CharField(max_length=127)
    languages = models.ManyToManyField(Language, related_name="users")

# serializers.py
class LanguageSerializer(serializers.ModelSerializer):
    class Meta:
        model = Language
        fields = ('id', 'name', 'code')


class UserSerializer(serializers.ModelSerializer):
    languages = LanguageSerializer(many=True)

    class Meta:
        model = User
        fields = ('id', 'fullname', 'languages')

# views.py
class UserListView(ListAPIView):
    serializer_class = UserSerializer
    queryset = User.objects.prefetch_related('languages')

According to prefetch_related documentation:

To avoid the much larger result set that would result from joining across a ‘many’ relationship, prefetch_related does a separate lookup for each relationship, and does the ‘joining’ in Python

So queryset User.objects.prefetch_related('languages') will be divided into 2 queries:

SELECT "user"."id", "user"."fullname" FROM "user"

SELECT "user_language"."user_id","language"."id", "language"."name",
"language"."code" 
FROM "language" INNER JOIN "user_language" 
ON ("language"."id" = "user_language"."language_id") 
WHERE "user_language"."user_id" IN (...);

And last thing is matching these queries with Python code

Django ORM Optimization Tips #1 select_related

Mahmoud Shawara — Sun, 20 Sep 2020 00:54:20 +0000

This is gonna be django performance series

The Best Way to Learn is to Teach

I'm a man of few words

So let's start with example of Users and Cities models.

# models.py

class City(models.Model):
    name = models.CharField(max_length=63)


class User(models.Model):
    fullname = models.CharField(max_length=127)
    city = models.ForeignKey(City, on_delete=models.SET_NULL, null=True)

And this is our serializers

# serializers.py
class CitySerializer(serializers.ModelSerializer):
    class Meta:
        model = City
        fields = ('id', 'name')


class UserSerializer(serializers.ModelSerializer):
    city = CitySerializer()

    class Meta:
        model = User
        fields = ('id', 'fullname', 'city')

And this is our views

# views.py
class UserListView(ListAPIView):
    serializer_class = UserSerializer
    queryset = User.objects.all()

So if you looked to the queryset User.objects.all() its sql is

SELECT "users"."id","users"."fullname","users"."city_id" 
FROM "users"

It is not efficient because for each user you are going to access database to select user city for the nested serializer CitySerializer , It is (n+1) queries

Solution: use select_related

after changes queryset will be User.objects.select_related('city')
sql:

SELECT "user"."id","user"."fullname", "user"."city_id","city"."id",
"city"."name"  FROM "user"
LEFT OUTER JOIN "city" ON ("user"."city_id" = "city"."id")

So now it is just one query