DEV Community: Shodhan Save

Python Core Utility - `contextlib.contextmanager`

Shodhan Save — Thu, 04 Dec 2025 11:01:04 +0000

Intro

In this post, we will look at the contextmanager function from the contextlib module.

The contextlib.contextmanager function is used as a decorator that allows you to create context managers using a generator function instead of writing a class with __enter__ and __exit__ methods. It's a cleaner, more pythonic way to handle resource management, setup/teardown operations, and temporary state changes.

Context managers in a nutshell
You might have come across code like:

with open(file_name, "w") as f:
    do_something(f)

This code block opens the file_name file in write-mode, stores the opened file handle in the variable f. This f can then be used in the scope of the with block.

The thing that is passed to with (in this case, the result of open(..)) is a context manager.

Context managers have methods that run before and after the with block execution. These methods are, respectively, __enter__() and __exit__(). That means, any traditional class that defines these methods can be used as a context manager (see example below). But, with the @contextmanager decorator, you can achieve the same result with a simple generator function. And that's what we are covering in this post here.

However, for the sake of completeness, here's how you can create your own context managers using a class. This example creates a requests.Session to load the auth token into headers. Then, upon entering the with block it first calls the /login endpoint, and upon exiting the block it calls the /logout endpoint. Inside the block, you can do your thing, call different endpoints, without worrying about login or logout.

import requests

class SessionManager:
    def __init__(self, base_url, token):
        self.base_url = base_url.rstrip("/")
        self.token = token

    def __enter__(self):
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {self.token}"})
        response = self.session.post(f"{self.base_url}/login")
        response.raise_for_status()
        return self.session

    def __exit__(self, exc_type, exc_value, traceback):
        self.session.post(f"{self.base_url}/logout")

with SessionManager("https://my-base-url.com", "my-fake-token") as session:
    refresh_reports(session)
    download_reports(session)
    email_reports(session)

A bit more about the __exit__ method - it accepts three arguments (all optional):

the exception class
the exception instance
the traceback object

If an exception occurs in the with block, these parameters contain the exception details. If you have set __exit__ to return True, the exception raised inside the with block is suppressed, and if it returns False (or None), the exception propagates to outside of the block.

From the standard docs:

If an exception is supplied, and the method wishes to suppress the exception (i.e., prevent it from being propagated), it should return a true value. Otherwise, the exception will be processed normally upon exit from this method.

Note that exit() methods should not reraise the passed-in exception; this is the caller’s responsibility.

Use cases

The primary use cases for using contextmanager could be:

Resource management - Opening and closing files, database connections, network sockets, wrapping API calls between generic login and logout, etc.
Setup and teardown - Setting up test fixtures, temporary configurations, mocking
Temporary state changes - Changing working directory, modifying environment variables, suppressing output
Locking and synchronization - Acquiring and releasing locks
Transaction management - Begin/commit/rollback database transactions

Usage

You probably already know the traditional, class-based approach to creating context managers (using the __enter__ and __exit__ methods). If not, I have covered it briefly in the Context managers in a nutshell part in Intro section above.

Now let's see how the @contextmanager decorator can be used:

Consider the same example as given before - a wrapper that calls the /login and /logout endpoints via requests.Session before and after the with block:

from contextlib import contextmanager
import requests

@contextmanager
def session_manager(base_url, token):
    try:
        session = requests.Session()
        session.headers.update({"Authorization": f"Bearer {token}"})
        response = session.post(f"{base_url}/login")
        response.raise_for_status()
        yield session
    finally:
        session.post(f"{base_url}/logout")

with session_manager("https://my-base-url.com", "my-fake-token") as session:
    refresh_reports(session)
    download_reports(session)
    email_reports(session)

Here's how it works:

Code before yield runs when entering the with block (like __enter__)
The yield statement returns (more correctly, yields) what gets assigned to the with variable (session in this case)
Code after yield runs when exiting the with block (like __exit__)
In the above example, we could also get rid of the try..finally completely and have the /logout call immediately after the yield statement. But putting the logout (or any closing functionality as such) in finally ensures that it gets called no matter what.

One might get inclined in the favour of the class-based implementation over this decorator based one. And that's perfectly fine! I myself have used the class-based method several times - it just depends on how complex your use case is, and how much more work you wish to do around __enter__ and __exit__.

Some other common use cases of the decorator method can include:

Timing code execution:

from contextlib import contextmanager
import time

@contextmanager
def timer(label):
    start = time.time()
    try:
        yield
    finally:
        end = time.time()
        print(f"{label}: {end - start:.4f} seconds")

with timer("Processing"):
    do_something()

# [Out]: Processing: 0.5001 seconds

Temporary directory change:

import os
from contextlib import contextmanager

@contextmanager
def change_dir(new_dir):
    old_dir = os.getcwd()
    os.chdir(new_dir)
    print(f"Changing dir from {old_dir} to {new_dir}")
    try:
        yield
    finally:
        os.chdir(old_dir)
        print(f"Changing dir back to {old_dir} from {new_dir}")

with change_dir("/tmp"):
    print(f"Inside {os.getcwd()}")

# [Out]:
# Changing dir from /home/shodhan/projects/blog to /tmp
# Inside /tmp
# Changing dir back to /home/shodhan/projects/blog from /tmp

Suppressing stdout:

import sys
from contextlib import contextmanager
from io import StringIO

@contextmanager
def suppress_stdout():
    old_stdout = sys.stdout
    sys.stdout = StringIO()
    try:
        yield
    finally:
        sys.stdout = old_stdout

print("This prints")
with suppress_stdout():
    print("This doesn't print")
print("This prints again")
# [Out]:
# This prints
# This prints again

Real-life scenarios

Here are some practical examples where @contextmanager shines.

Database transaction management

One of the most common uses - ensuring transactions are committed or rolled back:

from contextlib import contextmanager
import sqlite3

@contextmanager
def db_transaction(db_path, commit=False):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise
    finally:
        conn.close()

with db_transaction("mydb.db") as cursor:
    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))

Note the raise statement here - Above we said that we do not raise from the context manager, but this is different - here we are raising the same exception that we caught explicitly, because we wanted to do another thing (conn.rollback) before it bubbled up.

Temporary environment variables

When testing code that depends on environment variables:

import os
from contextlib import contextmanager

@contextmanager
def temp_env_var(key, new_value):
    old_value = os.environ.get(key)
    os.environ[key] = new_value
    try:
        yield
    finally:
        if old_value is None:
            os.environ.pop(key, None)
        else:
            os.environ[key] = old_value

print(os.environ.get("DEBUG"))
# [Out]: None

with temp_env_var("DEBUG", "true"):
    print(os.environ.get("DEBUG"))
    # [Out]: true

print(os.environ.get("DEBUG"))
# [Out]: None

A special context manager (`nullcontext`) to parametrize pytest fixtures

This was a sweet surprise when I found it! The nullcontext is another function of contextlib, and is a special case of contextmanager, in the sense that it is a context manager that does nothing! I will write in-depth about it in a separate post. For now, let's see how it can help in the pytest parametrized fixtures.

When writing pytest parametrized fixtures, where some cases should raise exceptions and others shouldn't, you can use nullcontext for cases that should not raise exceptions. It can be used here to avoid writing an if..else block for cases that raise exceptions and cases that don't. This keeps the code quite elegant!

import pytest
from contextlib import nullcontext

def divide(a, b):
    """Division function that raises on zero."""
    return a / b

@pytest.mark.parametrize(
    "a, b, expected, raised_exception",
    [
        (10, 2, 5, nullcontext()),
        (20, 4, 5, nullcontext()),
        (10, 0, None, pytest.raises(ZeroDivisionError)),
        (0, 0, None, pytest.raises(ZeroDivisionError)),
    ],
)
def test_divide(a, b, expected, raised_exception):
    with raised_exception:
        result = divide(a, b)
        if expected is not None:
            assert result == expected

Managing multiple resources

When you need to manage multiple related resources, you can call the same context manager multiple times. Consider the example of temporary environment variables from above, we can call it multiple times like this:

with (
    temp_env_var("DEBUG", "true"),
    temp_env_var("RUN_ENV", "prod"),
):
    do_something()

Async context managers

All the examples mentioned here are synchronous, including the ones opening database connection, or making the session requests. But contextlib also provides another function - asynccontextmanager for async operations. Although this is technically a different function than what this post is about, I'd like to include it here as I feel it is closely related and will also make sense to you.

This works similarly to the regular, "sync" context manager, except that instead of calling it with with, you must call it with async with block. And of course, the decorator function should also be async:

Code copied from the standard docs

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_connection():
    conn = await acquire_db_connection()
    try:
        yield conn
    finally:
        await release_db_connection(conn)

async def get_all_users():
    async with get_connection() as aconn:
        return aconn.query("SELECT ...")

Nuances

Must use try/finally if you need cleanup

If you have cleanup code, it MUST be in a finally block to ensure it runs even if an exception occurs:

from contextlib import contextmanager

# WRONG - cleanup might not happen
@contextmanager
def bad_context():
    resource = acquire_resource()
    yield resource
    release_resource(resource)  # might not run if exception occurs!

# CORRECT - cleanup always happens
@contextmanager
def good_context():
    resource = acquire_resource()
    try:
        yield resource
    finally:
        release_resource(resource)  # always runs

Only one yield allowed

A context manager can only have one yield statement. If you have multiple yields, only the first one is used:

from contextlib import contextmanager

@contextmanager
def bad_multiple_yields():
    yield 1
    yield 2  # this never happens

with bad_multiple_yields() as value:
    print(value)
# [Out]: 1

Exceptions before `yield`

If an exception occurs before the yield, it propagates immediately and __exit__ is never called:

from contextlib import contextmanager

@contextmanager
def failing_context():
    print("Setting up")
    raise ValueError("Setup failed")
    yield  # never reached
    print("Cleaning up")  # never reached

try:
    with failing_context() as ctx:
        print("In context")  # never reached
except ValueError as e:
    print(f"Caught: {e}")

# [Out]:
# Setting up
# ValueError: Setup failed

This is expected behavior - if setup fails, there's nothing to clean up.

This is the reason why I put response.raise_for_status() in the session manager example above, so that if there is any error during login, it must not proceed ahead!

Cannot suppress exceptions by default

Unlike class-based context managers where you can return True from __exit__ to suppress exceptions, with @contextmanager you need to catch and handle exceptions explicitly:

from contextlib import contextmanager

# This does NOT suppress exceptions
@contextmanager
def does_not_suppress():
    try:
        yield
    finally:
        pass  # exception still propagates

# To suppress, you must catch explicitly
@contextmanager
def suppress_errors():
    try:
        yield
    except Exception as e:
        print(f"Suppressed: {e}")
        # by not re-raising, we suppress the exception

with suppress_errors():
    raise ValueError("This is suppressed")
    print("This won't print")

print("Program continues")
# [Out]:
# Suppressed: This is suppressed
# Program continues

Nested context managers can be tricky

Be careful when nesting context managers - exceptions in inner context managers can affect outer ones:

from contextlib import contextmanager

@contextmanager
def outer():
    print("Outer enter")
    try:
        yield "outer"
    finally:
        print("Outer exit")

@contextmanager
def inner():
    print("Inner enter")
    try:
        yield "inner"
    finally:
        print("Inner exit")

# nested usage
with outer():
    with inner():
        print("Inside inner")
        raise ValueError("Error!")

# [Out]:
# Outer enter
# Inner enter
# Inside inner
# Inner exit
# Outer exit
# ValueError: Error!

Also, looking at the print statements, you can see that the cleanup order should be inside-out (like nested try/finally blocks).

Reusability

Context managers created with @contextmanager are reusable - you can use them multiple times:

from contextlib import contextmanager

@contextmanager
def my_context():
    print("Enter")
    yield
    print("Exit")

ctx = my_context()

with ctx:
    print("First use")
# [Out]:
# Enter
# First use
# Exit

with ctx:
    print("Second use")
# [Out]:
# Enter
# Second use
# Exit

However, be aware that the generator function runs fresh each time, so any state is reset.

Other types of context managers

If you check the standard docs, you will see that there are several different types of context managers, made for special cases that are believed to be quite frequent.

I would really like to write about them, but perhaps in a separate blog post. I will update this space once I have that post published.

🙏 Thanks for taking out the time to read this! And as always, please do not hesitate to give feedback on my posts!

Python Core Utility - `itertools.groupby`

Shodhan Save — Sat, 15 Nov 2025 19:22:26 +0000

Intro

In this post, we will look at the groupby function from the itertools module.

The itertools.groupby function groups consecutive elements from an iterable. It accepts an iterable and an optional callable key that computes a key value for each element. It returns an iterator that yields tuples of (key, group), where group itself is an iterator over the consecutive elements that share that key value. If key is not specified or passed as None, then the value of each element is used as key value.

⚠️ A few critical points to remember: groupby only groups consecutive elements with the same key. If your data has non-consecutive elements with the same key, you'll need to sort the data first. Also, the grouped iterators are only accessible once. So, if you want to access the groups later, you should store them as list. See the Caveats section for more details.

Use cases

The primary use case of groupby is when you need to group consecutive items in an iterable by some property or computed value. This is extremely useful when processing sorted data, log files, or any sequential data where similar items appear together.

Common scenarios include:

Processing sorted data where you need to aggregate items by category
Analyzing log files where consecutive entries often share attributes (timestamps, user IDs, etc.)
Run-length encoding for data compression
Finding consecutive sequences in data
Breaking down data into chunks based on some property

The beauty of groupby is its memory efficiency - it processes data lazily without loading everything into memory, making it perfect for large datasets.

Usage

Here's a basic example showing how groupby works.

Note that we have already sorted the list in a way that we want the elements to be grouped:

from itertools import groupby

# group names by their first alphabet
names = ["Alice", "Bob", "Charlie", "Chuck", "Dan"]

for key, group in groupby(names, key=lambda x: x[0]):
    print(f"{key}: {list(group)}")

# [Out]:
# A: ['Alice']
# B: ['Bob']
# C: ['Charlie', 'Chuck']
# D: ['Dan']

Let's see what happens when the elements are not sorted in the way that we want them to be grouped:

from itertools import groupby

# group by even or odd
numbers = [2, 4, 6, 11, 3, 5, 8, 10, 9]

for key, group in groupby(numbers, key=lambda x: x % 2 == 0):
    print(f"Even: {key}, Values: {list(group)}")

# [Out]:
# Even: True, Values: [2, 4, 6]
# Even: False, Values: [11, 3, 5]
# Even: True, Values: [8, 10]
# Even: False, Values: [9]

Notice how the even numbers [8, 10] formed a separate group from [2, 4, 6]. This happened because they were not consecutive in the original list, so when the element 11 was encountered, the value of key changed (earlier it was evaluated to 0 for even number, then it was evaluated to 1 for the number 11) and hence a new group was formed. Same thing happened for the element 9.

I mentioned above that elements need to be sorted in the same way that we want them grouped. Here's what I mean by that:

Say I have a list of dicts to be grouped by a key named city in each dict. Then, I also need to ensure that the list itself is sorted based on the same key. Generally, in case of complex data structures, it is a good idea to abstract the sorting function / lambda into a variable and then access that in both places - to sort the list and to pass key to groupby:

from itertools import groupby

details = [
    {"name": "Alice", "city": "Berlin"},
    {"name": "Bob", "city": "Amsterdam"},
    {"name": "Charlie", "city": "Mumbai"},
    {"name": "Chuck", "city": "Berlin"},
    {"name": "Dan", "city": "Warsaw"},
    {"name": "Eve", "city": "New York"},
    {"name": "Frank", "city": "Mumbai"},
    {"name": "Grace", "city": "Berlin"},
]

sort_key = lambda x: x["city"]
# alternatively:
# sort_key = operator.itemgetter("city")
details.sort(key=sort_key)
for key, group in groupby(details, sort_key):
    names = [each["name"] for each in group]
    print(f"{key}: {names}")

# [Out]:
# Amsterdam: ['Bob']
# Berlin: ['Alice', 'Chuck', 'Grace']
# Mumbai: ['Charlie', 'Frank']
# New York: ['Eve']
# Warsaw: ['Dan']

By now you must have figured that the key parameter can be any callable, including functions, lambdas, even None. I'd encourage you to try out different options!

Real-life scenarios

Here are some practical examples where groupby shines.

Processing sorted log entries

When analyzing logs that are stored chronologically, groupby can be used to group entries by timestamp, user, status code, or any other attribute. For e.g., grouping them by each hour:

from itertools import groupby

def get_hour_from_timestamp(entry):
    return entry["timestamp"].split(":")[0]

# log entries sorted by timestamp
logs = [
    {"timestamp": "10:00:36", "status": 200, "url": "/home"},
    {"timestamp": "10:10:41", "status": 200, "url": "/about"},
    {"timestamp": "10:20:14", "status": 404, "url": "/missing"},
    {"timestamp": "11:10:26", "status": 404, "url": "/gone"},
    {"timestamp": "11:40:45", "status": 500, "url": "/api"},
    {"timestamp": "12:30:16", "status": 200, "url": "/home"},
]

for hour, entries in groupby(logs, key=get_hour_from_timestamp):
    print(hour, entries)    # do further processing with `entries`

# [Out]:
# 10 <itertools._grouper object at 0x..>
# 11 <itertools._grouper object at 0x..>
# 12 <itertools._grouper object at 0x..>

Run-length encoding

One of the classic uses of groupby - compressing consecutive repeated values:

from itertools import groupby

def run_length_encode(data):
    """Compress consecutive repeated values."""
    return [(key, len(list(group))) for key, group in groupby(data)]

def run_length_decode(encoded):
    """Decompress run-length encoded data."""
    return [key for key, count in encoded for _ in range(count)]

# compress
original = "AAABBCCCCAAD"
compressed = run_length_encode(original)
print(compressed)
# [Out]: [('A', 3), ('B', 2), ('C', 4), ('A', 2), ('D', 1)]

# decompress
decompressed = "".join(run_length_decode(compressed))
assert decompressed == original 
print(decompressed)
# [Out]: AAABBCCCCAAD

Grouping database query results

When you fetch sorted data from a database and need to aggregate it.

ℹ️ Although it is usually better to delegate this functionality to the database engine itself (using GROUPBY operator in SQL), there might be instances when you need un-grouped data in multiple places. Then, groupby can be used.

from collections import defaultdict
from itertools import groupby
from operator import itemgetter

# simulating sorted database results
orders = [
    {"customer_id": 1, "product": "apple", "quantity": 5},
    {"customer_id": 1, "product": "banana", "quantity": 3},
    {"customer_id": 2, "product": "orange", "quantity": 2},
    {"customer_id": 2, "product": "apple", "quantity": 1},
    {"customer_id": 3, "product": "banana", "quantity": 4},
]

# aggregate by customer (data is already sorted by customer_id)
summary = {}
for customer_id, orders_group in groupby(orders, key=itemgetter("customer_id")):
    summary[customer_id] = {"total_items": 0, "products": []}
    for order in orders_group:
        summary[customer_id]["total_items"] += order["quantity"]
        summary[customer_id]["products"].append(order["product"])

print(summary)
# [Out]:
# {
#     1: {'total_items': 8, 'products': ['apple', 'banana']},
#     2: {'total_items': 3, 'products': ['orange', 'apple']},
#     3: {'total_items': 4, 'products': ['banana']}
# }

Caveats

Only groups consecutive elements

This is the most important caveat. groupby does NOT group all elements with the same key together - only consecutive ones. This differs from SQL's GROUPBY that groups elements irrespective of their order.

from itertools import groupby

data = [1, 2, 1, 2, 1]

# without sorting
for key, group in groupby(data):
    print(f"{key}: {list(group)}")
# [Out]:
# 1: [1]
# 2: [2]
# 1: [1]
# 2: [2]
# 1: [1]

# with sorting - now it groups properly
for key, group in groupby(sorted(data)):
    print(f"{key}: {list(group)}")
# [Out]:
# 1: [1, 1, 1]
# 2: [2, 2]

If you need to group all elements with the same key regardless of position, use collections.defaultdict or collections.Counter instead.

Not consuming the group before moving on

The group iterator is tied to the main groupby iterator. Once you move to the next group, the previous group's iterator is exhausted:

from itertools import groupby

data = [1, 1, 2, 2, 3]
result = []

for key, group in groupby(data):
    result.append((key, group))

# try accessing the groups again
for key, group in result:
    print(f"{key}: {list(group)}")

# [Out]:
# 1: []
# 2: []
# 3: []

Solution: Convert groups to lists immediately. In above snippet, this change must be done:

result.append((key, list(group)))

Advancing the groupby iterator invalidates previous groups

Even if you save a group iterator, advancing the main groupby iterator will invalidate it:

from itertools import groupby

data = [1, 1, 2, 2]
it = groupby(data)

key1, group1 = next(it)
key2, group2 = next(it)  # this invalidates group1

print(f"{key1}: {list(group1)}")  # empty!
# [Out]: 1: []

print(f"{key2}: {list(group2)}")
# [Out]: 2: [2, 2]

Memory considerations with large groups

While groupby itself is memory-efficient, if you convert large groups to lists, you'll use memory. Instead, process each group there itself, for the purpose you have for them, without converting into list (i.e., using like generator objects itself).

The beauty of groupby is that it's lazy - use it that way when dealing with large data!

Python Core Utility - `collections.defaultdict`

Shodhan Save — Sat, 15 Nov 2025 19:18:50 +0000

Intro

In this post, we will look at the defaultdict datatype from the collections module.

The collections.defaultdict is a subclass of the built-in dict that accepts a callable (default_factory) during its initialization. Then, when you try to access a key in the initialized object, it does a regular dict lookup to fetch the key's value. If the key is missing, it calls the default_factory that generates a value for the key requested. This key-value pair is then stored in the dict and the value is returned.

Use cases

The primary use case is when you need to build up a dict based on an iterable and don't want to check if keys exist before operating on them. This is extremely common when grouping items, counting occurrences, or building nested data structures.

For example, when grouping items by some attribute, you'd normally need to check if the key exists before appending to a list. With defaultdict(list), you can just append directly - if the key doesn't exist, it automatically creates an empty list first.

Another common scenario is counting: instead of checking if a key exists before incrementing a counter, you can use defaultdict(int) and increment directly, since missing keys default to 0.

The beauty of defaultdict is that it eliminates boilerplate conditional logic, making your code cleaner and often more readable!

Usage

Here's a basic example showing the difference between regular dict and defaultdict:

from collections import defaultdict

# regular dict - requires existence checks
d = {}
if "fruits" not in d:
    d["fruits"] = []
d["fruits"].append("apple")
d["fruits"].append("banana")

# another way to do it using traditional dict
# but this isn't very pythonic and less readable as well
d.setdefault("fruits", []).append("apple")
d.setdefault("fruits", []).append("banana")

# defaultdict - no checks needed
dd = defaultdict(list)
dd["fruits"].append("apple")  # just works!
dd["fruits"].append("banana")
print(dd)
# [Out]: defaultdict(<class 'list'>, {'fruits': ['apple', 'banana']})

A bit about dict internals and default_factory

In the standard dict implementation, when we do a lookup like d["key"], it calls the __getitem__ method. If this method cannot find the key, it calls the __missing__ method.

In a regular dict, this then raises a KeyError. In a defaultdict however, it calls the default_factory callable (which is the first argument passed to defaultdict as stated above). This callable is then executed, and its return value is assigned to key and is also returned.

The default_factory can be any callable, including builtins, functions, lambdas, None or even left blank. If None or nothing is passed, then upon lookup for a non-existent key, a KeyError is raised, like a regular dict.

As this callable decides what is returned when a non-existent key is looked up for, it must be a callable that can be called without any arguments. Then, the return value of this callable is assigned to the key.

Consider some examples:

from collections import defaultdict

# callable as `list`
dd = defaultdict(list)
print(dd["foo"])
# [Out]: []

# callable as `int`
dd = defaultdict(int)
print(dd["foo"])
# [Out]: 0

# callable as a lambda
dd = defaultdict(lambda: {"bar": []})
print(dd["foo"])
# [Out]: {'bar': []}

# callable as `None`
dd = defaultdict(None)
print(dd["foo"])
# [Out]: KeyError 'foo'

# callable as a function
def cust():
    return "DEFAULT"
dd = defaultdict(cust)
print(dd["foo"])
# [Out]: DEFAULT

The default_factory argument can be any callable that returns a value. This can be used to demonstrate various use cases, viz.:

To directly append into a list:

dd = defaultdict(list)
dd["items"].append(1)
print(dd["items"])
# [Out]: [1]

To count grouped objects:

dd = defaultdict(int)
for i in range(11):
    if i % 2 == 0:
        dd["even"] += 1
    else:
        dd["odd"] += 1
print(dd)
# [Out]: defaultdict(<class 'int'>, {'even': 6, 'odd': 5})

Set custom default values:

dd = defaultdict(lambda: "N/A")
print(dd["missing_key"])
# [Out]: N/A

Nested (2D) data structures:

dd = defaultdict(lambda: defaultdict(int))
dd["user1"]["score"] += 10
dd["user1"]["score"] += 5
print(dd["user1"]["score"])
# [Out]: 15

Real-life scenarios

Here are some practical examples where defaultdict shines.

Grouping items by attribute

One of the most common uses - grouping a list of items by some property:

from collections import defaultdict

# group users by their department
users = [
    {"name": "Alice", "dept": "Engineering"},
    {"name": "Bob", "dept": "Sales"},
    {"name": "Charlie", "dept": "Engineering"},
    {"name": "Diana", "dept": "HR"},
    {"name": "Eve", "dept": "Sales"},
]

by_dept = defaultdict(list)
for user in users:
    by_dept[user["dept"]].append(user["name"])

print(dict(by_dept))
# [Out]: {
#   'Engineering': ['Alice', 'Charlie'],
#   'Sales': ['Bob', 'Eve'],
#   'HR': ['Diana']
# }

Building an inverted index

When working with search functionality, you often need to build an inverted index - mapping words to document IDs:

from collections import defaultdict

documents = {
    1: "Learn Python",
    2: "Webdev with Python",
    3: "Webdev with NodeJS",
}

# build inverted index: word -> set of doc IDs
iindex = defaultdict(set)
for doc_id, content in documents.items():
    for word in content.split():
        iindex[word.lower()].add(doc_id)

print(dict(iindex))
# [Out]: {
#    'learn': {1},
#    'python': {1, 2},
#    'webdev': {2, 3},
#    'with': {2, 3},
#    'nodejs': {3}
# }

# now you can quickly find documents containing a word
print(iindex["python"])
# [Out]: {1, 2}

Counting occurrences with grouping

A scenario I encounter frequently - counting events or items with multiple dimensions:

import json
from collections import defaultdict

# log entries with user and action
logs = [
    ("alice", "login"),
    ("bob", "login"),
    ("alice", "view_page"),
    ("alice", "view_page"),
    ("bob", "logout"),
    ("alice", "login"),
]

# count actions per user
user_actions = defaultdict(lambda: defaultdict(int))
for user, action in logs:
    user_actions[user][action] += 1

print(json.dumps(user_actions, indent=4))
# [Out]: {
#     "alice": {
#         "login": 2,
#         "view_page": 2
#     },
#     "bob": {
#         "login": 1,
#         "logout": 1
#     }
# }

Accumulating values for API response aggregation

When aggregating data from multiple API calls or database queries:

from collections import defaultdict

# simulating results from multiple API calls
api_responses = [
    {'user_id': 1, 'purchase': 100},
    {'user_id': 2, 'purchase': 50},
    {'user_id': 1, 'purchase': 200},
    {'user_id': 3, 'purchase': 75},
    {'user_id': 2, 'purchase': 30},
]

# aggregate total purchases per user
user_totals = defaultdict(int)
for response in api_responses:
    user_totals[response["user_id"]] += response["purchase"]

print(dict(user_totals))
# [Out]: {1: 300, 2: 80, 3: 75}

Caveats

The `.get()` method of `dict` does NOT use the default factory

This is probably the most surprising behavior. When you use .get() on a defaultdict, it behaves exactly like a regular dict - it does NOT call the default factory.

From the official docs:

Note that __missing__() is not called for any operations besides __getitem__(). This means that get() will, like normal dictionaries, return None as a default rather than using default_factory.

In case you aren't aware, the __getitem__ is the method of dict that is called when we do a lookup using square brackets - any_dict["key"]. So what this means is dd["missing_key"] (where dd is a defaultdict) will call the default_factory, but dd.get("missing_key") will not call the default_factory:

from collections import defaultdict

dd = defaultdict(list)

# using bracket notation - triggers default factory
print(dd[1])
# [Out]: []


# using .get() - does NOT trigger default factory!
print(dd.get(2))
# [Out]: None

# similar to `dict`, `.get` also does not add an entry
print(dd)
# [Out]: defaultdict(<class 'list'>, {1: []})

# and it works just like in the usual dict
print(dd.get(3, []))
# [Out]: []

Passing `None` or nothing as default_factory

If you create a defaultdict without a factory function (or with None), it behaves like a regular dict when accessing missing keys - you get a KeyError:

from collections import defaultdict

dd1 = defaultdict()
print(dd1[1])
# [Out]: KeyError 1

dd2 = defaultdict(None)
print(dd2[1])
# [Out]: KeyError 1

# But this works
dd3 = defaultdict(lambda: None)  # [1]
print(dd3[1])
# [Out]: None
print(dd3)  # key WAS added this time
# [Out]: defaultdict(<function <lambda> at 0x...>, {1: None})

Note [1]: This works because here we are setting a default_factory to a callable lambda that just returns None when a new key is requested for.

Comparing defaultdict with regular dict

Two defaultdicts are equal to each other and to a regular dict if they all have the same items, even if they have different default_factory:

from collections import defaultdict

dd1 = defaultdict(list, {"a": 1})
dd2 = defaultdict(int, {"a": 1})   # different factory
regular = {"a": 1}

print(dd1 == dd2)
# [Out]: True

print(dd1 == regular)
# [Out]: True

# but watch out for empty defaultdicts with different factories
dd3 = defaultdict(list)
dd4 = defaultdict(int)

print(dd3 == dd4)  # both empty, so equal
# [Out]: True

This means, default_factory is not considered for equality check.

Calling `in` on the dict does not trigger `default_factory`

The in operator checks for key existence but does NOT trigger the default factory:

from collections import defaultdict

dd = defaultdict(list)

# "in" operator doesn't trigger factory
print("a" in dd)
# [Out]: False

print(dd)  # still empty
# [Out]: defaultdict(<class 'list'>, {})

# but bracket access does
_ = dd["a"]
print("a" in dd)
# [Out]: True

This actually makes sense - you don't want membership testing to have side effects!

Unexpected mutations during iteration

While iterating over a defaultdict, accessing non-existent keys during iteration triggers the default_factory. This is obvious, as we know that a lookup triggers __missing__() method, which in turn, calls the default_factory that creates the entry.

from collections import defaultdict

dd = defaultdict(int)
dd["x"] = [1]

for key in ["x", "y", "z"]:
    print(key, dd[key])  # this would add `y` and `z` to `dd`!

# instead, check first or use .get()
for key in ["x", "y", "z"]:
    if key in dd:
        print(key, dd[key])

Python Core Utility - `functools.partial`

Shodhan Save — Fri, 14 Nov 2025 15:56:49 +0000

Intro

In this post, we will look at the partial function in the functools module.

The functools.partial function accepts a callable, with arguments (positional and keyword), and returns a partial object that has not yet executed the callable. When this partial object is called, it then passes the arguments to the callable and returns the output. The idea is to have an object that can store even some of the arguments required by the callable, such that it can be executed in the future. Think of it like taking the callable, with any arguments that you can pass, and then freezing it for a later point.

Use cases

The first use case as mentioned above is when you do not have all the arguments required by a callable, and for some reason, you need to have an object ready with whatever partial arguments you have. One reason could be to pass this partial object to another function, which holds other arguments.

Another use case (which I encounter more) is when a callable with fixed values for one or more arguments is to be used by a set of objects, and with other values for another set of objects. Rather than creating a function or a lambda, I prefer partial and it feels more pythonic to me (see this example below).

In both the cases, it simplifies the signature of the callable, because now the code only needs to call the partial object with the remaining arguments.

Let's see some code for each of these use cases.

Usage

Consider a function that accepts multiple arguments, and we wish to create a partial object and pass around, populating arguments at different points in the code:

from functools import partial


def foo(x, y, *z, k=0):
    print(x, y, z, k)


##############################
# using `foo` from elsewhere #
##############################

# regular call, without `partial`
foo(1, 2, 3, 4, k=5)
# [Out]: 1 2 (3, 4) 5


# create a partial object with just the first argument
p = partial(foo, 1)

# add more arguments [1]
p = partial(p, 2)

# try executing the partial object [2]
p()
# [Out]: 1 2 () 0

# add more arguments
p = partial(p, 3, 4)    # or, partial(p, *(3, 4))
p()
# [Out]: 1 2 (3, 4) 0

# you can pass kwargs at any point, though
p = partial(p, k=5)
p()
# [Out]: 1 2 (3, 4) 5

A couple of points to note about the code:

[1] Probably obvious, but in order to pass more arguments to the original callable (foo in this case), you need to pass it to the partial object.
[2] The partial object can be called multiple times, it's not like a generator object that will lose the value once called.

Real-life scenarios

Here are some real-life examples where I was able to find partial helpful.

Reusing config object in different 'modes'

I once had to design a module for storing configuration values that were to be read from either the OS environment, or an env file, or from constants defined elsewhere, or default to some value if none of the above had it (more on this in another post). For this purpose I used the python-decouple library.

python-decouple in a nutshell
Using this library, you can define a config variable in your file:

##################
# test_config.py #
##################
from decouple import config

PG_PORT = config("PG_PORT", default=5432, cast=int)
print(f"PG_PORT is {PG_PORT}")

Then you can call this file in these three ways:

$ python test_config.py
PG_PORT is 5432

$ PG_PORT=5433 python test_config.py
PG_PORT is 5433

$ echo "PG_PORT=5434" >> .env
$ python test_config.py
PG_PORT is 5434

The use cases here were that some config variables were to be mandatory (meaning, the system should throw an exception if they didn't exist), and some optional. In python-decouple, setting default=None makes it optional.

Now, I didn't want to write default=None for all the variables that were to be optional, and this is where partial came in handy. I defined a partial object that already had default=None passed to it, and then just used that object to define optional config variables. Similarly, I also defined another partial object for config variables that should default to False:

from functools import partial

from decouple import config


opt_config = partial(config, default=None)
falsy_config = partial(config, default=False)

# mandatory vars
PG_HOST = config("PG_HOST")
PG_PORT = config("PG_PORT")
PG_PASSWORD = config("PG_PASSWORD")
...

# optional vars
SENTRY_DSN = opt_config("SENTRY_DSN")
HTTPS_PROXY = opt_config("HTTPS_PROXY")
...

# default False vars
DEBUG = falsy_config("DEBUG")
SEND_EMAIL = falsy_config("SEND_EMAIL")
...

Setting up FastAPI router

This one was when I had a FastAPI app with only some routes having the prefix /v1. Routes like health-check, status, app-info need not be under version prefix, but routes with CRUD operations on resources had to be (see the FastAPI documentation on this topic).
Here I created a partial object from app.include_router that included the prefix. Then I could use this object to define routes that would have the prefix:

from functools import partial

from fastapi import Depends, FastAPI

from app.api import deps
from app.api.api_v1.routers import inventory, orders, probes, users
from app.api.probing.routers import probes

APP_VERSION = "v1"
ROUTE_PREFIX = f"/api/{APP_VERSION}"

app = FastAPI(
    title="My Web App",
    description="This is nice app.",
    version=APP_VERSION,
)

# probing-related routes without any prefix
app.include_router(probes.router, tags=["probes"])

# routes with prefix
route_include = partial(
    app.include_router,
    prefix=ROUTE_PREFIX,
    dependencies=[Depends(deps.authorize)],
)
route_include(users.router, tags=["users"])
route_include(orders.router, tags=["orders"])
route_include(inventory.router, tags=["inventory"])

Uploading to S3

There was a situation where I had to upload different files to different S3 buckets. I could use partial to have different objects for each bucket:


# function to upload files to S3
def upload_to_s3(client, bucket_name, path, file):
    ...

# from elsewhere

_upload_to_b = partial(upload_to_s3, env.client)
_upload_to_b1 = partial(_upload_to_b, env.bucket_1, env.path_1)
_upload_to_b2 = partial(_upload_to_b, env.bucket_2, env.path_2)
_upload_to_b3 = partial(_upload_to_b, env.bucket_3, env.path_3)

# few lines later...

# call relevant partial object for files
_upload_to_b1(file1)
_upload_to_b2(file2)
_upload_to_b3(file3)

Caveats

Different than functions

partial objects are slightly different than function objects, in that, you cannot access to the callable function's __name__ and __doc__ attributes. In order to access them you need to do it via the .func attribute of the partial:

def foo(x, y):
    """My fav func."""
    return x, y

p = partial(foo, 1)

p.func.__name__
# [Out]: 'foo'

p.func.__doc__
# [Out]: 'My fav func.'

See this for more information.

Incomplete or repetitive arguments

If you create a partial without all the mandatory arguments, it will not throw error at that point, but when you call it. This is obvious because at the time of creation, the partial is not executing the callable, just freezing it in memory:

def foo(x, y):
    return x, y

# this will not throw error
p = partial(foo, 1)

# this will throw error
p()

If you create a partial with all the mandatory arguments, then you cannot pass any more arguments to it - else, it will throw error - again, not while defining the partial, but while executing it:

def foo(x, y):
    return x, y

p = partial(foo, 1, 2)

p()
# [Out]: (1, 2)

# this will not throw error yet
p2 = partial(p, 3)

# this will throw error now
p2()

# this will also throw error
p(3)

Behaviour with variable (`*args`) and keyword arguments (`**kwargs`)

If you are trying to create a partial out of a function that accepts *args and **kwargs, then there are few points that you need to take care of.

Consider the below code:

def foo(x, y, *args, **kwargs):
    return x, y, args, kwargs

p = partial(foo, 1, 2, 3, 4, p=5)

p()
# [Out]: (1, 2, (3, 4), {'p': 5})

p(6)
# [Out]: (1, 2, (3, 4, 6), {'p': 5})

p2 = partial(p, 6)
p2()
# same as above
# [Out]: (1, 2, (3, 4, 6), {'p': 5})

p(q=7)
# [Out]: (1, 2, (3, 4), {'p': 5, 'q': 7})

p(p=7)
# [Out]: (1, 2, (3, 4, 6), {'p': 7})

Notice how the behaviour is, with *args and **kwargs. Ignore the positional arguments x and y, they are here just for completion.

If you pass an argument as positional to the partial, it gets appended to the *args. This can be seen when we execute p(6).
If you pass a keyword argument, it gets updated in the kwargs dict. This means, if you pass value to an existing keyword, it will replace the previous one, but if you pass a new keyword it will get added to the kwargs dict.

Working with classes

A partial cannot be defined as a method inside a class:

class Foo:
    def __init__(self, x):
        self.x = x
        self.y = {}

    def add_to_y(self, **kwargs):
        self.y.update(kwargs)

    p = partial(add_to_y, foo="bar")

f = Foo(1)

# throws error
f.p()

To make this work, you need partialmethod. Simply replace partial with partialmethod:

class Foo:
    def __init__(self, x):
        self.x = x
        self.y = {}

    def add_to_y(self, **kwargs):
        self.y.update(kwargs)

    p = partialmethod(add_to_y, foo="bar")

f = Foo(1)

f.p()

f.y
# [Out]: {'foo': 'bar'}

Python Core Utilities

Shodhan Save — Fri, 14 Nov 2025 15:49:20 +0000

Hello folks! 👋

Python comes with a lot of cool and helpful modules, data types, functions, which may not be very well known.

In this series, we are going to see some of the lesser known built-ins and standard libraries that I came across in my time of doing Python. We will look into the details, real-life examples, nuances, internal workings, and some caveats or gotchas as well. Of course, there are still quite some that I haven’t had the chance to deal with, and some that might be too common; let’s figure out which ones to include, as we move ahead in the series.

So, let’s get straight to it! 🚀

Python's "all" and "any" keywords, and what happens when an empty iterable is passed

Shodhan Save — Sun, 27 Dec 2020 10:04:49 +0000

Hello, world! 👋🏼

In this short post, let me take you through the happy li'l any and all keywords, along with a nice gotcha later!

Already aware of these keywords? Jump to the gotcha then.

The `any` keyword

For the ones new to Python, both any and all accept an iterable.

A bit about iterables
In case you are wondering what an iterable is, it's just an object that you can iterate over. Common examples include list, tuple, even dict and str (yes, you can iterate over them - try a for loop over them). There's more to iterables; some other day..

any simply returns True if any of the elements in the iterable are truthy. Else, it returns False.

More about truthy truthy or falsy are simply those objects that evaluate to True or False when considered as boolean. To quickly check this, consider the below snippet:
>> bool(1)
True
>> bool('a')
True
>> bool(' ')
True
>> bool(True)
True
>> bool(0)
False
>> bool('')
False
>> bool(None)
False
>> bool(False)
False

Let's see a few examples:

>>> any([0, 1])
True
>>> any([1, 2])
True
>>> any([None, 1])
True
>>> any([0, 0])
False
>>> any([None, 0])
False
>>> any([False, None, 0])
False

The `all` keyword

As you might have guessed, all returns True if all the elements in the iterable are truthy, else, False.

Let's see this with the same examples as above:

>>> all([0, 1])    # this returned True for "any"
False
>>> all([1, 2])
True
>>> all([None, 1])
False
>>> all([0, 0])
False
>>> all([None, 0])
False
>>> all([False, None, 0])
False

Notice why only [1, 2] returned True for all..

Gotcha

Now that we know what these keywords do, think about what will they return when you pass them an empty iterable?

Think for a while before going ahead..

Let's check:

>>> any([])
False
>>> 
>>> all([])
True
>>>

So, when we pass an empty list to any, it returns False (may be because an empty list is considered as falsy - quite intuitive). But behold! all returns True for an empty list!! Not very intuitive, eh?!

How could one ever know this? May be, by looking at the help on these functions:

>>> help(any)
Help on built-in function any in module builtins:

any(iterable, /)
    Return True if bool(x) is True for any x in the iterable.

    If the iterable is empty, return False.

>>> help(all)
Help on built-in function all in module builtins:

all(iterable, /)
    Return True if bool(x) is True for all values x in the iterable.

    If the iterable is empty, return True.

Note the last lines in both sections -
any - If the iterable is empty, return False.
all - If the iterable is empty, return True.

As a good practice, and towards becoming better at Python (or any language in general), always consider looking up at the documentation - it gives good insights about how things are implemented in the language.

Happy coding..

**kwargs, you got me!

Shodhan Save — Sat, 26 Dec 2020 16:17:43 +0000

Hello, world! 👋🏼

Being a Python developer since quite a few years now, this language still doesn't stop to teach me!

The other day I was refactoring a function that took just kwargs as a parameter. We wanted to make functions more readable, so we thought of having few important entries from kwargs as positional arguments.

So this:

def foo(**kwargs):
    print(kwargs)

turned into this:

def foo(status, **kwargs):
    print(kwargs)

Of course this requires that kwargs should have the status key, otherwise it would throw an error

TypeError: foo() missing 1 required positional argument: 'status'

After making this sure, things looked pretty readable; until a test case failed:

KeyError: 'status'

Wait. What? How?? I just changed the arguments, not a single line of code change inside the function..?
And yet, I got KeyError for kwargs['status']

At first, this seemed as if it should not have had happened. But then, it did! Going back to the REPL and trying out few things quickly shed light onto a thing that I didn't know about kwargs till that day.

Consider the below code:

>>> d = {'a': 'A', 'b': 'B'}
>>> 
>>> def foo(x, **kwargs):
...     print(x, kwargs)
... 
>>> def bar(x, c, **kwargs):
...     print(x, c, kwargs)
... 
>>>

Now let's see what happens when we call these functions:

>>> foo(1, **d)
1 {'a': 'A', 'b': 'B'}
>>>

>>> bar(1, **d)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: bar() missing 1 required positional argument: 'c'
>>>

This error is the same as mentioned above. We solve this by making sure the key is present in the dict:

>>> d['c'] = 'C'
>>>

And call the function again:

>>> bar(1, **d)
1 C {'a': 'A', 'b': 'B'}
>>>

Notice what happens when we print kwargs - the key that we accepted as a positional argument (c) got popped out of kwargs! It was no more part of kwargs, which explains why we got the KeyError above - because before refactoring, the key status was a part of kwargs, which, after refactoring, was coming in as an individual argument and was no more a part of kwargs.

It's a small thing, but can go unnoticed. Thanks to test cases, we were saved from creating another incident report..

DEV Community: Shodhan Save

Python Core Utility - `contextlib.contextmanager`

Intro

Use cases

Usage

Real-life scenarios

Database transaction management

Temporary environment variables

A special context manager (nullcontext) to parametrize pytest fixtures

Managing multiple resources

Async context managers

Nuances

Must use try/finally if you need cleanup

Only one yield allowed

Exceptions before yield

Cannot suppress exceptions by default

Nested context managers can be tricky

Reusability

Other types of context managers

Python Core Utility - `itertools.groupby`

Intro

Use cases

Usage

Real-life scenarios

Processing sorted log entries

Run-length encoding

Grouping database query results

Caveats

Only groups consecutive elements

Not consuming the group before moving on

Advancing the groupby iterator invalidates previous groups

Memory considerations with large groups

Python Core Utility - `collections.defaultdict`

Intro

Use cases

Usage

Real-life scenarios

Grouping items by attribute

Building an inverted index

Counting occurrences with grouping

Accumulating values for API response aggregation

Caveats

The .get() method of dict does NOT use the default factory

Passing None or nothing as default_factory

Comparing defaultdict with regular dict

Calling in on the dict does not trigger default_factory

Unexpected mutations during iteration

Python Core Utility - `functools.partial`

Intro

Use cases

Usage

Real-life scenarios

Reusing config object in different 'modes'

Setting up FastAPI router

Uploading to S3

Caveats

Different than functions

Incomplete or repetitive arguments

Behaviour with variable (*args) and keyword arguments (**kwargs)

Working with classes

Python Core Utilities

Python's "all" and "any" keywords, and what happens when an empty iterable is passed

The any keyword

The all keyword

Gotcha

**kwargs, you got me!

A special context manager (`nullcontext`) to parametrize pytest fixtures

Exceptions before `yield`

The `.get()` method of `dict` does NOT use the default factory

Passing `None` or nothing as default_factory

Calling `in` on the dict does not trigger `default_factory`

Behaviour with variable (`*args`) and keyword arguments (`**kwargs`)

The `any` keyword

The `all` keyword