In transformer models, the attention block is typically followed by a feed forward layer (FF), which is a simple fully-connected NN with a hidden layer and nonlinearity. Here's the code for such a block that uses ReLU:

def feed_forward_relu(x, W1, W2):
    """Feed-forward layer with ReLU activation.

    Args:
        x: Input tensor (B, N, D).
        Wh: Weights for the hidden layer (D, DH).
        Wo: Weights for the output layer (DH, D).

    Returns:
        Output tensor (B, N, D).
    """
    x = x @ W1  # hidden layer (B, N, DH)
    x = np.maximum(0, x)  # ReLU activation (B, N, DH)
    x = x @ W2  # output layer (B, N, D)
    return x

This layer typically holds most of the weights in the transformer, because the hidden dimension (DH in this post, hidden_dim in some papers) is large - 4x the embedding depth D is common. Intuitively, this makes sense because this layer does the majority of the heavy lifting; while the attention block mixes the embeddings of tokens together to express their relationships to one another, the FF block does the actual "reasoning" on these tokens.

Transformer blocks are repeated dozens of times in a model, so the total size of these layers becomes problematic. One approach for improving efficiency that became very popular is called sparsely-gated mixture of experts (paper).

Mixture of Experts architecture (MoE)

The basic idea of MoE is this:

• The large FF layer is split into a number (NEXP) of blocks called "experts". Each expert is still a FF layer. It takes a vector of size D and transforms it to another vector of size D.
• There's an additional piece called "router" or "gate". This is just a fully-connected layer (D, NEXP) that takes a token and produces a score for each expert. The router is learned by the model, along with the experts themselves.
• K experts with the highest scores are selected for each token, and the token is only fed through these experts.
• The scores are also used to calculate a weighted average from the experts' outputs, eventually producing an answer of size D.

Here's a diagram for a single token, assuming NEXP=8 and TOPK=2 (the two highest scoring experts are selected for each token, out of a total of eight):

Notes:

• Experts #1 and #5 are selected because the router produced the highest scores for them among all the experts. The input token is routed to these experts, but not to the others.
• The output of each expert is element-wise multiplied by a corresponding weight, calculated from the scores of the selected experts using a softmax function (to ensure balanced weighting across multiple tokens and experts).
• The weighted expert outputs are added up for the final output of the layer.

The key point to understand about this architecture is: the experts that were not among the top K for a token aren't used at all - the computation required to propagate the token through these experts is eschewed (both on the forward and backward passes).

This is the goal of the MoE architecture - we increase the overall model size, but keep the computational cost in check by only using a portion of the parameters for every single token. This is also reflected in the models' names; for example, the Mixtral model has size 8x7B; it has 8 experts, and it would be incorrect to just multiply the size of each expert by 8 because not all these parameters participate in the calculation of every token [1]. According to the Mixtral paper, the model only uses 13B active parameters for each token.

A summary of the idea behind MoE is:

MoE increases the model's capacity without proportionally increasing its computational cost.

# Parameters for a feed-forward layer with a fixed activation function.
@dataclass
class FFParams:
    Wh: np.ndarray
    Wo: np.ndarray


# Parameters for a Mixture of Experts (MoE) layer.
@dataclass
class MoEParams:
    # Embedding dimension of each token (a.k.a. model dimension, Dmodel)
    D: int

    # Hidden dimension in FF layers
    DH: int

    # Total number of experts
    NEXP: int

    # K in the top-k selection of top experts per token
    TOPK: int

    # List of experts: each expert is a forward layer with FFParams.
    ff_weights: List[FFParams]

    # Router weights: a linear layer (D, NEXP) that maps input to expert scores.
    router_weights: np.ndarray

def moe(x: np.ndarray, params: MoEParams):
    """Mixture of Experts (MoE) layer.

    Args:
        x: Input tensor (B, N, D).
        params: MoEParams.

    Returns:
        Output tensor (B, N, D).
    """
    # Run input through router to get expert scores for each token.
    expert_scores = x @ params.router_weights  # (B, N, NEXP)

    # Select the top-k expert scores and their indices for each token.
    top_scores, top_experts = topk_lastdim(expert_scores, params.TOPK)  # (B, N, TOPK)

    # Apply softmax to the top scores to get weights that sum to 1.
    weights = softmax_lastdim(top_scores)  # (B, N, TOPK)

    out = np.zeros_like(x)
    for b in range(x.shape[0]):
        for n in range(x.shape[1]):
            # Unvectorized implementation: for each token in the batch and
            # sequence, select the top-k experts and apply them with the
            # calculated weights.
            for expert_idx, weight in zip(top_experts[b, n], weights[b, n]):
                expert = params.ff_weights[expert_idx]
                out[b, n] += weight * feed_forward_relu(x[b, n], expert.Wh, expert.Wo)

    return out

def topk_lastdim(x, k):
    """Get the top k elements and their indices.

    x is an arbitrary array with at least two dimensions. The returned
    array has the same shape as x, but its elements are the top k elements
    across the last dimension. The indices of the top k elements are also
    returned.
    """
    idx = np.argpartition(x, -k, axis=-1)[..., -k:]
    return np.take_along_axis(x, idx, axis=-1), idx

def softmax_lastdim(x):
    """Compute softmax across last dimension of x.

    x is an arbitrary array with at least two dimensions. The returned array has
    the same shape as x, but its elements sum up to 1 across the last dimension.
    """
    # Subtract the max for numerical stability
    ex = np.exp(x - np.max(x, axis=-1, keepdims=True))
    # Divide by sums across last dimension
    return ex / np.sum(ex, axis=-1, keepdims=True)

18 Apr 2025 11:33pm GMT

Last year I updated my having a great first PyCon post to note that Mastodon would likely be more popular than Twitter at PyCon.

My guess was correct. During PyCon US 2024, Mastodon overtook Twitter for the most posts on the #PyConUS hashtag.

In the fall of 2024, Bluesky really took off. It currently seems like Bluesky is now a bit more popular than Mastodon for Python posts in general… but that doesn't necessarily mean it will be the most popular social media network during the conference.

This year I'm guessing that Mastodon will still be the most popular network for #PyConUS posts, though I wouldn't be very surprised if Bluesky took the lead instead.

What do the social networks think?

I decided to pose this question to the various social networks via a poll. I made sure to share these polls using both the #Python and #PyConUS hashtags for visibility's sake.

The results?

• LinkedIn poll: LinkedIn isn't sure whether Twitter is the new Twitter
• Twitter poll Twitter (mostly) knows that Bluesky is the new Twitter
• Bluesky poll Bluesky knows the really nerdy tech folks hang out on Mastodon
• Mastodon poll Mastodon loves itself above all else

In other words, LinkedIn leans more toward Bluesky being the leader than Mastodon, Twitter leans toward Bluesky, but Bluesky and Mastodon both lean toward Mastodon being the leader during PyCon.

I didn't mention Threads or other platforms because they didn't seem like real contenders given where the most active PyCon-attending Python-posting folks seem to hang out in 2025.

The actual results

Here are the actual results of the polls.

LinkedIn

Twitter

Bluesky

Mastodon

We're still fragmented

Unfortunately, regardless of which network is the leader this year during PyCon US, we're still going to be much more fragmented than we used to be. Twitter was the clear leader years ago and it very clearly isn't anymore… at least not for PyCon US conference chatter.

I'll be checking Mastodon and Bluesky and will post on at least Mastodon and possibly also Bluesky. I hope that other folks will also use one of these 2 social networks to organize dinners and gatherings! 🤞

Feel free to follow me on Mastodon and Bluesky during the conference.

Also let me know if you'd like to join my PyCon US starter pack on Bluesky. Lists on Mastodon require following and I prefer not to follow everyone I meet at PyCon so, unfortunately, I probably won't have a Mastodon equivalent of this. 😢

I recommend checking the #PyConUS hashtag on both networks as well. Note that you can subscribe to hashtags on Mastodon which is pretty neat!

See you at PyCon!

If this will be your first PyCon US, I recommend signing up for both Mastodon and Bluesky and checking the #PyConUS hashtag during the conference.

Also, be sure to see my post on having a great first PyCon and see this additional tips.

18 Apr 2025 6:45pm GMT

Python regexes have a number of features that bring new power to text manipulation. I'm not talking about fancy matching features like negative look-behinds, but ways you can construct and use regexes. As a demonstration, I'll show you some real code from a real project.

Coverage.py will expand environment variables in values read from its configuration files. It does this with a function called substitute_variables:

def substitute_variables(
text: str,
variables: dict[str, str],
) -> str:
"""
Substitute ``${VAR}`` variables in `text`.

Variables in the text can take a number of
shell-inspired forms::

$VAR
${VAR}
${VAR?} strict: an error if no VAR.
${VAR-miss} defaulted: "miss" if no VAR.
$$ just a dollar sign.

`variables` is a dictionary of variable values.

Returns the resulting text with values substituted.

"""

Call it with a string and a dictionary, and it makes the substitutions:

>>> substitute_variables(
... text="Look: $FOO ${BAR-default} $$",
... variables={'FOO': 'Xyzzy'},
... )

'Look: Xyzzy default $'

We use a regex to pick apart the text:

dollar_pattern = r"""(?x) # Verbose regex syntax
\$ # A dollar sign,
(?: # then
(?P<dollar> \$ ) | # a dollar sign, or
(?P<word1> \w+ ) | # a plain word, or
\{ # a {-wrapped
(?P<word2> \w+ ) # word,
(?: # either
(?P<strict> \? ) | # strict or
-(?P<defval> [^}]* ) # defaulted
)? # maybe
}
)
"""

This isn't a super-fancy regex: it doesn't use advanced pattern matching. But there are some useful regex features at work here:

• The (?x) flag at the beginning turns on "verbose" regex syntax. In this mode, all white space is ignored so the regex can be multi-line and we can indent to help see the structure, and comments are allowed at the ends of lines.
• Named groups like (?P<word1> … ) are used to capture parts of the text that we can retrieve later by name.
• There are also two groups used to get the precedence of operators right, but we don't want to capture those values separately, so I use the non-capturing group syntax for them: (?: … ). In this code, we only ever access groups by name, so I could have left them as regular capturing groups, but I think it's clearer to indicate up-front that we won't be using them.

The verbose syntax in particular makes it easier to understand the regex. Compare to what it would look like in one line:

r"\$(?:(?P<dollar>\$)|(?P<word1>\w+)|\{(?P<word2>\w+)(?:(?P<strict>\?)|-(?P<defval>[^}]*))?})"

Once we have the regex, we can use re.sub() to replace the variables with their values:

re.sub(dollar_pattern, dollar_replace, text)

But we're going to use another power feature of Python regexes: dollar_replace here isn't a string, it's a function! Each fragment the regex matches will be passed as a match object to our dollar_replace function. It returns a string which re.sub() uses as the replacement in the text:

def dollar_replace(match: re.Match[str]) -> str:
"""Called for each $replacement."""
# Get the one group that matched.
groups = match.group('dollar', 'word1', 'word2')
word = next(g for g in groups if g)

if word == "$":
return "$"
elif word in variables:
return variables[word]
elif match["strict"]:
msg = f"Variable {word} is undefined: {text!r}"
raise NameError(msg)
else:
return match["defval"]

First we use match.group(). Called with a number of names, it returns a tuple of what those named groups matched. They could be the matched text, or None if the group didn't match anything.

The way our regex is written only one of those three groups will match, so the tuple will have one string and two None's. To get the matched string, we use next() to find it. If the built-in any() returned the first true thing it found this code could be simpler, but it doesn't so we have to do it this way.

Now we can check the value to decide on the replacement:

• If the match was a dollar sign, we return a dollar sign.
• If the word is one of our defined variables, we return the value of the variable.
• Since the word isn't a defined variable, we check if the "strict" marker was found, and if so, raise an exception.
• Otherwise we return the default value provided.

The final piece of the implementation is to use re.sub() and return the result:

return re.sub(dollar_pattern, dollar_replace, text)

Regexes are often criticized for being too opaque and esoteric. But done right, they can be very powerful and don't have to be a burden. What we've done here is used simple pattern matching paired with useful API features to compactly write a useful transformation.

BTW, if you are interested, the real code is in coverage.py.

18 Apr 2025 3:19pm GMT

One of Django's most appreciated features is the built-in admin functionality. In fact, it was ranked as the most useful contrib app in the 2023 Django developer survey.

With a few lines of code, Django automatically generates an administrative interface to add, update, and edit objects in your database. While it's not meant to replace a full-featured frontend, the admin makes rapid prototyping possible and provides a lot of functionality out of the box.

However, the admin's focus is not on a flashy user interface and some people have found it to be a little plain - some have even called it ugly! But fortunately, like all Django applications, the admin's CSS and HTML templates can be overridden and tweaked. Here are a few projects which have done that, and are recently updated as of early 2025.

Chime in on the Django forum thread here with your favorite Django admin theme or if I missed any other options!

Note that these packages are listed in the order of the "easiest" integration to the hardest. However, the later libraries also tend to provide more features.

Dracula

A dark (and light) theme for the Django Admin based on the very popular Dracula which has themes for 400+ applications. This library is a quick win to give the admin a bit of pizazz without requiring much setup or changing the default admin functionality.

• Dracula Homepage
• Dracula Code

Django Daisy

Django Daisy is a responsive admin built with DaisyUI and TailwindCSS. Application icons can be added by utilizing Font Awesome. Very minimal (and completely optional!) configuration.

• Django Daisy Documentation
• Django Daisy Code
• Django Daisy Demo (demo/demo)

django-jazzmin

A drop-in theme for the Django admin that utilises AdminLTE 3.2 & Bootstrap 5. All of the configuration is optional which means the installation is very straight-forward. However, it also includes the ability to create custom menus, convert all pop-ups to modals, and a slick UI customizer. django-jazzmin also includes a wide selection of built-in themes.

• Django Jazzmin Documentation
• Django Jazzmin Code

django-admin-kubi

Kubi applies a Bootstrap 5 facelift to the Django admin, but also adds Sass support for custom styling and Font Awesome icons. It includes a sidebar menu for easy navigation and support for some third-party packages, including django-modeltranslation, django-modeltrans, django-import-export, django-two-factor-auth, and django-colorfield.

• Kubi Code

django-jet-reboot

Modern template for the Django admin interface with improved functionality. It provides the ability to create a custom dashboard and modules. Also includes user-selectable themes through the UI.

• Django JET Reboot Documentation
• Django JET Reboot Code

django-semantic-admin

A responsive Django admin theme based on Semantic UI. Includes JavaScript datepicker and timepicker components. Includes support for django_filter and django-import-export.

• Django Semantic UI Documentation
• Django Semantic UI Code
• Django Semantic UI Demo (admin/semantic)

Simple UI

A modern theme based on vue + element-ui which comes with 28 different themes. The documentation is originally in Chinese, but there is a translation in English.

• Simple UI Documentation
• Simple UI Code

Grapelli

Grappelli is a grid-based alternative to the Django admin which provides a few nifty features such as a custom TinyMCE integration, customizable dashboard, and inline sortables which can be updated by drag and drop.

• Grappelli Homepage
• Grappelli Documentation
• Grappelli Code

django-admin-interface

A modern responsive flat admin interface which comes with optional themes that can be installed for Bootstrap, Foundation, and U.S. Web Design Standards, and customizable by the admin itself. Other features include replacing admin pop-ups with modals, accordions in the navigation bar to collapse applications, sticky filters and buttons to prevent them from scrolling off the screen, and a language switcher. Also includes support for django-ckeditor, django-dynamic-raw-id, django-json-widget, django-modeltranslation, django-rangefilter, django-streamfield, django-tabbed-admin, and sorl-thumbnail.

• Django Admin Interface Code

Unfold

Unfold transforms the Django admin and is built with TailwindCSS. It includes custom widgets, pages, and admin sites. Also provides a language selector, conditional fields, custom filters, tabs, and additional features for actions. There are a lot of available settings and it is extremely customizable.

• Unfold Homepage
• Unfold Code

18 Apr 2025 2:19pm GMT

So, you're doing some I‍/‍O bound stuff, in parallel.

Maybe you're scraping some websites - a lot of websites.

Maybe you're deleting many millions of DynamoDB items.

You've got your ThreadPoolExecutor, you've increased the number of threads and tuned connection limits... but after some point, it's just not getting any faster. You look at your Python process, and you see CPU utilization hovers above 100%.

You could split the your work into batches and pass them to a ProcessPoolExecutor, which then runs your code in separate processes. But that requires yet more code, and a bunch of changes, which is no fun. And maybe your input is not that easy to split into batches.

If only we had an executor that works seamlessly across processes and threads.

Well, you're in luck, since that's exactly what we're building today!

And even better, in a couple years you won't even need it anymore.

• Establishing a baseline
  • Threads
  • Problem: CPU becomes a bottleneck
  • Processes?
  • Problem: processes use more memory
• Being unreasonable
  • A minimal plausible solution
  • Getting results
  • Fine, we'll make our own futures
  • Death becomes a problem
• Bonus: free threading

Establishing a baseline #

To measure things, we'll use a mock that pretends to do mostly I‍/‍O, with a sprinkling of CPU-bound work thrown in - a stand-in for something like a database connection, a Requests session, or a DynamoDB client.

class Client:
    io_time = 0.02
    cpu_time = 0.0008

    def method(self, arg):
        # simulate I/O
        time.sleep(self.io_time)

        # simulate CPU-bound work
        start = time.perf_counter()
        while time.perf_counter() - start < self.cpu_time:
            for i in range(100): i ** i

        return arg

We sleep() for the I‍/‍O, and do some math in a loop for the CPU stuff; it doesn't matter exactly how long each takes, as long I‍/‍O time dominates.

Real multi-threaded clients are usually backed by a connection pool; we could simulate one using a semaphore, but it's not relevant here - we're assuming the connection pool is effectively unbounded.

Since we'll use our client from multiple processes, we set up a global instance and a function that uses it; we can then pass init_client() as an executor initializer, which also allows us passing arguments to the client when creating it.

client = None

def init_client(*args):
    global client
    client = Client(*args)

def do_stuff(*args):
    return client.method(*args)

Finally, we make a simple timing context manager:

@contextmanager
def timer():
    start = time.perf_counter()
    yield
    end = time.perf_counter()
    print(f"elapsed: {end-start:1.3f}")

...and put everything together in a function that measures how long it takes to do a bunch of work using a concurrent.futures executor:

def benchmark(executor, n=10_000, timer=timer, chunksize=10):
    with executor:
        # make sure all the workers are started,
        # so we don't measure their startup time
        list(executor.map(time.sleep, [0] * 200))

        with timer():
            values = list(executor.map(do_stuff, range(n), chunksize=chunksize))

        assert values == list(range(n)), values

Threads #

So, a ThreadPoolExecutor should suffice here, since we're mostly doing I‍/‍O, right?

>>> from concurrent.futures import *
>>> from bench import *
>>> init_client()
>>> benchmark(ThreadPoolExecutor(10))
elapsed: 24.693

More threads!

>>> benchmark(ThreadPoolExecutor(20))
elapsed: 12.405

Twice the threads, twice as fast. More!

>>> benchmark(ThreadPoolExecutor(30))
elapsed: 8.718

Good, it's still scaling linearly. MORE!

>>> benchmark(ThreadPoolExecutor(40))
elapsed: 8.638

...more?

>>> benchmark(ThreadPoolExecutor(50))
elapsed: 8.458
>>> benchmark(ThreadPoolExecutor(60))
elapsed: 8.430
>>> benchmark(ThreadPoolExecutor(70))
elapsed: 8.428

Problem: CPU becomes a bottleneck #

It's time we take a closer look at what our process is doing.

I'd normally use the top command for this, but since the flags and output vary with the operating system, we'll implement our own using the excellent psutil library.

@contextmanager
def top():
    """Print information about current and child processes.

    RES is the resident set size. USS is the unique set size.
    %CPU is the CPU utilization. nTH is the number of threads.

    """
    process = psutil.Process()
    processes = [process] + process.children(True)
    for p in processes: p.cpu_percent()

    yield

    print(f"{'PID':>7} {'RES':>7} {'USS':>7} {'%CPU':>7} {'nTH':>7}")
    for p in processes:
        try:
            m = p.memory_full_info()
        except psutil.AccessDenied:
            m = p.memory_info()
        rss = m.rss / 2**20
        uss = getattr(m, 'uss', 0) / 2**20
        cpu = p.cpu_percent()
        nth = p.num_threads()
        print(f"{p.pid:>7} {rss:6.1f}m {uss:6.1f}m {cpu:7.1f} {nth:>7}")

And because it's a context manager, we can use it as a timer:

>>> init_client()
>>> benchmark(ThreadPoolExecutor(10), timer=top)
    PID     RES     USS    %CPU     nTH
  51395   35.2m   28.5m    38.7      11

So, what happens if we increase the number of threads?

>>> benchmark(ThreadPoolExecutor(20), timer=top)
    PID     RES     USS    %CPU     nTH
  13912   16.8m   13.2m    70.7      21
>>> benchmark(ThreadPoolExecutor(30), timer=top)
    PID     RES     USS    %CPU     nTH
  13912   17.0m   13.4m    99.1      31
>>> benchmark(ThreadPoolExecutor(40), timer=top)
    PID     RES     USS    %CPU     nTH
  13912   17.3m   13.7m   100.9      41

With more threads, the compute part of our I‍/‍O bound workload increases, eventually becoming high enough to saturate one CPU - and due to the global interpreter lock, one CPU is all we can use, regardless of the number of threads.¹

Processes? #

I know, let's use a ProcessPoolExecutor instead!

>>> benchmark(ProcessPoolExecutor(20, initializer=init_client))
elapsed: 12.374
>>> benchmark(ProcessPoolExecutor(30, initializer=init_client))
elapsed: 8.330
>>> benchmark(ProcessPoolExecutor(40, initializer=init_client))
elapsed: 6.273

Hmmm... I guess it is a little bit better.

More? More!

>>> benchmark(ProcessPoolExecutor(60, initializer=init_client))
elapsed: 4.751
>>> benchmark(ProcessPoolExecutor(80, initializer=init_client))
elapsed: 3.785
>>> benchmark(ProcessPoolExecutor(100, initializer=init_client))
elapsed: 3.824

OK, it's better, but with diminishing returns - there's no improvement after 80 processes, and even then, it's only 2.2x faster than the best time with threads, when, in theory, it should be able to make full use of all 4 CPUs.

Also, we're not making best use of connection pooling (relevant if the client connects to many different hosts, since we now have 80 pools), nor multiplexing (relevant with protocols like HTTP/2 or newer, since we now have 80 connections).

Problem: processes use more memory #

But it gets worse!

>>> benchmark(ProcessPoolExecutor(80, initializer=init_client), timer=top)
    PID     RES     USS    %CPU     nTH
   2479   21.2m   15.4m    15.0       3
   2480   11.2m    6.3m     0.0       1
   2481   13.8m    8.5m     3.4       1
  ... 78 more lines ...
   2560   13.8m    8.5m     4.4       1

13.8 MiB * 80 ~= 1 GiB ... that is a lot of memory.

Now, there's some nuance to be had here.

First, on most operating systems that have virtual memory, code segment pages are shared between processes - there's no point in having 80 copies of libc or the Python interpreter in memory.

The unique set size is probably a better measurement than the resident set size, since it excludes memory shared between processes.² So, for the macOS output above,³ the actual usage is more like 8.5 MiB * 80 = 680 MiB.

Second, if you use the fork or forkserver start methods, processes also share memory allocated before the fork() via copy-on-write; for Python, this includes module code and variables. On Linux, the actual usage is 1.7 MiB * 80 = 136 MiB:

>>> benchmark(ProcessPoolExecutor(80, initializer=init_client), timer=top)
    PID     RES     USS    %CPU     nTH
 329801   17.0m    6.6m     5.1       3
 329802   13.3m    1.6m     2.1       1
  ... 78 more lines ...
 329881   13.3m    1.7m     2.0       1

However, it's important to note that's just a lower bound; memory allocated after fork() is not shared, and most real work will unavoidably allocate more memory.

Being unreasonable #

One reasonable way of dealing with this would be to split the input into batches, one per CPU, and pass them to a ProcessPoolExecutor, which in turn runs the batch items using a ThreadPoolExecutor.⁴

But that would mean we need to change our code, and that's no fun. If only there was an executor that works seamlessly across processes and threads.

A minimal plausible solution #

In keeping with what has become tradition by now, we'll take an iterative, problem-solution approach; since we're not sure what to do yet, we start with the simplest thing that could possibly work.

We know we want a process pool executor that starts one thread pool executor per process, so let's deal with that first.

class ProcessThreadPoolExecutor(concurrent.futures.ProcessPoolExecutor):

    def __init__(self, max_threads=None, initializer=None, initargs=()):
        super().__init__(
            initializer=_init_process,
            initargs=(max_threads, initializer, initargs)
        )

By subclassing ProcessPoolExecutor, we get the map() implementation for free. By going with the default max_workers, we get one process per CPU (which is what we want); we can add more arguments later if needed.

In our custom process initializer, we set up a global thread pool executor, and then call the initializer provided by the user:

_executor = None

def _init_process(max_threads, initializer, initargs):
    global _executor

    _executor = concurrent.futures.ThreadPoolExecutor(max_threads)
    atexit.register(_executor.shutdown)

    if initializer:
        initializer(*initargs)

Likewise, submit() passes the work along to the thread pool executor:

class ProcessThreadPoolExecutor(concurrent.futures.ProcessPoolExecutor):
    # ...
    def submit(self, fn, *args, **kwargs):
        return super().submit(_submit, fn, *args, **kwargs)

def _submit(fn, *args, **kwargs):
    return _executor.submit(fn, *args, **kwargs).result()

OK, that looks good enough; let's use it and see if it works:

def _do_stuff(n):
    print(f"doing: {n}")
    return n ** 2

if __name__ == '__main__':
    with ProcessThreadPoolExecutor() as e:
        print(list(e.map(_do_stuff, [0, 1, 2])))

 $ python ptpe.py
doing: 0
doing: 1
doing: 2
[0, 1, 4]

Wait, we got it on the first try?!

Let's measure that:

>>> from bench import *
>>> from ptpe import *
>>> benchmark(ProcessThreadPoolExecutor(30, initializer=init_client), n=1000)
elapsed: 6.161

Hmmm... that's unexpectedly slow... almost as if:

>>> multiprocessing.cpu_count()
4
>>> benchmark(ProcessPoolExecutor(4, initializer=init_client), n=1000)
elapsed: 6.067

Ah, because _submit() waits for the result() in the main thread of the worker process, this is just a ProcessPoolExecutor with extra steps.

But what if we send back the future instead?

    def submit(self, fn, *args, **kwargs):
        return super().submit(_submit, fn, *args, **kwargs).result()

def _submit(fn, *args, **kwargs):
    return _executor.submit(fn, *args, **kwargs)

Alas:

$ python ptpe.py
doing: 0
doing: 1
doing: 2
concurrent.futures.process._RemoteTraceback:
"""
Traceback (most recent call last):
  File "concurrent/futures/process.py", line 210, in _sendback_result
    result_queue.put(_ResultItem(work_id, result=result,
  File "multiprocessing/queues.py", line 391, in put
    obj = _ForkingPickler.dumps(obj)
  File "multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
TypeError: cannot pickle '_thread.RLock' object
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "ptpe.py", line 42, in <module>
    print(list(e.map(_do_stuff, [0, 1, 2])))
  ...
TypeError: cannot pickle '_thread.RLock' object

It may not seem like it, but this is a partial success: the work happens, we just can't get anything back. Not surprising, to be honest, it couldn't have been that easy.

Getting results #

If you look carefully at the traceback, you'll find a hint of how ProcessPoolExecutor gets its own results back from workers - a queue; the module docstring even has a neat data-flow diagram:

|======================= In-process =====================|== Out-of-process ==|

+----------+     +----------+       +--------+     +-----------+    +---------+
|          |  => | Work Ids |       |        |     | Call Q    |    | Process |
|          |     +----------+       |        |     +-----------+    |  Pool   |
|          |     | ...      |       |        |     | ...       |    +---------+
|          |     | 6        |    => |        |  => | 5, call() | => |         |
|          |     | 7        |       |        |     | ...       |    |         |
| Process  |     | ...      |       | Local  |     +-----------+    | Process |
|  Pool    |     +----------+       | Worker |                      |  #1..n  |
| Executor |                        | Thread |                      |         |
|          |     +----------- +     |        |     +-----------+    |         |
|          | <=> | Work Items | <=> |        | <=  | Result Q  | <= |         |
|          |     +------------+     |        |     +-----------+    |         |
|          |     | 6: call()  |     |        |     | ...       |    |         |
|          |     |    future  |     |        |     | 4, result |    |         |
|          |     | ...        |     |        |     | 3, except |    |         |
+----------+     +------------+     +--------+     +-----------+    +---------+

Now, we could probably use the same queue somehow, but it would involve touching a lot of (private) internals.⁵ Instead, let's use a separate queue:

    def __init__(self, max_threads=None, initializer=None, initargs=()):
        self.__result_queue = multiprocessing.Queue()
        super().__init__(
            initializer=_init_process,
            initargs=(self.__result_queue, max_threads, initializer, initargs)
        )

On the worker side, we make it globally accessible:

_executor = None
_result_queue = None

def _init_process(queue, max_threads, initializer, initargs):
    global _executor, _result_queue

    _executor = concurrent.futures.ThreadPoolExecutor(max_threads)
    atexit.register(_executor.shutdown)

    _result_queue = queue
    atexit.register(_result_queue.close)

    if initializer:
        initializer(*initargs)

...so we can use it from a task callback registered by _submit():

def _submit(fn, *args, **kwargs):
    task = _executor.submit(fn, *args, **kwargs)
    task.add_done_callback(_put_result)

def _put_result(task):
    if exception := task.exception():
        _result_queue.put((False, exception))
    else:
        _result_queue.put((True, task.result()))

Back in the main process, we handle the results in a thread:

    def __init__(self, max_threads=None, initializer=None, initargs=()):
        # ...
        self.__result_handler = threading.Thread(target=self.__handle_results)
        self.__result_handler.start()

    def __handle_results(self):
        for ok, result in iter(self.__result_queue.get, None):
            print(f"{'ok' if ok else 'error'}: {result}")

Finally, to stop the handler, we use None as a sentinel on executor shutdown:

    def shutdown(self, wait=True):
        super().shutdown(wait=wait)
        if self.__result_queue:
            self.__result_queue.put(None)
            if wait:
                self.__result_handler.join()
            self.__result_queue.close()
            self.__result_queue = None

Let's see if it works:

$ python ptpe.py
doing: 0
ok: [0]
doing: 1
ok: [1]
doing: 2
ok: [4]
Traceback (most recent call last):
  File "concurrent/futures/_base.py", line 317, in _result_or_cancel
    return fut.result(timeout)
AttributeError: 'NoneType' object has no attribute 'result'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  ...
AttributeError: 'NoneType' object has no attribute 'cancel'

Yay, the results are making it to the handler!

The error happens because instead of returning a Future, our submit() returns the result of _submit(), which is always None.

Fine, we'll make our own futures #

But submit() must return a future, so we make our own:

    def __init__(self, max_threads=None, initializer=None, initargs=()):
        # ...
        self.__tasks = {}
        # ...

    def submit(self, fn, *args, **kwargs):
        outer = concurrent.futures.Future()
        task_id = id(outer)
        self.__tasks[task_id] = outer

        outer.set_running_or_notify_cancel()
        inner = super().submit(_submit, task_id, fn, *args, **kwargs)

        return outer

In order to map results to their futures, we need an unique identifier; the id() of the outer future should do, since it is unique for the object's lifetime.

We pass the id to _submit(), then to _put_result() as an attribute on the future, and finally back in the queue with the result:

def _submit(task_id, fn, *args, **kwargs):
    task = _executor.submit(fn, *args, **kwargs)
    task.task_id = task_id
    task.add_done_callback(_put_result)

def _put_result(task):
    if exception := task.exception():
        _result_queue.put((task.task_id, False, exception))
    else:
        _result_queue.put((task.task_id, True, task.result()))

Back in the result handler, we find the maching future, and set the result accordingly:

    def __handle_results(self):
        for task_id, ok, result in iter(self.__result_queue.get, None):
            outer = self.__tasks.pop(task_id)
            if ok:
                outer.set_result(result)
            else:
                outer.set_exception(result)

And it works:

$ python ptpe.py
doing: 0
doing: 1
doing: 2
[0, 1, 4]

I mean, it really works:

>>> benchmark(ProcessThreadPoolExecutor(10, initializer=init_client))
elapsed: 6.220
>>> benchmark(ProcessThreadPoolExecutor(20, initializer=init_client))
elapsed: 3.397
>>> benchmark(ProcessThreadPoolExecutor(30, initializer=init_client))
elapsed: 2.575
>>> benchmark(ProcessThreadPoolExecutor(40, initializer=init_client))
elapsed: 2.664

3.3x is not quite the 4 CPUs my laptop has, but it's pretty close, and much better than the 2.2x we got from processes alone.

Death becomes a problem #

I wonder what happens when a worker process dies.

For example, the initializer can fail:

>>> executor = ProcessPoolExecutor(initializer=divmod, initargs=(0, 0))
>>> executor.submit(int).result()
Exception in initializer:
Traceback (most recent call last):
  ...
ZeroDivisionError: integer division or modulo by zero
Traceback (most recent call last):
  ...
concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.

...or it can die some time later, which we can help with a custom timer:⁶

@contextmanager
def terminate_child(interval=1):
    threading.Timer(interval, psutil.Process().children()[-1].terminate).start()
    yield

>>> executor = ProcessPoolExecutor(initializer=init_client)
>>> benchmark(executor, timer=terminate_child)
[ one second later ]
Traceback (most recent call last):
  ...
concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.

Now let's see our executor:

>>> executor = ProcessThreadPoolExecutor(30, initializer=init_client)
>>> benchmark(executor, timer=terminate_child)
[ one second later ]
[ ... ]
[ still waiting ]
[ ... ]
[ hello? ]

There's actually two scenarios here:

1. if map() is still submitting tasks, inner will silently fail with BrokenProcessPool
2. if map() is already waiting for results, it'll keep waiting, forever

Either way, we want to fail all pending tasks with BrokenProcessPool.

For the first case, we don't need any special handling, since it's fixed by handling the second one. Nevertheless, we should still propagate inner exceptions to the outer task, otherwise we may ignore other exceptions.

    def submit(self, fn, *args, **kwargs):
        # ...
        inner = super().submit(_submit, task_id, fn, *args, **kwargs)
        inner.task_id = task_id
        inner.add_done_callback(self.__handle_inner)

        return outer

Of note, the task may have already been handled by the second case:

    def __handle_inner(self, inner):
        task_id = inner.task_id
        if exception := inner.exception():
            if outer := self.__tasks.pop(task_id, None):
                outer.set_exception(exception)

This already fixes the case where a worker dies almost instantly:

>>> executor = ProcessThreadPoolExecutor(30, initializer=init_client)
>>> benchmark(executor, timer=lambda: terminate_child(0))
Traceback (most recent call last):
  ...
concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.

For the second case, we need to check if the executor is broken - but how? ProcessPoolExecutor​._broken exists, but we decided we don't want to depend on internals. Maybe we can submit a dummy task and see if it fails:

    def __handle_results(self):
        while True:
            try:
                value = self.__result_queue.get(timeout=.1)
            except queue.Empty:
                try:
                    super().submit(int).cancel()
                except concurrent.futures.BrokenExecutor as e:
                    exc = type(e)(str(e))
                    break
                except RuntimeError as e:
                    assert 'shutdown' in str(e), e
                continue

            if not value:
                return

            task_id, ok, result = value
            if outer := self.__tasks.pop(task_id, None):
                if ok:
                    outer.set_result(result)
                else:
                    outer.set_exception(result)

        while self.__tasks:
            try:
                _, outer = self.__tasks.popitem()
            except KeyError:
                break
            outer.set_exception(exc)

We can use the Queue.get() timeout to submit the dummy task only if we're not receiving results fast enough. When the executor is broken, submit() fails, which breaks us out of the while loop where we can fail the pending tasks.

Like so:

>>> executor = ProcessThreadPoolExecutor(30, initializer=init_client)
>>> benchmark(executor, timer=terminate_child)
Traceback (most recent call last):
  ...
concurrent.futures.process.BrokenProcessPool: A child process terminated abruptly, the process pool is not usable anymore

So, yeah, I think we're done. Here's the final executor and benchmark code.

Some features are left as an exercise for the reader:

• providing an initializer for the ThreadPoolExecutor as well
• using other multiprocessing start methods
• shutdown()'s cancel_futures

Learned something new today? Share this with others, it really helps! PyCoder's Weekly HN Reddit linkedin Twitter

Want to know when new articles come out? Subscribe here to get new stuff straight to your inbox!

Bonus: free threading #

You may have heard people being excited about the experimental free threading support added in Python 3.13, which allows running Python code on multiple CPUs.

And for good reason:

$ python3.13t
Python 3.13.2 experimental free-threading build
>>> from concurrent.futures import *
>>> from bench import *
>>> init_client()
>>> benchmark(ThreadPoolExecutor(30))
elapsed: 8.224
>>> benchmark(ThreadPoolExecutor(40))
elapsed: 6.193
>>> benchmark(ThreadPoolExecutor(120))
elapsed: 2.323

3.6x over to the GIL version, with none of the shenanigans we've been up to!

Alas, packages with extensions need to be updated to support it:

>>> import psutil
zsh: segmentation fault  python3.13t

Soon.

1. At least, all we can use for pure-Python code. I‍/‍O always releases the global interpreter lock, and so do some extension modules. ^[return]

2. The psutil documentation for memory_full_info() explains the difference quite nicely and links to further resources, because good libraries educate. ^[return]

3. You may have to run Python as root to get the USS of child processes. ^[return]

4. And no, asyncio is not a solution, since the event loop runs in a single thread, so you'd still need to run one event loop per CPU in dedicated processes. ^[return]

5. Check out nilp0inter/threadedprocess for an idea of what that looks like. ^[return]

6. pkill -fn '[Pp]ython' would've done it too, but it gets tedious if you do it a lot, and it's a different command on Windows. ^[return]

18 Apr 2025 1:42pm GMT

Are you looking for a fast database that can handle large datasets in Python? What's the difference between a Python expression and a statement? Christopher Trudeau is back on the show this week, bringing another batch of PyCoder's Weekly articles and projects.

[ Improve Your Python With 🐍 Python Tricks 💌 - Get a short & sweet Python Trick delivered to your inbox every couple of days. >> Click here to learn more and see examples ]

18 Apr 2025 12:00pm GMT

We'll be at PyCon US 2025, and hope to see the Django community and all our Python friends there ❤️! We have been granted a community booth at the conference - come say hi in the Expo Hall during open hours. There may be Django stickers available to pick up!

Represent Django

For our Individual Members - if you'd like to help us showcase Django, we're looking for help staffing the booth (members-only forum)! This is a great opportunity to give back to support our project - consider it!

David, Kátia and Paolo representing the Foundation at EuroPython 2024.
Credit: Paolo Melchiorre (CC-BY-SA)

18 Apr 2025 7:07am GMT

FastHTML is built on Starlette, so we use Starlette's middleware tooling and then pass in the result. Just make sure you install pyinstrument.

WARNING: NOT FOR PRODUCTION ENVIRONMENTS Including a profiler like this in a production environment is dangerous. As it exposes infrastructure it is highly risky to include in any location where end users can access it.

"""WARNING: NOT FOR PRODUCTION ENVIRONMENTS"""
from fasthtml.common import *
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware import Middleware
try:
    from pyinstrument import Profiler
except ImportError:
    raise ImportError('Please install pyinstrument')

class ProfileMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        profiling = request.query_params.get("profile", False)
        if profiling:
            profiler = Profiler()
            profiler.start()
            response = await call_next(request)            
            profiler.stop()        
            return HTMLResponse(profiler.output_html())
        return await call_next(request)

app, rt = fast_app(middleware=(Middleware(ProfileMiddleware)))

@rt("/")
def get():
    return Titled("FastHTML", P("Hello, world!"))

serve()

To invoke, make any request to your application with the GET parameter profile=1 and it will print the HTML result from pyinstrument.

18 Apr 2025 2:17am GMT

Python comes with webbrowser, a library for opening webbrowsers to specific pages such as this one.

>>> import webbrowser
>>> url = 'https://daniel.feldroy.com/'
>>> webbrowser.open(url)

18 Apr 2025 2:17am GMT

The History

Before dataclasses were added to Python in version 3.7 - in June of 2018 - the __init__ special method had an important use. If you had a class representing a data structure - for example a 2DCoordinate, with x and y attributes - you would want to be able to construct it as 2DCoordinate(x=1, y=2), which would require you to add an __init__ method with x and y parameters.

The other options available at the time all had pretty bad problems:

1. You could remove 2DCoordinate from your public API and instead expose a make_2d_coordinate function and make it non-importable, but then how would you document your return or parameter types?
2. You could document the x and y attributes and make the user assign each one themselves, but then 2DCoordinate() would return an invalid object.
3. You could default your coordinates to 0 with class attributes, and while that would fix the problem with option 2, this would now require all 2DCoordinate objects to be not just mutable, but mutated at every call site.
4. You could fix the problems with option 1 by adding a new abstract class that you could expose in your public API, but this would explode the complexity of every new public class, no matter how simple. To make matters worse, typing.Protocol didn't even arrive until Python 3.8, so, in the pre-3.7 world this would condemn you to using concrete inheritance and declaring multiple classes even for the most basic data structure imaginable.

Also, an __init__ method that does nothing but assign a few attributes doesn't have any significant problems, so it is an obvious choice in this case. Given all the problems that I just described with the alternatives, it makes sense that it became the obvious default choice, in most cases.

However, by accepting "define a custom __init__" as the default way to allow users to create your objects, we make a habit of beginning every class with a pile of arbitrary code that gets executed every time it is instantiated.

Wherever there is arbitrary code, there are arbitrary problems.

The Problems

Let's consider a data structure more complex than one that simply holds a couple of attributes. We will create one that represents a reference to some I/O in the external world: a FileReader.

Of course Python has its own open-file object abstraction, but I will be ignoring that for the purposes of the example.

Let's assume a world where we have the following functions, in an imaginary fileio module:

• open(path: str) -> int
• read(fileno: int, length: int)
• close(fileno: int)

Our hypothetical fileio.open returns an integer representing a file descriptor¹, fileio.read allows us to read length bytes from an open file descriptor, and fileio.close closes that file descriptor, invalidating it for future use.

With the habit that we have built from writing thousands of __init__ methods, we might want to write our FileReader class like this:

1
2
3
4
5
6
7

class FileReader:
    def __init__(self, path: str) -> None:
        self._fd = fileio.open(path)
    def read(self, length: int) -> bytes:
        return fileio.read(self._fd, length)
    def close(self) -> None:
        fileio.close(self._fd)

For our initial use-case, this is fine. Client code creates a FileReader by doing something like FileReader("./config.json"), which always creates a FileReader that maintains its file descriptor int internally as private state. This is as it should be; we don't want user code to see or mess with _fd, as that might violate FileReader's invariants. All the necessary work to construct a valid FileReader - i.e. the call to open - is always taken care of for you by FileReader.__init__.

However, additional requirements will creep in, and as they do, FileReader.__init__ becomes increasingly awkward.

Initially we only care about fileio.open, but later, we may have to deal with a library that has its own reasons for managing the call to fileio.open by itself, and wants to give us an int that we use as our _fd, we now have to resort to weird workarounds like:

1
2
3
4

def reader_from_fd(fd: int) -> FileReader:
    fr = object.__new__(FileReader)
    fr._fd = fd
    return fr

Now, all those nice properties that we got from trying to force object construction to give us a valid object are gone. reader_from_fd's type signature, which takes a plain int, has no way of even suggesting to client code how to ensure that it has passed in the right kind of int.

Testing is much more of a hassle, because we have to patch in our own copy of fileio.open any time we want an instance of a FileReader in a test without doing any real-life file I/O, even if we could (for example) share a single file descriptor among many FileReader s for testing purposes.

All of this also assumes a fileio.open that is synchronous. Although for literal file I/O this is more of a hypothetical concern, there are many types of networked resource which are really only available via an asynchronous (and thus: potentially slow, potentially error-prone) API. If you've ever found yourself wanting to type async def __init__(self): ... then you have seen this limitation in practice.

Comprehensively describing all the possible problems with this approach would end up being a book-length treatise on a philosophy of object oriented design, so I will sum up by saying that the cause of all these problems is the same: we are inextricably linking the act of creating a data structure with whatever side-effects are most often associated with that data structure. If they are "often" associated with it, then by definition they are not "always" associated with it, and all the cases where they aren't associated become unweildy and potentially broken.

Defining an __init__ is an anti-pattern, and we need a replacement for it.

The Solutions

I believe this tripartite assemblage of design techniques will address the problems raised above:

• using dataclass to define attributes,
• replacing behavior that previously would have previously been in __init__ with a new classmethod that does the same thing, and
• using precise types to describe what a valid instance looks like.

Using dataclass attributes to create an __init__ for you

To begin, let's refactor FileReader into a dataclass. This does get us an __init__ method, but it won't be one an arbitrary one we define ourselves; it will get the useful constraint enforced on it that it will just assign attributes.

1
2
3
4
5
6
7

@dataclass
class FileReader:
    _fd: int
    def read(self, length: int) -> bytes:
        return fileio.read(self._fd, length)
    def close(self) -> None:
        fileio.close(self._fd)

Except... oops. In fixing the problems that we created with our custom __init__ that calls fileio.open, we have re-introduced several problems that it solved:

1. We have removed all the convenience of FileReader("path"). Now the user needs to import the low-level fileio.open again, making the most common type of construction both more verbose and less discoverable; if we want users to know how to build a FileReader in a practical scenario, we will have to add something in our documentation to point at a separate module entirely.
2. There's no enforcement of the validity of _fd as a file descriptor; it's just some integer, which the user could easily pass an incorrect instance of, with no error.

In isolation, dataclass by itself can't solve all our problems, so let's add in the second technique.

Using classmethod factories to create objects

We don't want to require any additional imports, or require users to go looking at any other modules - or indeed anything other than FileReader itself - to figure out how to create a FileReader for its intended usage.

Luckily we have a tool that can easily address all of these concerns at once: @classmethod. Let's define a FileReader.open class method:

1
2
3
4
5
6
7

from typing import Self
@dataclass
class FileReader:
    _fd: int
    @classmethod
    def open(cls, path: str) -> Self:
        return cls(fileio.open(path))

Now, your callers can replace FileReader("path") with FileReader.open("path"), and get all the same benefits.

Additionally, if we needed to await fileio.open(...), and thus we needed its signature to be @classmethod async def open, we are freed from the constraint of __init__ as a special method. There is nothing that would prevent a @classmethod from being async, or indeed, from having any other modification to its return value, such as returning a tuple of related values rather than just the object being constructed.

Using NewType to address object validity

Next, let's address the slightly trickier issue of enforcing object validity.

Our type signature calls this thing an int, and indeed, that is unfortunately what the lower-level fileio.open gives us, and that's beyond our control. But for our own purposes, we can be more precise in our definitions, using NewType:

1
2

from typing import NewType
FileDescriptor = NewType("FileDescriptor", int)

There are a few different ways to address the underlying library, but for the sake of brevity and to illustrate that this can be done with zero run-time overhead, let's just insist to Mypy that we have versions of fileio.open, fileio.read, and fileio.write which actually already take FileDescriptor integers rather than regular ones.

1
2
3
4

from typing import Callable
_open: Callable[[str], FileDescriptor] = fileio.open  # type:ignore[assignment]
_read: Callable[[FileDescriptor, int], bytes] = fileio.read
_close: Callable[[FileDescriptor], None] = fileio.close

We do of course have to slightly adjust FileReader, too, but the changes are very small. Putting it all together, we get:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from typing import Self
@dataclass
class FileReader:
    _fd: FileDescriptor
    @classmethod
    def open(cls, path: str) -> Self:
        return cls(_open(path))
    def read(self, length: int) -> bytes:
        return _read(self._fd, length)
    def close(self) -> None:
        _close(self._fd)

Note that the main technique here is not necessarily using NewType specifically, but rather aligning an instance's property of "has all attributes set" as closely as possible with an instance's property of "fully valid instance of its class"; NewType is just a handy tool to enforce any necessary constraints on the places where you need to use a primitive type like int, str or bytes.

In Summary - The New Best Practice

From now on, when you're defining a new Python class:

• Make it a dataclass².
• Use its default __init__ method³.
• Add @classmethods to provide your users convenient and discoverable ways to build your objects.
• Require that all dependencies be satisfied by attributes, so you always start with a valid object.
• Use typing.NewType to enforce any constraints on primitive data types (like int and str) which might have magical external attributes, like needing to come from a particular library, needing to be random, and so on.

If you define all your classes this way, you will get all the benefits of a custom __init__ method:

• All consumers of your data structures will receive valid objects, because an object with all its attributes populated correctly is inherently valid.
• Users of your library will be presented with convenient ways to create your objects that do as much work as is necessary to make them easy to use, and they can discover these just by looking at the methods on your class itself.

Along with some nice new benefits:

• You will be future-proofed against new requirements for different ways that users may need to construct your object.
• If there are already multiple ways to instantiate your class, you can now give each of them a meaningful name; no need to have monstrosities like def __init__(self, maybe_a_filename: int | str | None = None):
• Your test suite can always construct an object by satisfying all its dependencies; no need to monkey-patch anything when you can always call the type and never do any I/O or generate any side effects.

Before dataclasses, it was always a bit weird that such a basic feature of the Python language - giving data to a data structure to make it valid - required overriding a method with 4 underscores in its name. __init__ stuck out like a sore thumb. Other such methods like __add__ or even __repr__ were inherently customizing esoteric attributes of classes.

For many years now, that historical language wart has been resolved. @dataclass, @classmethod, and NewType give you everything you need to build classes which are convenient, idiomatic, flexible, testable, and robust.

Acknowledgments

Thank you to my patrons who are supporting my writing on this blog. If you like what you've read here and you'd like to read more of it, or you'd like to support my various open-source endeavors, you can support my work as a sponsor! I am also available for consulting work if you think your organization could benefit from expertise on topics like "but what is a 'class', really?".

17 Apr 2025 10:35pm GMT

Below is an example event modeling diagram showing part of the lifecycle of a shopping cart application. The blue boxes are commands. They represent an intent to change the state of the system. This is where domain rules are enforced, and a command either generates an event or an exception.

By creating an event, this changes the state of the system. Other slices listen to events and populate views (the green boxes). These can be caches, database tables, CSV files, or anything that formats data in a way that will be consumed.

Having done a couple event modeling diagrams, what I like about them are:

• Being able to understand the domain more as the diagram comes together
• Uncovering needs early in an project, instead of late in the process when you're implementing the feature under time pressure.
• Making it easier for everyone-developers and stakeholders-to stay aligned

Read more...

17 Apr 2025 4:38pm GMT

Calling all community organizers! We want to sit down together and share what's going well, what new tricks we've learned, and what's been challenging in the area of organizing Python and Python adjacent communities. So if you're attending PyCon US this year and you run a local Python meet-up, help organize a regional conference, or facilitate peer learning through workshops or hack nights, we hope you will join us at the Community Organizer Summit and the PSF & Meetups Discussion.

Community Organizer Summit: 10:30 am - 1 pm on Saturday, May 17th
PSF & Meetups Discussion: 1pm - 2pm on Sunday, May 18th

Saturday's Community Organizer Summit kicks off with a keynote and then three short "Show & Tell" slots, for which the CFP is open now. After the more formal sessions, we'll break into unconference discussions, so be sure to bring your ideas. If you're not able to make it to Pittsburgh but you want to join the ongoing conversation about conference organizing, please consider joining the bi-monthly Conference Chats on Discord. Thanks so much to Mason Egger, Heather White and Kevin Horn for organizing the Summit!

Sunday's PSF & Meetups Discussion is organized by the PSF board and is intended to kick off a more collaborative and purposeful era of Python community organizer knowledge sharing! Community organizer meetups have been happening casually at PyCon US and (more famously) at EuroPython for a few years now. We're looking to make this year's PyCon US session into a jumping off point for a global conversation that we hope to see happen regularly in lots of places around the world, as well as online. If you've got ideas or want to be involved in this conversation, please email community-organizing@pyfound.org

This is not just for experienced organizers! Looking to start a community and want to learn from a group of awesome people? Come join us!

17 Apr 2025 10:13am GMT

This is the blog version of a talk! If you prefer, watch the recording on YouTube:

Sage Abdullah - Run your tests against Django's main! - Django London Meetup

Django is known for its stability. The framework makes a strong commitment to API stability and forwards-compatibility, ensuring that developers can rely on it for building long-term, maintainable projects. A key aspect of this commitment involves extensive testing and structured releases-an area where testing by Django users can significantly enhance Django's reliability. Here's a closer look at how this works, and how you can contribute 🤝.

How Django stays stable

Django's stability is upheld through rigorous testing. As of Django 5.2, there are more than 18,000 tests run against all officially supported database backends, Python versions, and operating systems. Additionally, Django follows a well-structured deprecation policy, ensuring that public APIs are deprecated over at least two feature releases before being removed.

The feature release schedule is systematic and structured:

1. Active development happens on the main branch.
2. A stable branch (for example stable/5.2.x) is forked when an alpha release is made.
3. After a month, the beta release follows, where only release-blocking bug fixes are allowed.
4. A month later, a release candidate (RC) is published, marking the translation string freeze.
5. If no critical bugs are found, the final release is published after a couple of weeks.

With this structured approach, Django ensures that releases are stable. However, bugs can and do occasionally slip through the cracks!

Catching issues early

The best time to catch issues is before they reach the final release. Ideally, potential bugs should be caught at the pull request stage, but keeping up with all changes is challenging. This is where the community can help-by running their tests with Django's main branch.

How you can help

You can set up your test suite to run with Django's main branch in your tests pipeline. Here's an example using GitHub Actions, a popular Continuous Integration platform:

test:
    runs-on: ubuntu-latest
    continue-on-error: ${{ matrix.experimental }}
    strategy:
      matrix:
        include:
          - python: "3.13"
            django: "git+https://github.com/django/django.git@main#egg=Django"
            experimental: true
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python }}
      - run: pip install -r requirements.txt
      - if: ${{ matrix.experimental }}
        run: pip install "${{ matrix.django }}"
      - run: python -Wd manage.py test
  

:root[data-theme="light"] .highlight .l { color: var(--code-fg); }

If you maintain a Django package, you likely already test with multiple Django versions. Adding the main branch ensures that your project stays ahead of potential breaking changes.

Why this helps you

Running tests with Django main allows you to detect when changes in Django break your project. Sometimes, this happens due to the removal of internal APIs (that were never intended for reuse outside Django 🙈). If your tests fail, you can identify which commit caused the issue and adjust your code accordingly.

For example, on the Wagtail CMS project, recently caught an issue when an internal class, SubqueryConstraint, was removed from Django. This wasn't a bug in Django-it was the removal of an internal workaround that was no longer needed. If your project relies on internal APIs, testing against main is crucial to avoid surprises.

Why this helps Django

Testing with main doesn't just help your project-it helps Django too. Sometimes, your tests may fail due to legitimate regressions in Django that its test suite doesn't cover. Reporting these issues ensures they get fixed before the next release.

For example, just two days before Django 5.2 alpha was released, Wagtail tests on main helped detect a bug where calling .full_clean() on a child model in a multi-table inheritance setup triggered an unintended database query. This regression was promptly fixed, ensuring a smoother release for all users.

Take action: test against Django's main and report issues

By running your tests against Django's main branch and reporting any issues you find, you contribute to a more stable framework for everyone. It's a small step that makes a big impact.

So, take a few minutes today to update your automated tests setup and help keep Django as reliable as ever!

17 Apr 2025 9:00am GMT

Hello, Pythonistas! 🐍

Tickets are on sale now! You can get your tickets at europython.eu/tickets

📣 Programme

Keynote Speakers

We are excited to announce three more keynote speakers!

Sebastián Ramírez

Sebastián is the creator and maintainer of several widely used open-source Python libraries, including FastAPI, Typer, SQLModel, and Asyncer. His work focuses on tools for building web APIs, command-line interfaces, and working with data models in Python.

He has collaborated with teams and companies across Latin America, Europe, the Middle East, and the United States, working on systems involving APIs, distributed computing, and machine learning.

Currently, Sebastián works full-time on the development and maintenance of his open-source projects.

Brett Cannon

Brett has been a Python core developer since 2003 and has contributed to just about every corner of the language. He's one of the most prolific PEP authors, the creator of importlib, and played a major role in the Python 2 -> 3 transition and the creation of pyproject.toml.

He served on the inaugural Python Steering Council and remained on it for five years, guiding the language through some of its most critical transitions.

But beyond code, Brett is known for his love of the Python community - "I came for the language, but I stayed for the community". He's a former PSF board member and a recipient of the Frank Willison Award.

Petr Baudiš

Petr Baudiš is a co-founder and CTO of Rossum, a SaaS deep tech scaleup that focuses on automating document-based B2B transactional communication. He co-invented Rossum&aposs unique AI engine and currently oversees Rossum&aposs R&D and product engineering. In the past, Petr led multiple AI research and deep learning projects in academia as well as in commercial environments. His work was used by companies like Google DeepMind, Novartis, Seznam.cz and countless others.

Petr&aposs past exploits include building one of the key predecessors of AlphaGo, working with Linus Torvalds as one of the founding developers of Git, and leading many other open-source projects.

Sessions

Our session list is live! 🎉 https://ep2025.europython.eu/sessions/

🎤 Speaker Mentorship

Accepted Proposals

We're excited to share that 44 proposals were submitted by the mentees and 8 have been accepted! We're incredibly proud of everyone who took the time to submit a proposal-your creativity, passion, and dedication are what make this community so special. Thank you for helping us shape an inspiring and diverse conference program. We can't wait to see you in Prague! 🐍💛

First Time Speakers' Workshop

Whether you&aposre preparing to present at EuroPython 2025 or any other conference or your talk proposal wasn&apost accepted this year, we warmly invite you to our First-Time Speakers&apos Workshop. Join us online via ZOOM on June 4, 2025, at 18:00 CEST to gain essential presentation skills, practical tips, and valuable insights from experienced speakers. This supportive session also aims to inspire you and strengthen your future conference proposals. You can find out more details here: https://ep2025.europython.eu/programme/mentorship/

Register now: https://forms.gle/T8rc73sbyu3KbLNKA

We look forward to supporting your speaking journey!

💰 Sponsorship

If you&aposre passionate about supporting EuroPython and helping us make the event accessible to all, consider becoming a sponsor or asking your employer to join us in this effort.

By sponsoring EuroPython, you're not just backing an event - you&aposre gaining highly targeted visibility and the chance to present your company or personal brand to one of the largest and most diverse Python communities in Europe and beyond!

We offer a range of sponsorship tiers, some with limited slots available. Along with our main packages, there are optional add-ons and optional extras.

👉 More information at: https://ep2025.europython.eu/sponsorship/sponsor/

👉 Contact us at sponsors@europython.eu

EuroPython Society

🏰 Call for Venues - EuroPython 2026

Are you a community builder who dreams of bringing EuroPython to your city? The Call for Venues for EuroPython 2026 is now open!

If you want to propose a location on behalf of your community, please fill in the following form: https://forms.gle/ZGQA7WhTW4gc53MD6

📊 Board Report

The EuroPython conference wouldn't be what it is without the incredible volunteers who make it all happen. 💞 Behind the scenes, there's also the EuroPython Society-a volunteer-led non-profit that manages the fiscal and legal aspects of running the conference, oversees its organization, and works on a few smaller projects like the grants programme. To keep everyone in the loop and promote transparency, the Board is sharing regular updates on what we're working on.

The March board report is ready: https://www.europython-society.org/board-report-for-march/

💞 Community Outreach

Prague Python Meetup Pyvo

A big thank you to Jakub Červinka for giving a lightning talk about EuroPython at the Prague Pyvo meetup!

PyCon US

We're happy to share that EuroPython will have a booth at PyCon US this year! If you're attending, come over to say hi-pick up some stickers, meet members of our community, and learn more about our initiatives and upcoming plans.

PyCon DE & PyData, DjangoCon Europe

We're proud to be supporting both events this year! You'll find EuroPython stickers in the community area-stop by to pick some up and connect with fellow community members while you&aposre there.

Brno Python Pizza

We recently supported Brno Python Pizza, which took place this February in Brno. The organizers shared a lovely write-up about the event, which you can read here: https://www.europython-society.org/brno-python-pizza-great-things-come-in-threes/

💞Upcoming Events in the Python Community

• PyCon Lithuania, Vilnius, April 23-25 https://pycon.lt/
• DjangoCon Europe, Dublin, April 23-27 https://2025.djangocon.eu/
• PyCon DE & PyData, Darmstadt, April 23-25 https://2025.pycon.de
• PyCamp España 2025, Sevilla, 01 - 04 May https://pycamp.es/
• Pycon Italia, Bologna, May 28-31 https://2025.pycon.it/en
• EuroPython, Prague, 14 July-20 July https://ep2025.europython.eu
• EuroSciPy, Kraków, August 18-22 https://euroscipy.org/2025/
• PyCon Greece 2025, Athens, Greece, 29-30 August https://2025.pycon.gr/en/
• PyCamp CZ 25 beta, Třeštice, September 12-14 https://pycamp.cz/
• Pycon UK, Manchester, September 19-22 https://2025.pyconuk.org/
• PyCon Estonia, Tallinn, October 2-3 https://pycon.ee/
• PyCon Sweden, Stockholm, October 30-31 https://pycon.se/

🐣 See You All Next Month

And in the meantime, follow us on social media:

• LinkedIn: https://www.linkedin.com/company/europython/
• X: https://x.com/europython
• Mastodon: https://fosstodon.org/@europython
• BlueSky: https://bsky.app/profile/europython.eu
• YouTube: https://www.youtube.com/@EuroPythonConference

17 Apr 2025 5:00am GMT

In Python, the break statement lets you exit a loop prematurely, transferring control to the code that follows the loop. This tutorial guides you through using break in both for and while loops. You'll also briefly explore the continue keyword, which complements break by skipping the current loop iteration.

By the end of this tutorial, you'll understand that:

• A break in Python is a keyword that lets you exit a loop immediately, stopping further iterations.
• Using break outside of loops doesn't make sense because it's specifically designed to exit loops early.
• The break doesn't exit all loops, only the innermost loop that contains it.

To explore the use of break in Python, you'll determine if a student needs tutoring based on the number of failed test scores. Then, you'll print out a given number of test scores and calculate how many students failed at least one test.

You'll also take a brief detour from this main scenario to examine how you can use break statements to accept and process user input, using a number-guessing game.

Get Your Code: Click here to download the free sample code that shows you how to exit loops early with the Python break keyword.

Take the Quiz: Test your knowledge with our interactive "How to Exit Loops Early With the Python Break Keyword" quiz. You'll receive a score upon completion to help you track your learning progress:

---

Interactive Quiz

How to Exit Loops Early With the Python Break Keyword

In this quiz, you'll test your understanding of the Python break statement. This keyword allows you to exit a loop prematurely, transferring control to the code that follows the loop.

Introducing the break Statement

Before proceeding to the main examples, here's a basic explanation of what the break statement is and what it does. It's a Python keyword that, when used in a loop, immediately exits the loop and transfers control to the code that would normally run after the loop's standard conclusion.

You can see the basics of the break statement in a simple example. The following code demonstrates a loop that prints numbers within a range until the next number is greater than 5:

Python

>>> for number in range(10):
...     if number > 5:
...         break
...     print(number)
... 
0
1
2
3
4
5

Copied!

This short code example consists of a for loop that iterates through a range of numbers from 0 to 9. It prints out each number, but when the next number is greater than 5, a break statement terminates the loop early. So, this code will print the numbers from 0 to 5, and then the loop will end.

As break statements end loops early, it wouldn't make sense for you to use them in any context that doesn't involve a loop. In fact, Python will raise a SyntaxError if you try to use a break statement outside of a loop.

A key benefit of using break statements is that you can prevent unnecessary loop iterations by exiting early when appropriate. You'll see this in action in the next section.

Breaking Out of a Loop With a Set Number of Iterations

Imagine you're a teacher who evaluates the scores of your students. Based on the scores, you want to determine how many tests each student has failed. The following example demonstrates how you might accomplish this task using a for loop to iterate through the students' test scores:

Python

>>> scores = [90, 30, 50, 70, 85, 35]

>>> num_failed_scores = 0
>>> failed_score = 60
>>> for score in scores:
...     if score < failed_score:
...         num_failed_scores += 1
...
>>> print(f"Number of failed tests: {num_failed_scores}")
Number of failed tests: 3

Copied!

Read the full article at https://realpython.com/python-break/ »

[ Improve Your Python With 🐍 Python Tricks 💌 - Get a short & sweet Python Trick delivered to your inbox every couple of days. >> Click here to learn more and see examples ]

16 Apr 2025 2:00pm GMT

PyCharm 2025.1 brings major updates to improve your development experience.

PyCharm is now a unified product, combining PyCharm Professional and Community Edition. Version 2025.1 also brings a free AI tier, the public release of Junie, the launch of Cadence, significant Jupyter enhancements, support for Hatch, Data Wrangler, and many other improvements.

Get the latest version from our download page or update through our free Toolbox App.

Read this blog post to learn more about the updates.

Prefer video? Get an overview of the major news and improvements in this video:

PyCharm is now one powerful, unified product!

PyCharm is now one powerful, unified product! Its core functionality, including Jupyter Notebook support, will be free, and a Pro subscription with additional features will be available.

Starting with the 2025.1 release, every user will get instant access to a free one-month Pro trial, so you'll be able to access all of PyCharm's advanced features right away. After the trial, you can choose whether to continue with a Pro subscription or keep using the core features for free. Learn more about the change in this blog post.

Junie - your personal coding agent Pro

Junie, the coding agent by JetBrains, is now available in PyCharm via JetBrains AI. Junie autonomously plans, writes, refines, and tests code to make your development experience smooth, efficient, and enjoyable. It handles tedious tasks like restructuring code, creating tests, and implementing refinements, so you can focus on bigger challenges and innovation.

PyCharm goes AI

JetBrains AI has received a major upgrade, bringing both AI Assistant and the coding agent Junie under a single subscription. With this release, all JetBrains AI features are accessible for free in PyCharm Pro, with unlimited use for some, such as code completion and local model support, and limited credit-based access to others.

We're also introducing a new subscription system that makes it easy to scale up as needed with the AI Pro and AI Ultimate tiers. Other highlights of this release include smarter completion, advanced context awareness, and support for Claude 3.7 Sonnet and Gemini 2.0 Flash. Head to the What's New page to learn more about the latest AI features.

Cadence - effortless cloud execution for ML workflows Pro

We're introducing Cadence. You can now run your machine learning code on powerful cloud hardware directly from PyCharm in minutes - no complex setup or cloud expertise is required. The Cadence plugin simplifies ML workflows, allowing you to focus on your code while leveraging scalable computing resources.

Data Wrangler Pro

We've implemented Data Wrangler, a powerful tool to help Python data professionals streamline data manipulation and focus on higher-level analysis. Use the interactive UI to perform common dataframe transformations - like filtering, cleaning, handling outliers, and more - without writing repetitive code.

You can view and explore column statistics, generate Python code for transformations automatically, track the history of changes, export data easily, and insert transformations as new cells in your notebook.

SQL cells in notebooks Pro

PyCharm 2025.1 introduces SQL cells. This new cell type allows you to query databases, dataframes, and attached CSV files in Jupyter notebooks and automatically save query results to pandas DataFrames.

We've also introduced many other improvements to enhance the Jupyter notebook experience. Learn more about them in the What's New.

Support for Hatch

We're introducing support for Hatch, a modern and extensible Python project manager from the Python Packaging Authority (PyPA). Hatch can automatically migrate setuptools configurations, create isolated environments, and run and publish builds, making Python package management more efficient.

PyCharm also allows you to create new projects managed by Hatch. The IDE will automatically recognize Hatch projects when they are imported from a local machine or a remote source.

Looking for more?

• Visit our What's New page to learn about other 2025.1 features and the latest bug fixes.
• Read the release notes for the full breakdown of the changes.
• If you encounter any problems, please report them via our issue tracker so we can address them promptly.

We'd love to hear your feedback on PyCharm 2025.1 - leave your comments below or connect with us on X.

16 Apr 2025 1:58pm GMT