Planet Python
Last update: September 09, 2026 04:48 PM UTC
September 09, 2026
Python GUIs
Clean up on exit — Stopping threads when closing a PyQt6 application — How to properly shut down background threads and workers when your application window is closed
I'm using QThreadPool and worker threads in my PyQt application. When I click the X button to close the window, the threads keep running in the background. What's the best way to clean everything up on application exit?
This is a very common situation when working with multithreaded PyQt6 applications. You've set up background workers using QThreadPool or QThread, everything runs great — but when you close the window, the application doesn't fully exit. The threads keep going, and you might even see errors in your console.
The solution is to hook into your window's close event and explicitly stop your background work before the window finishes closing. Let's walk through how to do this.
Understanding the problem
When you close a PyQt6 window by clicking the X button (or calling close()), Qt destroys the window and its widgets. But any threads you've started — whether via QThreadPool, QThread, or QRunnable — are managed separately. They don't automatically stop just because the window is gone.
This means your Python process can hang, or you might see tracebacks as threads try to interact with widgets that no longer exist.
Overriding closeEvent
Every QWidget (including QMainWindow) has a method called closeEvent that Qt calls whenever the widget is about to close. By overriding this method, you can run your own cleanup code at exactly the right moment.
Here's a minimal example:
from PyQt6.QtWidgets import QMainWindow
class MainWindow(QMainWindow):
def closeEvent(self, event):
# Put your cleanup code here
print("Window is closing — cleaning up!")
event.accept()
The event parameter is a QCloseEvent. Calling event.accept() tells Qt to go ahead and close the window. If you wanted to cancel the close (for example, to show a "Save changes?" dialog), you would call event.ignore() instead.
Stopping workers on close
If you're managing background workers — for example, through a QThreadPool — you'll want to signal them to stop and then wait for them to finish before allowing the window to close.
Here's how that looks in practice. First, let's set up a simple worker using QRunnable:
import time
from PyQt6.QtCore import QRunnable, pyqtSlot, QObject, pyqtSignal
class WorkerSignals(QObject):
finished = pyqtSignal()
class Worker(QRunnable):
def __init__(self):
super().__init__()
self.signals = WorkerSignals()
self.is_running = True
@pyqtSlot()
def run(self):
while self.is_running:
print("Worker is working...")
time.sleep(1)
print("Worker stopped.")
self.signals.finished.emit()
def stop(self):
self.is_running = False
The worker runs in a loop, checking self.is_running on each iteration. When stop() is called, it sets the flag to False, and the loop exits on the next check.
To stop QRunnable objects you need to use this flag-watching approach.
If your runnable doesn't have a loop, and is instead doing a long series of processing steps, you instead will need check the flag state multiple times during that code and return or raise to exit the runner.
The exit can only happen at the points where the flag is checked.
To avoid multiple if checks in the code, an alternative is to have a check method that raises an exception for the stop state. For example:
def maybe_stop(self):
if not self.is_running:
raise Exception("Worker stopped.")
You can then use this as follows:
import time
from PyQt6.QtCore import QRunnable, pyqtSlot, QObject, pyqtSignal
class WorkerSignals(QObject):
finished = pyqtSignal()
class Worker(QRunnable):
def __init__(self):
super().__init__()
self.signals = WorkerSignals()
self.is_running = True
@pyqtSlot()
def run(self):
print("Worker is working...")
self.maybe_stop()
time.sleep(5) # <- do some work.
self.maybe_stop()
time.sleep(5) # <- do some more work.
self.maybe_stop()
time.sleep(5) # <- do some more work.
self.maybe_stop()
time.sleep(5) # <- do some more work.
self.maybe_stop()
time.sleep(5) # <- do some more work.
# <- no point stopping now.
print("Worker stopped.")
self.signals.finished.emit()
def stop(self):
self.is_running = False
def maybe_stop(self):
if not self.is_running:
raise Exception("Worker stopped.")
We'll not use this approach in our example here, since for testing purposes it is better to have a worker that doesn't stop. But you may find it useful in your own code.
Now let's put together a QMainWindow that starts a worker and cleans it up on close:
import sys
import time
from PyQt6.QtCore import QRunnable, QThreadPool, pyqtSlot, QObject, pyqtSignal
from PyQt6.QtWidgets import QApplication, QMainWindow, QLabel
class WorkerSignals(QObject):
finished = pyqtSignal()
class Worker(QRunnable):
def __init__(self):
super().__init__()
self.signals = WorkerSignals()
self.is_running = True
@pyqtSlot()
def run(self):
while self.is_running:
print("Worker is working...")
time.sleep(1)
print("Worker stopped.")
self.signals.finished.emit()
def stop(self):
self.is_running = False
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("Thread Cleanup Example")
label = QLabel("Close this window to stop the worker.")
self.setCentralWidget(label)
self.threadpool = QThreadPool()
self.workers = []
# Start a background worker.
worker = Worker()
self.workers.append(worker)
self.threadpool.start(worker)
def closeEvent(self, event):
# Signal all workers to stop.
for worker in self.workers:
worker.stop()
# Wait for all threads in the pool to finish.
self.threadpool.waitForDone()
print("All workers stopped. Closing application.")
event.accept()
app = QApplication(sys.argv)
window = MainWindow()
window.show()
sys.exit(app.exec())
When you run this and close the window, you'll see the worker print its "stopped" message, and the application exits cleanly.
Let's look at what's happening in closeEvent:
First, we loop through all our tracked workers and call stop() on each one, setting the flag that tells them to finish. Then we call self.threadpool.waitForDone(), which blocks until every runnable in the pool has completed. This ensures we don't pull the rug out from under a running thread. Finally, we call event.accept() to let the window close.
Managing multiple worker groups
If your application has different categories of workers — say, one group handling camera feeds and another running inference engines — you can keep separate lists and stop them all in closeEvent:
def closeEvent(self, event):
for worker in self.feed_workers:
worker.stop()
for worker in self.engine_workers:
worker.stop()
self.threadpool.waitForDone()
event.accept()
The same principle applies: signal every worker to stop, then wait for the thread pool to drain.
Ensuring a clean exit with sys.exit
You might notice that even after the window closes, the Python process occasionally doesn't exit cleanly. This usually happens when sys.exit() isn't receiving the application's return code properly.
The standard way to launch and exit a PyQt6 application is:
app = QApplication(sys.argv)
window = MainWindow()
window.show()
sys.exit(app.exec())
app.exec() starts the Qt event loop and returns an exit code (an integer) when the loop ends. Passing that code to sys.exit() ensures Python terminates with the correct status.
If you skip sys.exit() and just call app.exec(), Python may not tear down all its resources properly — particularly if threads or other objects are still being referenced. Wrapping it in sys.exit() triggers a proper SystemExit exception, which gives Python the chance to clean everything up.
Complete working example
Here's the full working example:
import sys
import time
from PyQt6.QtCore import QRunnable, QThreadPool, pyqtSlot, QObject, pyqtSignal
from PyQt6.QtWidgets import QApplication, QMainWindow, QLabel
class WorkerSignals(QObject):
finished = pyqtSignal()
class Worker(QRunnable):
def __init__(self, worker_id):
super().__init__()
self.worker_id = worker_id
self.signals = WorkerSignals()
self.is_running = True
@pyqtSlot()
def run(self):
while self.is_running:
print(f"Worker {self.worker_id} is working...")
time.sleep(1)
print(f"Worker {self.worker_id} stopped.")
self.signals.finished.emit()
def stop(self):
self.is_running = False
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("Thread Cleanup on Exit")
self.resize(400, 200)
label = QLabel("Close this window to stop all workers.")
label.setMargin(20)
self.setCentralWidget(label)
self.threadpool = QThreadPool()
self.workers = []
# Start a few background workers.
for i in range(3):
worker = Worker(worker_id=i)
self.workers.append(worker)
self.threadpool.start(worker)
def closeEvent(self, event):
print("Close event received. Stopping workers...")
for worker in self.workers:
worker.stop()
self.threadpool.waitForDone()
print("All workers stopped. Goodbye!")
event.accept()
app = QApplication(sys.argv)
window = MainWindow()
window.show()
sys.exit(app.exec())
When you run this, you'll see three workers printing messages to the console. Close the window, and you'll see each worker confirm it has stopped before the application exits.
Summary
- Override
closeEventon your main window to run cleanup code when the user closes the application. - Set a flag on each worker to signal it should stop, and check that flag in the worker's loop.
- Use
QThreadPool.waitForDone()to block until all workers have finished before allowing the window to close. - Always wrap
app.exec()withsys.exit()to ensure a clean process exit.
This pattern works well for any PyQt6 application with background threads, whether you're processing data, handling network requests, or running live camera feeds. Once you have it in place, your application will shut down gracefully every time. For a complete introduction to using QThreadPool and QRunnable for multithreading in PyQt6, see our Multithreading PyQt6 applications with QThreadPool tutorial. You may also find our guides on signals and slots and creating your first PyQt6 window helpful as you build out your application.
For an in-depth guide to building Python GUIs with PyQt6 see my book, Create GUI Applications with Python & Qt6.
September 09, 2026 06:00 AM UTC
Fixing Crashes When Using NumPy Arrays with QImage in Qt Threads — How to safely pass image data between threads when streaming video or updating displays
I'm using a threaded runner to stream a live video feed by converting a NumPy array to a
QImage, then to aQPixmap, and displaying it on aQLabel. But I'm frequently encountering crashes when the label is resized too quickly or the scroll area is scrolled. Could this be a problem with theQImagememory buffer getting cleared before theQPixmapcan update? Is this fixable, or is it a fundamental issue with threads in Python/Qt?
This is a common problem when working with NumPy arrays and QImage across threads. The good news is that it's fixable. The crashes come from how QImage handles the underlying memory of a NumPy array.
Why the crash happens
When you create a QImage from a NumPy array, QImage doesn't copy the data. Instead it holds a reference to the original memory buffer provided by the NumPy array.
import numpy as np
from PyQt6.QtGui import QImage
# Create a NumPy array (e.g. a video frame)
array = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
# QImage points to the same memory — no copy is made
image = QImage(
array.data,
array.shape[1],
array.shape[0],
array.strides[0],
QImage.Format.Format_RGB8888,
)
This is efficient, but it creates a dangerous situation in a multithreaded application. If the NumPy array is modified or goes out of scope in the worker thread while the GUI thread is still using the QImage to paint, the memory that QImage is pointing to may no longer be valid. The result is a segfault or (worse) a silent crash without any information about what has happened.
This is especially likely when frames are arriving quickly (as with a live video feed) and the GUI is being redrawn frequently — for example, during a resize or a scroll.
The fix: copy the image data
The simplest and most reliable fix is to make sure the QImage owns its own copy of the pixel data before you pass it to the GUI thread. You can do this by calling .copy() on the QImage:
image = QImage(
array.data,
array.shape[1],
array.shape[0],
array.strides[0],
QImage.Format.Format_RGB888,
).copy()
The .copy() call creates a new QImage with its own independent memory buffer. Now it doesn't matter if the original NumPy array changes or disappears — the QImage is safe to use from the GUI thread.
If you've already tried using .copy() and it hasn't worked, bear in mind that where you use the copy matters just as much as using it at all.
Where to copy matters
If you create the QImage in the worker thread and emit it via a signal, the copy needs to happen before the signal is emitted. If the signal carries a reference to the original (non-copied) QImage, the data might still be invalidated before the GUI thread processes it.
Here's the pattern that works:
# In the worker thread
image = QImage(
frame.data,
frame.shape[1],
frame.shape[0],
frame.strides[0],
QImage.Format.Format_RGB888,
).copy() # Copy immediately, before emitting
self.signals.result.emit(image) # Now safe to send to GUI thread
And in the main thread, connect that signal to a slot that updates the display:
def update_display(self, image):
pixmap = QPixmap.fromImage(image)
self.label.setPixmap(pixmap)
Because the QImage was copied before it crossed the thread boundary, the GUI thread has full ownership of the data and can paint it safely.
Use signals to control the update flow
Another source of crashes is calling GUI methods directly from a worker thread. In Qt, all GUI updates must happen on the main thread. If you're calling label.setPixmap(...) from inside a worker or a callback running on a background thread, that's undefined behavior and will eventually crash.
The solution is to always use signals and slots to communicate between threads. Emit a signal from the worker carrying the processed image, and connect it to a slot on the main thread that performs the GUI update.
This also gives you a natural way to throttle updates. If frames are arriving faster than the GUI can paint them, you can use a flag to skip frames that arrive while the previous one is still being displayed.
Complete working example
Here's a full example that simulates a video feed using a QRunnable and a QThreadPool. It generates random NumPy frames in a background thread and safely displays them on a QLabel. If you're new to running background tasks with QThreadPool, see our detailed guide to multithreading PyQt6 applications.
import sys
import time
import numpy as np
from PyQt6.QtCore import (
QObject,
QRunnable,
QThreadPool,
pyqtSignal,
pyqtSlot,
)
from PyQt6.QtGui import QImage, QPixmap
from PyQt6.QtWidgets import (
QApplication,
QLabel,
QMainWindow,
QScrollArea,
QVBoxLayout,
QWidget,
)
class WorkerSignals(QObject):
frame_ready = pyqtSignal(QImage)
finished = pyqtSignal()
class VideoWorker(QRunnable):
def __init__(self):
super().__init__()
self.signals = WorkerSignals()
self.running = True
@pyqtSlot()
def run(self):
while self.running:
# Simulate a video frame (e.g. from a camera or stream)
frame = np.random.randint(
0, 255, (480, 640, 3), dtype=np.uint8
)
# Create QImage and copy it immediately so it owns its data
image = QImage(
frame.data,
frame.shape[1],
frame.shape[0],
frame.strides[0],
QImage.Format.Format_RGB888,
).copy()
# Emit the safe, copied image to the main thread
self.signals.frame_ready.emit(image)
# Simulate ~30 fps
time.sleep(1 / 30)
self.signals.finished.emit()
def stop(self):
self.running = False
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("Threaded Video Display")
self.label = QLabel("Waiting for frames...")
self.label.setScaledContents(True)
scroll_area = QScrollArea()
scroll_area.setWidget(self.label)
scroll_area.setWidgetResiizable(True)
container = QWidget()
layout = QVBoxLayout(container)
layout.addWidget(scroll_area)
self.setCentralWidget(container)
self.resize(700, 520)
# Set up threading
self.threadpool = QThreadPool()
self.worker = VideoWorker()
self.worker.signals.frame_ready.connect(self.update_display)
self.worker.signals.finished.connect(self.on_finished)
self.threadpool.start(self.worker)
def update_display(self, image):
"""Runs on the main thread — safe to update the GUI here."""
pixmap = QPixmap.fromImage(image)
self.label.setPixmap(pixmap)
def on_finished(self):
print("Worker finished.")
def closeEvent(self, event):
self.worker.stop()
self.threadpool.waitForDone(2000)
event.accept()
app = QApplication(sys.argv)
window = MainWindow()
window.show()
app.exec()
When you run this, you'll see a window displaying rapidly changing random noise — a stand-in for a real video stream. You can resize the window and scroll around without crashes, because the image data is safely copied before it crosses the thread boundary, and all GUI updates happen on the main thread via signals.
Recap
When working with NumPy arrays and QImage across threads, keep these three things in mind:
-
QImagedoes not copy NumPy data. It points to the original array's memory buffer. If that buffer changes or is freed, theQImagebecomes invalid. -
Call
.copy()on theQImagebefore emitting it across threads. This gives theQImageits own memory, independent of the NumPy array. -
Always update the GUI from the main thread. Use signals to send data from background workers to slots connected on the main thread, where it's safe to call
setPixmap()and other GUI methods.
With these practices in place, you can stream video or display rapidly changing image data without encountering the mysterious crashes that come from shared memory across threads.
For an in-depth guide to building Python GUIs with PyQt6 see my book, Create GUI Applications with Python & Qt6.
September 09, 2026 06:00 AM UTC
How to Check if a QLineEdit is Empty in Python — Empty strings are falsey in Python
A reader asked:
I just want to know, how do I check whether a QLineEdit is empty or not?
The QLineEdit class doesn't have an isEmpty() method which you can call to find out if the line edit is empty, but we don't need one! Instead we can get the current text using .text() and then check if the returned value is an empty string.
Checking QLineEdit Text with .text()
In the code below lineedit is our already created QLineEdit widget.
text = lineedit.text()
if text == '': # if the line edit is empty, .text() will return an empty string.
# do something
Using Python's Falsey Empty Strings
We can simplify this further. In Python empty strings are falsey -- they are considered False values in conditional expressions. So instead of checking the string is empty, we can check if it is true (non-empty) or false (empty).
if lineedit.text():
# do something if there is content in the line edit.
Or, to check if it is empty:
if not lineedit.text():
# do something if the line edit is empty.
Complete Example: Detecting Empty QLineEdit with Signals
Below is a small demo application which updates a label to indicate if the QLineEdit has text in it or not. In this we use Qt signals to send the current text to a slot method every time it is updated.
- PyQt5
- PyQt6
- PySide2
- PySide6
import sys
from PyQt5.QtWidgets import QApplication, QLabel, QLineEdit, QVBoxLayout, QWidget
class Window(QWidget):
def __init__(self):
super().__init__()
self.lineedit = QLineEdit()
self.lineedit.textChanged.connect(self.text_changed)
self.label = QLabel()
vlayout = QVBoxLayout()
vlayout.addWidget(self.lineedit)
vlayout.addWidget(self.label)
self.setLayout(vlayout)
def text_changed(self, s):
# s contains the text of the line edit, we could also test self.lineedit.text()
if s:
self.label.setText("Not empty")
else:
self.label.setText("Empty")
app = QApplication(sys.argv)
w = Window()
w.show()
app.exec_()
import sys
from PyQt6.QtWidgets import QApplication, QLabel, QLineEdit, QVBoxLayout, QWidget
class Window(QWidget):
def __init__(self):
super().__init__()
self.lineedit = QLineEdit()
self.lineedit.textChanged.connect(self.text_changed)
self.label = QLabel()
vlayout = QVBoxLayout()
vlayout.addWidget(self.lineedit)
vlayout.addWidget(self.label)
self.setLayout(vlayout)
def text_changed(self, s):
# s contains the text of the line edit
if s:
self.label.setText("Not empty")
else:
self.label.setText("Empty")
app = QApplication(sys.argv)
w = Window()
w.show()
app.exec()
import sys
from PySide2.QtWidgets import QApplication, QLabel, QLineEdit, QVBoxLayout, QWidget
class Window(QWidget):
def __init__(self):
super().__init__()
self.lineedit = QLineEdit()
self.lineedit.textChanged.connect(self.text_changed)
self.label = QLabel()
vlayout = QVBoxLayout()
vlayout.addWidget(self.lineedit)
vlayout.addWidget(self.label)
self.setLayout(vlayout)
def text_changed(self, s):
# s contains the text of the line edit
if s:
self.label.setText("Not empty")
else:
self.label.setText("Empty")
app = QApplication(sys.argv)
w = Window()
w.show()
app.exec_()
import sys
from PySide6.QtWidgets import QApplication, QLabel, QLineEdit, QVBoxLayout, QWidget
class Window(QWidget):
def __init__(self):
super().__init__()
self.lineedit = QLineEdit()
self.lineedit.textChanged.connect(self.text_changed)
self.label = QLabel()
vlayout = QVBoxLayout()
vlayout.addWidget(self.lineedit)
vlayout.addWidget(self.label)
self.setLayout(vlayout)
def text_changed(self, s):
# s contains the text of the line edit
if s:
self.label.setText("Not empty")
else:
self.label.setText("Empty")
app = QApplication(sys.argv)
w = Window()
w.show()
app.exec_()
Run the above and you'll see the label update as you add and remove text in the QLineEdit.


This approach works across all Python Qt bindings including PyQt5, PyQt6, PySide2 and PySide6. By leveraging Python's truthiness checks on strings, you can validate QLineEdit input cleanly without needing a dedicated isEmpty() method. For more advanced input validation techniques, you may also want to look at input validation in Tkinter or explore the full range of PyQt6 widgets available for building your applications.
For an in-depth guide to building Python GUIs with PySide6 see my book, Create GUI Applications with Python & Qt6.
September 09, 2026 06:00 AM UTC
Graham Dumpleton
Tracing Flask with wrapture
For a web application the natural unit of tracing is the request: one HTTP request, its method, path and status, and every observed call made while handling it, as one tree. The config file from last time cannot give you that on its own, and it is worth being clear about why before showing what does.
A WSGI application looks like any other callable, but it routes the interesting facts around the return value. The status and headers travel through the start_response callback rather than being returned. The body is an iterable that the server consumes after the call has returned, so a streaming application does most of its work after a call event would already have closed. And when a view raises, the framework catches the exception and turns it into a 500 response before any wrapper on the application ever sees it. A binding on the application callable would record a call that returned an iterable and raised nothing, which is true and useless.
The shop behind Flask
Here is the shop from the earlier posts behind a small Flask application. A /quote/<item> route renders a template, a /order route places an order through the OrderService from before, and a /health route exists because every deployed service has one.
from flask import Flask, jsonify, render_template, request
from shop import CardDeclined, OrderService
CATALOG = {"widget": 25, "gadget": 120}
app = Flask("webshop")
service = OrderService()
@app.get("/health")
def health():
return "ok\n"
@app.get("/quote/<item>")
def quote(item):
price = CATALOG[item]
return render_template("quote.html", item=item, price=price)
@app.post("/order")
def order():
data = request.get_json()
try:
charge = service.place(data["amount"], data["card"], tenant=data["tenant"])
except CardDeclined as exc:
return jsonify(error=str(exc)), 402
return jsonify(charge)
Nothing in it mentions wrapture. The quote view will raise a KeyError for an item that is not in the catalog, which Flask will turn into a 500, and that is the request I most want to see.
One entry
The Flask knowledge lives in an instrumentation package rather than in the config. With wrapture-instrumentation installed alongside wrapture, the config gains a single [[instrument]] entry naming Flask, and keeps the observe entries for the shop's own methods from last time:
[[instrument]]
name = "flask"
[[observe]]
target = "shop:OrderService"
name = "place"
redact = ["card"]
[[observe]]
target = "shop:Gateway"
name = "charge"
redact = ["card"]
[[observe]]
target = "shop:Ledger"
name = "record"
[[sink]]
type = "printer"
The development server runs under the runner exactly as the script did, with everything after -m flask belonging to Flask:
$ python -m wrapture -m flask --app webshop run --port 5001
Then from another shell, a quote, an order, a declined order and the item that does not exist:
$ curl http://127.0.0.1:5001/quote/widget
$ curl -X POST -H 'Content-Type: application/json' \
-d '{"amount": 500, "card": "4111-1111-1111-1111", "tenant": "acme"}' \
http://127.0.0.1:5001/order
$ curl -X POST -H 'Content-Type: application/json' \
-d '{"amount": 250, "card": "4000-0000-0000-0000", "tenant": "globex"}' \
http://127.0.0.1:5001/order
$ curl http://127.0.0.1:5001/quote/missing
In the server's log, interleaved with Flask's own access log lines which I have removed here, each request arrives as one tree. The quote:
GET /quote/widget (webshop.wsgi_app)
quote(item='widget')
flask:render_template(template_name_or_list='quote.html', context='<context>')
flask:render_template -> '<17 chars>' [1.5ms]
quote -> '<p>widget: 25</p>' [1.6ms]
webshop.wsgi_app -> '200 OK' [2.3ms, body 5us over 1 chunk]
The request line opens the tree, the view sits beneath it labelled by its endpoint, the template render sits beneath the view with the template's name and its context masked (it is arbitrary application data, and the render is captured only as its size), and the closing line carries the status as the request's result along with the time to the last byte of the body. The order, with the shop's own methods nesting beneath the view because their bindings fire while the request is in flight:
POST /order (webshop.wsgi_app)
order()
shop:OrderService.place(amount=500, card='<redacted>', tenant='acme')
shop:Gateway.charge(amount=500, card='<redacted>')
shop:Gateway.charge -> {'id': 'ch_500', 'amount': 500} [7us]
shop:Ledger.record(entry={'id': 'ch_500', 'amount': 500})
shop:Ledger.record -> 'led_ch_500' [4us]
shop:OrderService.place -> {'id': 'ch_500', 'amount': 500} [205us]
order -> <Response 29 bytes [200 OK]> [471us]
webshop.wsgi_app -> '200 OK' [922us, body 4us over 1 chunk]
The declined card, where the view caught the exception and answered 402, so the failure is on the gateway and the service but not on the request:
POST /order (webshop.wsgi_app)
order()
shop:OrderService.place(amount=250, card='<redacted>', tenant='globex')
shop:Gateway.charge(amount=250, card='<redacted>')
shop:Gateway.charge !! CardDeclined [6us]
shop:OrderService.place !! CardDeclined [74us]
order -> (<Response 38 bytes [200 OK]>, 402) [265us]
webshop.wsgi_app -> '402 PAYMENT REQUIRED' [673us, body 4us over 1 chunk]
And the one I wanted:
GET /quote/missing (webshop.wsgi_app)
quote(item='missing')
quote !! KeyError [4us]
webshop.wsgi_app -> '500 INTERNAL SERVER ERROR' !! KeyError [3.2ms, body 6us over 1 chunk]
The request line says two things at once. It answered 500, and the KeyError was the reason. That second half is the part a reader would not guess, because as far as the WSGI middleware recording the request is concerned, the application returned normally. Flask caught the exception on its way out of the view and handed it to handle_exception, which built the 500 response and returned it, so the request completed with a status and no exception. The only place the failure can be seen is inside that handler, where the exception arrives as an argument, and that is where the instrumentation looks. A binding on handle_exception notes the exception against the nearest enclosing request event, using the same note_exception() that the testing series used for a failure the code handled itself, aimed past the handler's own call with current_event(kind="request"). The view's event carries the KeyError as the exception that escaped it, the request's event carries it as a note, and both show up on their lines because two scopes failed for the same reason.
Keeping the noise out
Health checks and static assets make up most of the traffic on a lot of services and none of the interest. In the log above every /health probe printed its own tree, which after a day of a load balancer polling it is most of the file. The instrumentation takes a list of paths not to record:
[[instrument]]
name = "flask"
ignore_paths = ["/health"]
A matching request runs and answers as normal but records nothing at all, and the "at all" matters. Declining the request event alone would leave the view, any lifecycle callbacks and any template render it made on the trace as anonymous roots with no request above them, the same problem tree=True solved on a plain binding in the first post. The setting silences everything beneath an ignored request for its whole extent, so with it in place the health probe leaves only Flask's access log line behind and the next quote prints as before:
127.0.0.1 - - [01/Sep/2026 14:51:43] "GET /health HTTP/1.1" 200 -
GET /quote/gadget (webshop.wsgi_app)
quote(item='gadget')
flask:render_template(template_name_or_list='quote.html', context='<context>')
flask:render_template -> '<18 chars>' [1.5ms]
quote -> '<p>gadget: 120</p>' [1.7ms]
webshop.wsgi_app -> '200 OK' [2.3ms, body 5us over 1 chunk]
The other switch worth knowing about is lifecycle = false. Flask extensions register before_request and after_request callbacks liberally, for loading users, cleaning up sessions and stamping headers, and the instrumentation observes every one of them by default in the order Flask runs them. For an application with several extensions that is faithful but noisy, and switching it off leaves the callbacks running unobserved.
What the instrumentation is
There is no magic in the [[instrument]] entry. It names an Instrumentation class whose hooks run when Flask is imported, and those hooks apply bindings to three choke points in Flask, using the same bindings as everywhere else. Constructing a Flask instance installs the recording WSGI middleware on its wsgi_app attribute, so every application the process creates is covered however it was made, application factories included. Registering a route substitutes an observed version of the view function, since Flask captures views into its dispatch table the moment @app.route runs, before any binding on the module could have seen them. And handle_exception gets the binding that notes the failure described above. The flask-app example in the wrapture repository is that class written out in full, as a local file next to a config, for anyone who wants to do the same for a framework that has no package yet. The packaged version adds the lifecycle callbacks, error handlers, blueprints and template rendering on top.
The request as an event
Everything the tree shows is on the request event itself, which is what a sink or a test reads. The result is the status line, so every existing filter and assertion that works on a return value works on a request. The duration is wall time from the call to the close of the body, time to last byte, with the synchronous phase and the body's own share recorded separately. The HTTP details, method, path, query string with sensitive parameters already masked, scheme, remote address and the bytes actually served, sit in the event's data, and the instrumentation adds the matched route pattern and endpoint once routing has run, which are the low-cardinality keys a backend groups by. The WSGI request tracing page has the full event, the mode="wsgi" binding form for applications with no framework package, and the redaction rules; ASGI applications get the same treatment on the page beside it.
With a request as one tree and timings on every line, the next question is the one every web application eventually asks, which is where the time is going.
September 09, 2026 01:39 AM UTC
Bob Belderbos
Database-Driven RBAC with FastAPI and Azure Entra ID
This came out of a coaching project: a FastAPI service backing a geotechnical database. We use Azure Entra ID for authentication, but we needed role-based access control (RBAC) granularly per endpoint and across different user roles. It turned into an interesting sprint where we designed it so that role changes did not require code changes.
I've seen FastAPI auth examples hardcode a policy like: if "Admin" not in user.roles: raise 403, copy-pasted into every route. It works until the day someone asks "can you let Finance approve expenses too?" and now you are editing code and redeploying to answer a question that is really about data.
I built a small demo that does the opposite: Azure Entra ID handles authentication and which roles a user has, but which roles each endpoint demands lives in a database table. An admin re-points access at runtime through an endpoint that is itself protected by the same mechanism.
Full code, setup steps, and tests: bbelderbos/azure_fastapi_rbac_demo.
The one decision that matters
The roles a user carries come from Entra, baked into the JWT roles claim. That part is fixed. The interesting choice is where you store the mapping from endpoint to required roles.
Put it in code and every policy change is a deploy. Put it in a table and it becomes editable state. It is the 12-factor instinct applied to authorization: anything that varies on a schedule you do not control belongs outside the code, as data:
from sqlalchemy import Column, JSON
from sqlmodel import Field, SQLModel
class EndpointPermission(SQLModel, table=True):
endpoint_key: str = Field(primary_key=True)
roles: list[str] = Field(default_factory=list, sa_column=Column(JSON))
One row per protected endpoint. The demo seeds three on startup, which is the whole policy at a glance:
SEED = {
"expenses:read": ["Viewer", "Approver", "Admin"],
"expenses:approve": ["Approver", "Admin"],
"admin:permissions": ["Admin"],
}
So require("expenses:read") lets any of the three roles through, while expenses:approve drops Viewer. The only thing hardcoded in a route is its key string. The guard reads the rest from the table:
def require(endpoint_key: str) -> Callable[..., Awaitable[User]]:
async def checker(
user: User = Depends(azure_scheme),
session: Session = Depends(get_session),
) -> User:
perm = session.get(EndpointPermission, endpoint_key)
allowed = set(perm.roles) if perm is not None else set()
if not allowed.intersection(set(user.roles)):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have permission to access this endpoint.",
)
return user
return checker
azure_scheme is fastapi-azure-auth's SingleTenantAzureAuthorizationCodeBearer. It validates the token and hands back a User with .roles already populated. The guard does one set intersection. That is the entire authorization model.
Routes read clean, the key string is the only literal:
@app.get("/expenses")
async def list_expenses(user: Annotated[User, Depends(require("expenses:read"))]):
return {"expenses": ["expense1", "expense2"], "user": user.name}
And it applies to itself as well. Editing the policy table is itself an endpoint guarded by require("admin:permissions"), so an Admin can change who approves expenses without touching code. Two guards keep that power from locking everyone out: roles can't be empty, and Admin can't remove itself from the admin key.
def locks_out_admins(endpoint_key: str, roles: list[str]) -> bool:
return endpoint_key == "admin:permissions" and "Admin" not in roles
@app.put("/admin/permissions")
async def set_permission(endpoint_key: str, roles: list[str], ...):
if not roles:
raise HTTPException(422, "roles must be non-empty")
if locks_out_admins(endpoint_key, roles):
raise HTTPException(422, "cannot remove Admin from admin:permissions")
... # upsert the row, return itTwo Azure traps that cost an afternoon
The full setup (two app registrations, the user_impersonation scope, defining the app roles, assigning users) is in the repo README. Two gotchas are worth pulling out here, because neither errors in a way that points back at the cause:
v2 tokens, or everything 401s. fastapi-azure-auth validates against the v2 token endpoint. It expects aud to be the bare API client-id GUID and iss to end in /v2.0. By default a fresh registration mints v1 tokens (aud = api://<client-id>, iss = sts.windows.net/...), so validation rejects a token that looks perfectly valid at a glance, with invalid_token: "Token contains invalid claims". The fix is one line in the backend API's Manifest:
{ "api": { "requestedAccessTokenVersion": 2 } }
Mint a token, paste it into jwt.ms, and confirm ver reads 2.0.
Direct assignment only. When you assign test users to roles, only direct assignment flows into the roles claim; nested group membership does not. The call succeeds, the token validates, and the user simply has no roles, which is its own silent time sink.
Testing without Azure
Because authorization is a dependency, you swap it in tests. Override azure_scheme with a fake user and get_session with an in-memory SQLite session, and the entire role matrix is unit-testable with no network and no tenant:
def client_as(engine: Engine, *roles: str) -> TestClient:
def fake_user():
return SimpleNamespace(roles=list(roles), name="tester")
def session_override(): # in-memory SQLite
with Session(engine) as s:
yield s
app.dependency_overrides[azure_scheme] = fake_user
app.dependency_overrides[get_session] = session_override
return TestClient(app)
def test_viewer_cannot_approve(engine):
assert client_as(engine, "Viewer").post("/expenses/1/approve").status_code == 403
And because the policy lives in a row, the headline claim is itself a test: an Admin rewrites the row and the same request flips from allowed to forbidden, no redeploy:
def test_repoint_revokes_access(engine):
assert client_as(engine, "Approver").post("/expenses/1/approve").status_code == 200
client_as(engine, "Admin").put( # drop Approver from expenses:approve
"/admin/permissions",
params={"endpoint_key": "expenses:approve"}, json=["Admin"],
)
assert client_as(engine, "Approver").post("/expenses/1/approve").status_code == 403
This is the same Depends paying off again. In production it lets an Admin edit the rules in a table. In tests it lets you swap that table for in-memory SQLite and the real login for a fake user, one line each. One design choice, two wins.
When you reach for an if role == check, ask whether the thing you are encoding is logic or policy. Logic belongs in code. Policy tends to want to change on a schedule you do not control, and the table was cheaper than you think.
September 09, 2026 12:00 AM UTC
Graham Dumpleton
Zero-code tracing with wrapture
The previous post traced the shop with three bindings and a sink, all applied from the program's own entry point. That is fine when the program is yours. It is less fine when the application is one you inherited and would rather not touch, when someone else owns the deployment, or when you simply do not want observation code living inside the thing being observed. For all of those the entry point edit is one edit too many.
The same setup can live in a file next to the project instead, with nothing in the program saying so.
The file
A wrapture.toml says what to observe and where the events go. For the shop from last time, with the card number redacted as before, that is one [[observe]] entry per method and one sink:
[[observe]]
target = "shop:OrderService"
name = "place"
redact = ["card"]
[[observe]]
target = "shop:Gateway"
name = "charge"
redact = ["card"]
[[observe]]
target = "shop:Ledger"
name = "record"
[[sink]]
type = "printer"
The target is always an exact module or module:path, never a pattern, and the members within it come from name for exact members or match for a glob over the target's own immediate members. That is deliberate. A pattern's blast radius is one level of one named container, stated on the line above it, so match = "*" on shop:OrderService can never accidentally wrap something in another module.
The program itself is main.py, and it now contains no mention of wrapture at all:
import orders
orders.run()
The python -m wrapture runner applies the config and then runs the program as __main__, the same -m convention as pdb, cProfile and coverage:
$ python -m wrapture main.py
shop:OrderService.place(amount=500, card='<redacted>', tenant='acme')
shop:Gateway.charge(amount=500, card='<redacted>')
shop:Gateway.charge -> {'id': 'ch_500', 'amount': 500} [8us]
shop:Ledger.record(entry={'id': 'ch_500', 'amount': 500})
shop:Ledger.record -> 'led_ch_500' [7us]
shop:OrderService.place -> {'id': 'ch_500', 'amount': 500} [285us]
shop:OrderService.place(amount=250, card='<redacted>', tenant='globex')
shop:Gateway.charge(amount=250, card='<redacted>')
shop:Gateway.charge !! CardDeclined [6us]
shop:OrderService.place !! CardDeclined [90us]
shop:OrderService.place(amount=120, card='<redacted>', tenant='globex')
shop:Gateway.charge(amount=120, card='<redacted>')
shop:Gateway.charge -> {'id': 'ch_120', 'amount': 120} [4us]
shop:Ledger.record(entry={'id': 'ch_120', 'amount': 120})
shop:Ledger.record -> 'led_ch_120' [4us]
shop:OrderService.place -> {'id': 'ch_120', 'amount': 120} [115us]
That is the same trace as before, from a program whose source has not changed. The ordering is what makes it work. The config is applied before the target runs, but applying it imports nothing: each observe entry registers a post-import hook for its target module, and the bindings land at the moment the application itself imports shop, in the application's own import order. A from shop import OrderService somewhere in the program still picks up the observed class, because the observation is already in place when that line runs, and the program's import order is never changed by observing it.
Keeping the trace
A printer is for watching. For a program that runs longer than you are willing to sit and look at it, the sink is a file. Swapping the [[sink]] entry for a JSON Lines one is the only change:
[[sink]]
type = "jsonlines"
path = "trace.jsonl"
Each completed event is written as one JSON object per line, when the event closes, so every line carries the outcome and the timing. The declined charge from the second order looks like this:
{
"seq": 5,
"parent_id": 4,
"depth": 1,
"kind": "call",
"path": "shop:Gateway.charge",
"thread_id": 140704287927360,
"thread_name": "MainThread",
"started": 1150729.898723529,
"duration": 0.000004197005182504654,
"arguments": {
"amount": 250,
"card": "<redacted>"
},
"exception": {
"type": "CardDeclined",
"message": "card ending 0000 declined"
},
"trace": {
"w3c": {
"trace_id": "12cd461196239288a8b50e265b6a0f1a",
"sampled": true
}
}
}
The seq and parent_id fields are enough to rebuild the tree, and a field that is absent means it was not captured, so a call that returned None and a call whose result was never recorded stay distinguishable. The format is the one that jq, pandas and most log tooling read directly, which means the questions I would otherwise have scrolled a terminal to answer become one-liners. Every call to the gateway, with what it was given and what came back:
$ jq -c 'select(.path | endswith("Gateway.charge")) | {seq, arguments, result, exception}' trace.jsonl
{"seq":2,"arguments":{"amount":500,"card":"<redacted>"},"result":{"id":"ch_500","amount":500},"exception":null}
{"seq":5,"arguments":{"amount":250,"card":"<redacted>"},"result":null,"exception":{"type":"CardDeclined","message":"card ending 0000 declined"}}
{"seq":7,"arguments":{"amount":120,"card":"<redacted>"},"result":{"id":"ch_120","amount":120},"exception":null}
And everything that raised, which shows the exception at the gateway and again at the order that let it escape:
$ jq -c 'select(.exception) | {path, exception}' trace.jsonl
{"path":"shop:Gateway.charge","exception":{"type":"CardDeclined","message":"card ending 0000 declined"}}
{"path":"shop:OrderService.place","exception":{"type":"CardDeclined","message":"card ending 0000 declined"}}
Two properties make this safe to leave running against something real. The application never waits on the file: lines go onto a bounded queue drained by a background thread, and if the queue fills the line is dropped and counted rather than making the observed call block. And the sink captures values as bounded summaries, so an unserialisable argument becomes a short description rather than an error, and no live object is retained. For a process that runs for days the path can carry a date or time variable and rotate on an interval; the output paths section of the documentation has that. The file is also what the exporters read afterwards, so a trace recorded overnight can be rendered for Perfetto the next morning.
No launcher at all
The runner still owns the command line, and sometimes that is not available either. A service manager, a container entry point or a WSGI server starts the process and you do not get to put python -m wrapture in front of it. For that case the same config can be injected at interpreter startup through autowrapt, a package of mine from some years ago that exists precisely to run registered code once site initialisation completes. Two opt-ins gate it, both outside wrapture:
$ pip install autowrapt
$ AUTOWRAPT_BOOTSTRAP=wrapture python main.py
The output is identical to the runner's. Installing autowrapt is what makes interpreter startup do anything at all, and the environment variable names wrapture as the thing to bootstrap. Absent either, the entry in wrapture's package metadata is inert, and wrapture itself has no dependency on autowrapt. Underneath, both doors lead to the same place: the post-import hook machinery in wrapt, which autowrapt was originally built on, is what lets wrapture apply a config to modules that have not been imported yet.
The positioning matters here. Injection is a development, staging and break-glass tool. The unwritten rule for autowrapt has always been that it is not installed on production systems in normal circumstances, precisely because of what it enables, and that installation gate is the feature. Production tracing is the code-level path from the previous post, or a config applied deliberately by the application at startup. Two consequences follow from the mechanism. A config that is missing, or that cannot be applied, warns and lets the process start untraced, because an error at bootstrap would be fatal to an interpreter that has not even started, and the environment variable reaches every Python process launched under it, not only the one you meant. And the bootstrap imports no application code, so bindings still land as the application imports its own modules.
Operating a traced process
Once injected, the process is still operable. The bootstrap keeps its record of what was applied on wrapture.bootstrap.applied, and from a console, a debugger or a signal handler that record answers what is installed and lets you switch it off and on without a restart. Running the shop under python -i so the interpreter drops to a prompt afterwards:
$ AUTOWRAPT_BOOTSTRAP=wrapture python -i main.py
...
>>> import wrapture.bootstrap
>>> applied = wrapture.bootstrap.applied
>>> print(applied.report())
sink: Printer()
applied:
shop:OrderService.place
shop:Gateway.charge
shop:Ledger.record
>>> applied.suspend()
>>> import orders; orders.run()
>>> applied.resume()
>>> orders.run()
shop:OrderService.place(amount=500, card='<redacted>', tenant='acme')
shop:Gateway.charge(amount=500, card='<redacted>')
...
While suspended, the wrappers stay in place and the calls pass straight through, so the second orders.run() printed nothing; after resume() the third printed the full trace again. revert() takes the whole intervention down, restoring the patched locations. The config section of the ad-hoc tracing page covers everything the file can say beyond what I have used here, including capturing log messages as events beside the calls, and naming instrumentation that a package ships for a framework.
That last one is where this goes next, because the shop is not really a program that runs three orders and exits. It is a web application, and a web application has a unit of work that a plain binding cannot see.
September 09, 2026 12:00 AM UTC
September 08, 2026
PyCoder’s Weekly
Issue #751: Profiling, From pandas to Polars, NotImplemented, and More (2026-09-08)
#751 – SEPTEMBER 8, 2026
View in Browser »
Profiling and Making Apps Fast by Default
How do you plan for the performance of your Python applications? What does a performance budget entail, and where should you spend your resources? This week on the show, we speak with Den Odell about his new book “Fast by Default: Practical Performance Engineering.”
REAL PYTHON podcast
Migration Strategies for Going From pandas to Polars
How to scope a pandas to Polars migration, from a single performance-sensitive section to the whole pipeline, and how to execute a full migration by hand or with an LLM.
THIJS NIEUWDORP
The Top Open-Source Code Reviewer on Code-Review-Bench
PR-AF places #2 of 42 overall on Martian’s Code-Review-Bench, ahead of CodeRabbit, Copilot, and Devin. Roughly 3x more valid findings than the commercial tools, at ~10x lower cost per review. Verified findings only. Apache 2.0, self-hosted, runs on any open or closed model. Star & Deploy →
AGENTFIELD sponsor
When to Use NotImplemented
When should you return NotImplemented from a dunder method? Why not return False or raise an exception instead?
TREY HUNNER
Articles & Tutorials
Build a Plugin Architecture With a Pydantic and FastAPI
Learn how to build a plugin architecture across service boundaries using a shared Pydantic API contract. This article walks through registration-time validation, FastAPI endpoints, ownership and authorization decisions, and continuous health checks for independently deployed services.
PATRICKM.DE • Shared by Patrick Müller
Metadata Requests No Longer Tracked as PyPI Downloads
Previously, requests for information about a package on PyPI got counted as a download in the package statistics. This was recently changed to more accurately account only for the downloads of wheels, tar balls, and zip files. This article explains the change.
PYPI.ORG
Which AI Tools Are Worth Using? A Live Course for Python Devs With “No Time to Try Them Out”
Stop stressing over every new AI coding tool release: in one live session on September 12 you learn which categories are worth it, which to skip, and a 60-second test that settles every launch after that. Reserve Your Spot →
REAL PYTHON sponsor
Build Your Own Face Recognition Tool With Python
In this tutorial, you’ll build your own face recognition command-line tool with Python. You’ll learn how to use face detection to identify faces in an image and label them using face recognition. With this knowledge, you can create your own face recognition tool from start to finish!
REAL PYTHON
How to Fix ‘NoneType’ Object Has No Attribute Errors
When Python throws AttributeError: ‘NoneType’ object has no attribute ‘x’, it reads like the interpreter is being deliberately unhelpful, but it’s actually telling you something precise: a variable you expected to hold an object turned out to be None.
SYSTEM CRAFT PRESS • Shared by Bob Morrison
Primer on Python Decorators
In this tutorial, you’ll look at what Python decorators are and how you define and use them. Decorators can make your code more readable and reusable. Come take a look at how decorators work under the hood and practice writing your own decorators.
REAL PYTHON
Type Checking Could Be the Guardrail Your Agent Is Missing
Coding agents write a lot of Python, and they write it fast. Having your agent call a typechecker can prevent common type bugs creeping in. Pyrefly is an open-source typechecker built in Rust that’s fast enough to keep up with your agent’s inference.
PYREFLY TEAM sponsor
Testing Async Python Without Losing Your Mind
Async Python testing patterns that actually work: event loop scope, async fixture lifecycle, and the specific pytest-asyncio / anyio patterns that break under default assumptions.
DEV.TO • Shared by Anonymous
Storing Django Static and Media Files on Cloudflare R2
This tutorial shows how to configure Django to load and serve up static and media files, public and private, via Cloudflare R2 an AWS S3-like cloud storage service.
NIK TOMAZIC
Analysis Paralysis Sucks
Junior developers don’t start because they don’t know enough. Senior developers don’t start because they know too many things that could go wrong. Both are stuck.
KEVIN RENSKERS
Async Programming in Python: From Generators to asyncio
Learn how Python async programming works. Write async functions with async and await, and run slow I/O operations concurrently with asyncio.
REAL PYTHON
Why OOP Exists
Learn the fundamental principles behind Object Oriented Programming (OOP) and how that connects to the Python syntax for class definition.
RODRIGO GIRÃO SERRÃO
Python 3.15 Preview: UTF-8 by Default
Preview the Python 3.15 UTF-8 default: see what changes, try it on a pre-release, and keep your file I/O portable across every platform.
REAL PYTHON
Optimal Seating on the Airbus A380
Mark analyzes the results from a paper that determined the optimal seating arrangement on an Airbus A380.
MARK LITWINTSCHIK
Projects & Code
A Browser DOM, in Python!
GITHUB.COM/BYTEFACE • Shared by byteface
pandas-silent-bugs: 182 Examples Where pandas Is Wrong
GITHUB.COM/THIBAUDLEPAN77-SVG • Shared by Thibaud Lepan
Events
Weekly Real Python Office Hours Q&A (Virtual)
September 9, 2026
REALPYTHON.COM
Python Atlanta
September 10 to September 11, 2026
MEETUP.COM
PyDay Boyacá 2026
September 12 to September 13, 2026
PYDAY.CO
DFW Pythoneers 2nd Saturday Teaching Meeting
September 12, 2026
MEETUP.COM
DjangoCologne
September 15, 2026
MEETUP.COM
PyCon Cameroon 2026
September 17 to September 20, 2026
PYTHONCAMEROON.ORG
Happy Pythoning!
This was PyCoder’s Weekly Issue #751.
View in Browser »
[ Subscribe to 🐍 PyCoder’s Weekly 💌 – Get the best Python news, articles, and tutorials delivered to your inbox once a week >> Click here to learn more ]
September 08, 2026 07:30 PM UTC
Django Weblog
Call for volunteers: Fundraising Working Group
The Django Software Foundation is looking for people to join the Fundraising Working Group.
This is a particularly interesting time to get involved.
The DSF has raised its 2026 fundraising goal to $500,000. That funding is what allows us to continue supporting the Django Fellows, Django Girls, community events, Djangonaut Space, infrastructure, and the many other things that keep the Django ecosystem going. It also gives the DSF the room to do something new: hire its first Executive Director.
You can read more about the DSF's fundraising goals for 2026 in this post.
Getting there will take more than asking people to donate. We need to think about how we build relationships with companies that depend on Django, how we make sponsorships meaningful, how we find new ways for organisations to support the project, and how we communicate the value of investing in Django.
That is where the Fundraising Working Group comes in.
The DSF is hiring an Executive Director who will bring dedicated, day-to-day leadership to the Foundation, including sponsorship development and partner relationships. The Fundraising Working Group will have an opportunity to work closely with the person in this role as we build out our fundraising efforts.
Who are we looking for?
We'd love to have people who have done this before.
If you have experience with fundraising, sponsorships, partnerships, business development, sales, donor relationships, or building relationships with companies, there is plenty of scope to put that experience to work. We need people who can help identify opportunities, open doors, develop ideas, and turn them into actual fundraising initiatives.
But you don't need to be a fundraising expert to join.
Maybe you've never worked on fundraising before, but you know how companies make decisions about supporting open source. Maybe you have ideas for how Django could engage organisations that rely on it. Maybe you are good at building relationships, telling a compelling story, organising initiatives, or simply getting things moving.
Those perspectives are useful too.
We're looking for a group that can bring both experience and fresh ideas; people who can help drive the work as well as people who are excited to learn and contribute.
The working group meets monthly and works asynchronously between meetings. You can read more about how the group operates in the Fundraising Working Group charter.
Interested in joining?
Apply to join the Fundraising Working Group
Whether you have years of fundraising experience or are completely new to it but ready to help, we would love to hear from you.
September 08, 2026 07:27 PM UTC
Python Bytes
#495 Banned
<strong>Topics covered in this episode:</strong><br> <ul> <li><strong><a href="https://www.youtube.com/playlist?list=PLd3Y9yzyC5Uo">EuroPython 2026 videos are online</a></strong></li> <li><strong><a href="https://blog.jetbrains.com/pycharm/2026/08/the-state-of-django-2026-boring-is-so-back/?featured_on=pythonbytes">The State of Django 2026: Boring is so back</a></strong></li> <li><strong><a href="https://four.htmx.org/announcements/2026-08-28-htmx-4.0.0-is-released?featured_on=pythonbytes">htmx 4.0.0 has been released</a></strong></li> <li><strong>🐍 <a href="https://testdouble.com/insights/functionally-zen?featured_on=pythonbytes">Functionally Zen</a></strong></li> <li><strong>Extras</strong></li> <li><strong>Joke</strong></li> </ul><a href='https://www.youtube.com/watch?v=yP5MY1R00LU' style='font-weight: bold;'data-umami-event="Livestream-Past" data-umami-event-episode="495">Watch on YouTube</a><br> <p>Sponsored by us! Support our work through:</p> <ul> <li>Our <strong>courses at Talk Python</strong></li> <li>Consulting from <strong>Six Feet Up</strong></li> </ul> <p><strong>Connect with the hosts</strong></p> <ul> <li>Michael: Mastodon / BlueSky / X / LinkedIn</li> <li>Calvin: Mastodon / BlueSky / X / LinkedIn</li> <li>Show: Mastodon / BlueSky / X</li> </ul> <p>Join us on YouTube at <strong>pythonbytes.fm/live</strong> to be part of the audience. Usually <strong>Tuesday at 7am PT</strong>. Older video versions available there too.</p> <p>Finally, if you want an artisanal digest of every week of the show notes in email form? Add your name and email to our friends of the show list, we'll never share it.</p> <p><strong>Michael #1:</strong> <a href="https://www.youtube.com/playlist?list=PLd3Y9yzyC5Uo">EuroPython 2026 videos are online</a></p> <p>The EuroPython Society has published all 117 recordings from EuroPython 2026 on the official EuroPython Conference YouTube channel. The conference ran July 13-19 in Krakow, Poland and celebrated the conference series' 25th anniversary. The playlist covers keynotes, panels, lightning talks, and full talk recordings across Python core, web, DevOps, data/ML, embedded, and other tracks.</p> <ul> <li>If you missed EuroPython 2026 in Krakow, this is the complete free on-demand archive of one of the year's biggest European Python events.</li> <li><strong>117 videos</strong> now live on the EuroPython Conference YouTube channel, last updated Aug 17, 2026.</li> <li>Michael’s personal watch list.</li> </ul> <p><strong>Calvin #2: <a href="https://blog.jetbrains.com/pycharm/2026/08/the-state-of-django-2026-boring-is-so-back/?featured_on=pythonbytes">The State of Django 2026: Boring is so back</a></strong></p> <ul> <li>State of Django 2026 (JetBrains/DSF survey, ~3,500 devs, 40+ countries) - "boring is so back": Django's core stays reliable while everything around it moves fast</li> <li>Core is stable: Postgres 76–79% for 5 years running, templates ~80%, 43% already on Django 6.0</li> <li>AI is routine now (only 10% use none) but workflow's unsettled - Claude Code leads at 35%, and 56% still just use it for chat, not autonomous edits</li> <li>Tooling is consolidating: uv and Ruff both at 43% adoption, each replacing several older single-purpose tools</li> <li>Type hints are winning (57% use them) but the checker is up for grabs - IDE-built-in leads at 40%, Mypy 32%, with ty/Pyrefly emerging</li> <li>Two Django communities coexist happily: 72% server-rendered templates vs. 53% API-only - and htmx adoption jumped from 5% to 34% in five years</li> </ul> <p><strong>Michael #3:</strong> <a href="https://four.htmx.org/announcements/2026-08-28-htmx-4.0.0-is-released?featured_on=pythonbytes">htmx 4.0.0 has been released</a></p> <p>After 8 months of work, the htmx team shipped 4.0.0, a rewrite that moves internals from XMLHttpRequest to fetch() while keeping the API almost identical to htmx 2. Three changes may need action: attribute inheritance is now explicit via an :inherited suffix, event names follow a htmx:phase:action pattern, and history no longer caches pages in localStorage. Additions include built-in morph swaps, the new hx-partial tag, and many core extensions. htmx 2 stays supported and remains latest on npm until early 2027.</p> <ul> <li>htmx is the go-to frontend layer for Python server-rendered apps (Flask, Django, FastAPI), and 4.0 is deliberately low-drama: nearly behavior-compatible, so teams can upgrade on their own schedule and pick up morph swaps and streaming extensions.</li> <li>Explicit inheritance is the biggest migration item: hx-confirm, hx-headers, hx-target and friends no longer cascade to children unless you append :inherited; hx-disinherit and hx-inherit are gone</li> <li>A CLI upgrade checker (npx htmx.org@4.0.0 upgrade-check) flags spots needing :inherited, renames like hx-disable to hx-ignore, removed attrs like hx-vars, and old event names in templates and JS</li> <li>Events follow htmx:phase:action (htmx:beforeRequest becomes htmx:before:request); most error events collapse into htmx:error and htmx:xhr:* events are removed with XMLHttpRequest</li> <li>History no longer snapshots pages in localStorage; back navigation re-fetches and swaps into the body, fixing a chronic support headache, with a new hx-history-cache extension for sessionStorage caching</li> <li>New features: out-of-the-box morphing swaps, the [HTML_REMOVED] tag for multi-element updates, streaming over SSE/WebSockets/multipart, and hx-live, their Alpine-inspired DOM scripting solution</li> <li>No forced upgrade: 2.x stays latest on npm until early 2027 (4.0 remains next) and is supported indefinitely; the team even ships official LLM skill files for guidance and upgrading</li> </ul> <p><strong>Calvin #4: 🐍 <a href="https://testdouble.com/insights/functionally-zen?featured_on=pythonbytes">Functionally Zen</a></strong></p> <ul> <li>Functionally Zen (Kyle Adams, Test Double) - riffs on "simple is better than complex" with 7 extra tenets for Python simplicity</li> <li>Core claims: idiomatic > non-idiomatic, data > functions, pure functions > impure functions > classes</li> <li>Favorite example: a medical-dosage calculator replaced with a plain lookup dict - no logic, no tests needed</li> <li>Big idea: keep a thin "impure shell" around a "pure core" (Gary Bernhardt's functional core / imperative shell) - push side effects (API calls, DB, files) to the edges</li> <li>Side note: constructors that do I/O are "poison pills" - the side effect infects every class that depends on them</li> <li>Payoff: pure functions and no-mock tests are just easier to read and reason about than the alternative</li> </ul> <p><strong>Extras</strong></p> <p><strong>Calvin</strong>:</p> <ul> <li><strong><a href="https://github.com/astral-sh/uv/releases/tag/0.12.10?featured_on=pythonbytes">uv ships trusted-publisher token revocation and Python 3.15 support</a></strong> </li> <li><a href="https://austinhenley.com/blog/python1024.html?featured_on=pythonbytes">Making a Python interpreter in 1024 bytes</a></li> </ul> <p><strong>Michael</strong>:</p> <ul> <li>Steering council voting is now open</li> </ul> <p><strong>Joke: <a href="https://www.reddit.com/r/iiiiiiitttttttttttt/comments/1w4j5ka/what_single_word_in_it_makes_you_look_like_this/?share_id=pVKlDEXbUPKdu-UH8fspy&featured_on=pythonbytes">Makes you look like this?</a></strong></p>
September 08, 2026 06:48 PM UTC
LernerPython blog, from Reuven Lerner
Claude Code always produces something. That’s the hard part.
For many months, I’ve been using Claude Code several hours each day. Not to experiment, but to get work done. A growing amount of the functionality behind LernerPython.com is now written with Claude Code: the events system, the membership plumbing, and the scheduled jobs that keep the site’s session listings current. And, of course, my Socratic AI tutor, which I’ve not only integrated into my courses, but also into my “Better Developers” and “Bamboo Weekly” newsletters. Even my course-setup software, which I use several times each week when teaching live sessions, was written with Claude Code.
I write less code by hand than I did a year ago, and I ship more of it. Also? I’m having a blast.
The good news: Claude Code always produces something
Long ago, someone told me, “Computers don’t do what you want them to do. They do what you tell them to do.” This was always the case when programming. But with Claude Code, or any agentic coding system, the implications are much bigger. There are gaps between what you wanted to happen, what you specified, and what AI then actually implemented. Just today, Claude told me that gee, it really should have implemented a feature that I had asked for earlier today, and it’s so sorry that it forgot.
Um, you’re forgiven? I guess?
Generally speaking, agents will do what you ask. But you need to understand where and how it might fail, and keep it honest with checklists, follow-up questions, and a very tight development environment. You need to think very carefully not just about the code you want Claude to write, but also about how you can be sure it is really working. Validating your results is, in some ways, more important than the results themselves.
What actually changed how I work
Four things, in rough order of how much they’ve mattered.
A CLAUDE.md file that reads like a contract. Mine says to use uv rather than pip, to add type hints everywhere, to write the test before the implementation, to run ruff before committing, and to commit small and often. It is not documentation. It’s the set of standing instructions I got tired of repeating, and every one of those lines was added as soon as something went wrong. It’s sort of how a company will update its employment contract when they discover something that hadn’t previously been included, and which let an employee do something they disliked.
Tests, not diffs. A test is a claim about behavior, written in a form I can read in ten seconds and disagree with. Test-driven development was always good practice; with an agent in the loop it becomes the primary way you steer. Write the failing test first and the agent has a target it can’t talk its way around. Ensuring 100% coverage, and adding mutation testing into the mix, make it even less likely that things will go off the rails.
Telemetry and logging, much earlier than I used to add them. When code was something I typed, I carried a mental model of it. When code is something I approve, that model is thinner — so I compensate by making the running system easier to observe. I log everything in incredible detail. I get reports e-mailed to me, including self-reflective reports on the AI system itself. I make things visible via APIs, so that I can access and observe information as an administrator.
Small commits as a rollback strategy. If every step is its own commit, a session that goes wrong costs you one git revert, not an afternoon of frustration and debugging. I was always a fan of small commits, and now I’m even more convinced of their use.
Three workshops
I’m running three Claude Code workshops this month. Each gives you four hours of hands-on work. As usual when I teach, I won’t use any slides.
- Wednesday, September 16 — Intro Claude Code with Python. Start from the beginning: set Claude Code up properly, then build a command-line utility and a FastAPI app.
- Thursday, September 17 — Intro Claude Code with Pandas. The same starting point, pointed at data: retrieve, clean, analyze and report on real data sets, including inflation and trade data.
- Wednesday, September 30 — Advanced Claude Code. Commands and configuration tricks, plugins and skills (including Superpowers), writing your own skills, using APIs from within Claude Code, and a strong emphasis on testing and telemetry. Plus building a complex data-analysis web app.
Each runs 5:30–9:30 p.m. London / 12:30–4:30 p.m. Eastern / 9:30 a.m.–1:30 p.m. Pacific. Each is $300, on top of a LernerPython membership. If you’re in PythonDAB, all three are included at no extra charge.
I’ve taught versions of this material inside Apple and Cisco, among other companies. If you attended one of my earlier rounds: the exercises are new, and so is a good deal of the material — but I won’t pretend there’s zero overlap. The two introductions cover some of the same ground, because they have to. The advanced session revisits a little and then spends most of its time on things I’ve learned and folded into my own daily work since the spring.
Not sure yet? Come to the free info session on Monday, September 14, at 5:30 p.m. London / 12:30 p.m. Eastern / 9:30 a.m. Pacific. It’s an hour, it’s free, and you can ask me anything before you decide.
Register for the free info session · Full details and syllabi
The post Claude Code always produces something. That’s the hard part. appeared first on LernerPython.
September 08, 2026 01:04 PM UTC
Graham Dumpleton
Live tracing with wrapture
When I wrote about unit testing with wrapture the pattern in every test was the same: create a binding on a method, open a timeline(), run the code, and read the recorded calls off the tape. What I did not say at the time is that nothing about a binding is specific to testing. A binding observes a call site and emits events, and what happens to those events is decided by whoever is listening. In a test the listener is a tape. Take the tape away and register something else, and the same binding narrates a running program as it goes.
That is the whole idea behind the tracing side of wrapture, and this post is the minimal version of it: the shop from the testing series, three bindings, and one sink.
The shop
The code is the order service from the earlier posts, grown just enough to have something worth watching. A card number now travels with the order, the gateway declines cards ending in four zeros, and each order belongs to a tenant.
class CardDeclined(Exception):
pass
class Gateway:
def charge(self, amount, card):
if card.endswith("0000"):
raise CardDeclined(f"card ending {card[-4:]} declined")
return {"id": f"ch_{amount}", "amount": amount}
def refund(self, charge_id):
return {"id": f"re_{charge_id}"}
class Ledger:
def record(self, entry):
return f"led_{entry['id']}"
class Notifier:
def send(self, message):
return True
class OrderService:
def __init__(self, gateway=None, ledger=None, notifier=None):
self.gateway = Gateway() if gateway is None else gateway
self.ledger = Ledger() if ledger is None else ledger
self.notifier = Notifier() if notifier is None else notifier
def place(self, amount, card, tenant):
charge = self._take_payment(amount, card)
try:
self.ledger.record(charge)
except Exception:
self.gateway.refund(charge["id"])
raise
self.notifier.send(f"order {charge['id']} placed")
return charge
def _take_payment(self, amount, card):
return self.gateway.charge(amount, card)
That lives in shop.py. A second module, orders.py, places three orders, one of which will be declined:
from shop import CardDeclined, OrderService
ORDERS = [
(500, "4111-1111-1111-1111", "acme"),
(250, "4000-0000-0000-0000", "globex"),
(120, "5555-4444-3333-2222", "globex"),
]
def run():
service = OrderService()
for amount, card, tenant in ORDERS:
try:
service.place(amount, card, tenant=tenant)
except CardDeclined:
pass
The question to answer is a simple one. When an order is placed, what actually happens? Which methods run, with what, and what comes back? A log line would answer that only in the places where someone had already thought to add one, and this code has none.
Three bindings and a sink
The entry point applies a binding to each of the three methods that matter and registers a Printer, which is the simplest sink wrapture ships: it prints each event to standard error as it happens.
import wrapture
from shop import Gateway, Ledger, OrderService
import orders
wrapture.binding(OrderService, "place").apply()
wrapture.binding(Gateway, "charge").apply()
wrapture.binding(Ledger, "record").apply()
wrapture.add_sink(wrapture.Printer())
orders.run()
There is no timeline() anywhere in that. The bindings are applied for the life of the process, the sink is registered for the life of the process, and events flow from one to the other. Running it, the output is:
shop:OrderService.place(amount=500, card='4111-1111-1111-1111', tenant='acme')
shop:Gateway.charge(amount=500, card='4111-1111-1111-1111')
shop:Gateway.charge -> {'id': 'ch_500', 'amount': 500} [8us]
shop:Ledger.record(entry={'id': 'ch_500', 'amount': 500})
shop:Ledger.record -> 'led_ch_500' [6us]
shop:OrderService.place -> {'id': 'ch_500', 'amount': 500} [239us]
shop:OrderService.place(amount=250, card='4000-0000-0000-0000', tenant='globex')
shop:Gateway.charge(amount=250, card='4000-0000-0000-0000')
shop:Gateway.charge !! CardDeclined [5us]
shop:OrderService.place !! CardDeclined [60us]
shop:OrderService.place(amount=120, card='5555-4444-3333-2222', tenant='globex')
shop:Gateway.charge(amount=120, card='5555-4444-3333-2222')
shop:Gateway.charge -> {'id': 'ch_120', 'amount': 120} [4us]
shop:Ledger.record(entry={'id': 'ch_120', 'amount': 120})
shop:Ledger.record -> 'led_ch_120' [3us]
shop:OrderService.place -> {'id': 'ch_120', 'amount': 120} [104us]
Each operation gets a line when it begins, indented by how deeply it is nested, and a closing line with the outcome and how long it took. A -> marks a return value and !! marks an exception, so the declined card is visible at a glance, and so is the fact that Ledger.record never ran for that order. These are the real arguments and the real results, the same -> and !! markers that tape.tree() uses in a test, only arriving live rather than being reconstructed afterwards.
The first thing I noticed in that output is something the trace should not contain. The card numbers are in it, in full, because the bindings captured the arguments as given. The same redact() capture policy the testing series used for keeping secrets off a tape works here, since the binding is the same object:
wrapture.binding(OrderService, "place", capture=wrapture.redact("card")).apply()
wrapture.binding(Gateway, "charge", capture=wrapture.redact("card")).apply()
With that in place the opening lines read card='<redacted>' and everything else is unchanged. I have left it on for the rest of the post, since a trace that is going to be looked at, streamed to a file, or sent anywhere, is exactly the place a card number should not be.
What it costs when nobody is listening
The obvious worry about leaving bindings applied in a program is what they cost when nothing is being traced. The recording gate in wrapture is not "is there a timeline" but "is anything listening". A tape scoped to a test is one kind of listener, a process sink is another, and when neither is present an applied binding constructs no event at all. The wrapped method runs with only wrapt's own dispatch on top, which the documentation puts at about half a microsecond per call on the machine it was measured on. That is what makes it reasonable to bind the interesting methods once, in the entry point, and let the sink decide whether anything is recorded.
Seeing less
Three orders is a readable trace. Three thousand is not, and the answer is rarely to bind fewer things, because the point of binding the layers is to have them there when a question comes up. The tools for narrowing sit either at the sink or at the binding.
At the sink, combinators wrap a sink and gate what reaches it. Depth(1, ...) forwards only the roots of each tree, which turns the trace into one opening and one closing line per order:
wrapture.add_sink(wrapture.Depth(1, wrapture.Printer()))
shop:OrderService.place(amount=500, card='<redacted>', tenant='acme')
shop:OrderService.place -> {'id': 'ch_500', 'amount': 500} [149us]
shop:OrderService.place(amount=250, card='<redacted>', tenant='globex')
shop:OrderService.place !! CardDeclined [33us]
shop:OrderService.place(amount=120, card='<redacted>', tenant='globex')
shop:OrderService.place -> {'id': 'ch_120', 'amount': 120} [46us]
At the binding, when= takes a predicate that is consulted before any event exists. A falsey answer means no event is constructed, no arguments are captured and nothing is delivered, which is the cheap way to narrow a hot call site. Here it records orders for one tenant only:
def acme_only(instance, args, kwargs):
return kwargs.get("tenant") == "acme"
place = wrapture.binding(OrderService, "place", when=acme_only,
capture=wrapture.redact("card")).apply()
Running the three orders again gives this:
shop:OrderService.place(amount=500, card='<redacted>', tenant='acme')
shop:Gateway.charge(amount=500, card='<redacted>')
shop:Gateway.charge -> {'id': 'ch_500', 'amount': 500} [7us]
shop:Ledger.record(entry={'id': 'ch_500', 'amount': 500})
shop:Ledger.record -> 'led_ch_500' [6us]
shop:OrderService.place -> {'id': 'ch_500', 'amount': 500} [245us]
shop:Gateway.charge(amount=250, card='<redacted>')
shop:Gateway.charge !! CardDeclined [5us]
shop:Gateway.charge(amount=120, card='<redacted>')
shop:Gateway.charge -> {'id': 'ch_120', 'amount': 120} [4us]
shop:Ledger.record(entry={'id': 'ch_120', 'amount': 120})
shop:Ledger.record -> 'led_ch_120' [4us]
The globex orders are gone, but their gateway and ledger calls are not. A when= decline skips exactly one event, the declined operation's own, and whatever records beneath it still records, now with nothing above it, so each inner call turns up as an anonymous root with no place to explain it. Sometimes that is exactly what you want, since a binding whose only job is to intervene in a call should not silence what runs beneath it. When the intent is "nothing from here down", tree=True says so:
place = wrapture.binding(OrderService, "place", when=acme_only, tree=True,
capture=wrapture.redact("card")).apply()
shop:OrderService.place(amount=500, card='<redacted>', tenant='acme')
shop:Gateway.charge(amount=500, card='<redacted>')
shop:Gateway.charge -> {'id': 'ch_500', 'amount': 500} [7us]
shop:Ledger.record(entry={'id': 'ch_500', 'amount': 500})
shop:Ledger.record -> 'led_ch_500' [6us]
shop:OrderService.place -> {'id': 'ch_500', 'amount': 500} [251us]
Now the decline covers the whole extent of the declined operation, and the trace is one tenant's orders and nothing else. The skipped calls are not simply lost, either. Each binding counts the operations it declined on filtered_calls, and after this run place, charge and record report 2, 2 and 1 respectively (the second globex order raised before reaching the ledger), so a trace shorter than expected can be explained rather than guessed at.
Where this leaves things
The whole intervention is a few lines in the program's entry point: bind the methods that matter, register a sink, and the program describes what it is doing as it runs, with real arguments and real results, and costs next to nothing when nothing is listening. The sink protocol itself is three notifications, so a sink that counts, samples, filters, or writes somewhere of your own is a small class, and the ad-hoc tracing page of the documentation covers that side, along with the other combinators and the collectors that keep numbers rather than events.
Those few lines in the entry point are still lines in the program, though. For code you cannot or would rather not edit, they can move out of the program entirely, into a file that sits next to it.
September 08, 2026 12:00 AM UTC
September 07, 2026
Bob Belderbos
5 design patterns used in my new habit tracker app
There are plenty of habit trackers, but I wanted to build mine with constraints, a calendar view and habit streaks. So I built commitgraph: a small Django + HTMX app. Here are five Python design and testing patterns from building it.
1. Pure functions
Streak counting and calendar shading live in two plain modules. They take dates and return values. And work independently from Django, the database, or the user. That makes them easy to test.
# streaks.py
from datetime import date, timedelta
def longest_streak(dates: list[date], is_due=lambda _: True) -> int:
days = set(dates)
if not days:
return 0
best = run = 0
cur, last = min(days), max(days)
while cur <= last:
if is_due(cur):
run = run + 1 if cur in days else 0
best = max(best, run)
cur += timedelta(days=1)
return best
Non-due days don't break the streak, so a Mon/Wed/Fri habit can build a streak across its scheduled days.
Thanks to this boundary the streak logic is easy to unit test; no database nor user are needed.
Another example: each calendar cell gets a shade from how much you completed that day. It is tempting to compute that in the template. Instead, I turned it into a helper function:
# calendars.py
def shade(completed: int, active: int) -> str:
if active == 0 or completed == 0:
return "bg-[var(--g0)] text-muted"
frac = completed / active
if frac >= 1:
return "bg-[#a84420] text-white"
if frac >= 0.6:
return "bg-[#d97a4e] text-[#3a1c0e]"
return "bg-[#f6ddc9] text-[#3a1c0e]"
assert shade(5, 5).endswith("text-white") is again easy to test and it's reusable in different parts of the app.
2. A schedule is a 7-bit integer
Claude suggested this nifty approach: a habit runs on a subset of weekdays. That is seven yes/no answers, which fit cleanly into a single 7-bit integer.
class Habit(models.Model):
ALL_DAYS = 0b1111111 # every day
WEEKDAYS = 0b0011111 # Mon-Fri (weekday() 0-4)
due_days = models.PositiveSmallIntegerField(default=ALL_DAYS)
def is_due_on(self, day: date) -> bool:
return bool(self.due_days & (1 << day.weekday()))
No join table, no seven boolean columns. Adding "weekends only" is a new constant, not a migration.
The is_due_on one-liner treats due_days as a 7-bit binary calendar, one switch per weekday, ordered right-to-left from Monday (0) to Sunday (6).
day.weekday() gives 0-6 (Mon-Sun). 1 << day.weekday() puts a single bit at that day's position, and & self.due_days is non-zero only if the habit is scheduled that day. bool() turns that into True/False.
3. Let the standard library build the month grid
A month grid can be tricky: leading blanks, trailing blanks, weeks that straddle two months. Python's calendar module already knows all of it.
import calendar
for week in calendar.Calendar(firstweekday=0).monthdatescalendar(year, month):
for day in week:
if day.month != month:
... # a padding cell from the previous or next month
monthdatescalendar returns real date objects, one clean list of weeks; no need to write any logic around how many days April has.
4. A query vocabulary on the model
"Active habits" and "habits active on a given day" show up everywhere. Rather than repeating the filters in every view, I named them on a custom QuerySet so they are chainable:
class HabitQuerySet(models.QuerySet):
def active(self):
return self.filter(archived_at__isnull=True)
def active_on(self, day: date):
return self.filter(
Q(start_date__lte=day)
& (Q(archived_at__isnull=True) | Q(archived_at__date__gte=day))
)
class Habit(models.Model):
objects = HabitQuerySet.as_manager()
Now Habit.objects.filter(user=u).active() is very readable and the relatively complex Q expressions are abstracted into the model.
5. "Done and due" is set algebra
The Today screen needs to know which due habits are already checked off. Two sets and one operator.
due_ids = {h.id for h in habits}
done_ids = set(
HabitCompletion.objects
.filter(habit__user=user, date=day)
.values_list("habit_id", flat=True)
) & due_ids
The intersection drops any completion for a habit that is not due today, so a spurious record can't inflate the count. The logic is the math, not a nest of if statements.
Here is a REPL snippet that demonstrates how set intersection (&) filters down to only the elements present in both sets:
>>> due_ids = {10, 11, 12, 13} # Habits scheduled for today
>>> completed_ids = {11, 13, 99} # Logged completions (99 is a stray record)
>>> # The '&' operator keeps ONLY IDs present in BOTH sets
>>> done_today = completed_ids & due_ids
>>> print(done_today) # 10, 12 and 99 are dropped because they are not in both sets
{11, 13}
I have also seen this used in RBAC logic to cross-check user roles with endpoint permissions. The intersection of the two sets is the effective permissions a user has.
The thread running through all five patterns: when a piece of logic doesn't strictly need the database or a dependency, stick to this boundary. Fetch your primitive values, do the math in plain Python, so it's more testable in isolation.
What is one function in your Django views that would be much easier to trust and test if you decoupled it?
September 07, 2026 12:00 AM UTC
September 06, 2026
Glyph Lefkowitz
... but what about video games?
I get asked this rhetorical question a lot, in various forms:
Sure, datacenters might use a lot of energy, but you don’t have to use a hosted frontier model to do software development. What if I just run a local open-weights model to do some coding, with an open-source coding agent? Video games also use my GPU. Is local model development any worse than playing a video game?
So I want to write down my comprehensive answer to this: Yes, using an LLM to write some code is worse than playing a video game, for a few reasons.
Video Games Are Interactive, LLMs Are Batch Jobs
Video games use compute to respond to human input. You are using your GPU while you are looking at a screen, displaying an image. When you are done playing, you shut off the game, and your computer goes back to idle. It’s much less energy. By contrast, agentic loops with evals (the only kind of “AI” that is meaningfully any good at coding) are running hot, for days. To use the most recent example of such a thing, a very rough first sketch of an implementation of a Windows graphics API backend to help port a paint program to other platforms, it took 3 weeks of Claude time, “day and night”. Do you play a lot of video games for 500 hours to make it past the tutorial level, while also using other computers for other things, as well as the rest of your carbon footprint?
Video Games Need Development, LLMs Need Training
Video games use compute to respond to human input during development, too. Your game has to be made, but your LLM has to be trained. LLMs use a historically extreme amount of power, probably using more than the entire Internet, but it’s kind of hard to say. Still, it seems a reasonable estimate to within several orders of magnitude that even over a multi-year project with hundreds of developers, the power used to develop an individual video game is nowhere close to training even a small LLM.
This is true even for local models. OpenAI has openly claimed that DeepSeek “stole its intellectual property”, and I have heard grumblings that none of the open-weights generalist models could realistically exist without the massive lift that the frontier labs are doing with their training, in various other ways too. Secrecy throughout the industry makes this kind of impossible to understand rigorously, but it seems fair to say that you are partially culpable for all that famously energy-intensive frontier lab training if you’re using a local model.
And They Keep Needing Training
You also can’t dismiss this as a sunk cost, because in order to stay current with industry developments, models need to be updated with new information from the rest of the world, which means that you need to keep training them. Beyond the energy for your own use, if you want a real-life agentic workflow that actually does useful stuff, practically speaking you would still need to update your local models over and over again, at least once every few months, which means you would be incentivizing continued energy consumption by whoever was doing that training for you, including the energy cost of scraping.
Let’s Be Real Here, You Aren’t Actually Using A Local Model
This question is a hypothetical thought experiment. Despite synthetic benchmarks that keep showing there isn’t much difference between open weight and frontier models, nobody’s actually using local models for much of anything beyond sharing those talking points. Depending on which benchmark you’re looking at, maybe it’s good enough or maybe it’s worse.
As an inveterate AI hater, all these systems seem pretty bad to me, but it seems that people who find them useful tend to subjectively believe the frontier models are worth the premium, and that’s what they’re actually using. Once you have accepted that it is OK to use LLMs for coding at all, it seems like a very quick slippery slope on down to “we’ll go ahead and use the frontier models for now anyway, but we could be ethically better in the future by switching to an open weights one, that option is always available”.
There’s A Reason We Have Data Centers
Devolving power usage to local LLMs might be good to make users responsible for their costs and decrease the impacts to communities that are physically next to huge concentrations of power utilization, not to mention generation. However, there’s a reason that it makes sense for the providers to build these giant facilities: economies of scale reduce total power consumption, they don’t increase it. If you do all the same stuff with a local model that they have to do in hosted environments, it will probably take more power, even though you will be incentivized to do different stuff. This incentive to “do different stuff” is why although local models can hypothetically hold their own against the frontier labs for some tasks, when people or businesses take their inference costs in-house they often find that it’s too painful and move back to hosted LLMs.
There Are Problems Other Than Power
These are subjects for a different post, but you have to consider a lot of other externalities: AI psychosis, de-skilling, comprehension debt, cultivating a dependency, introducing security defects, limiting your design space based on what LLMs can understand, context rot, wasting time on invalid solutions, introducing unpredictability into your workflows. You still have to consider the total cost benefit ratio.
To Sum Up
Local LLMs might alleviate some of the harms from using the hosted frontier providers. There are fewer privacy concerns, you can measure your power utilization and be more directly responsible for it, you can build interfaces with affordances that are less oriented towards addiction and dependency than the major frontier labs’ harnesses.
But they’re not automatically “the same as playing a video game” just because they can use the same GPU.
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!
September 06, 2026 10:57 PM UTC
The Python Coding Stack
How I Code (Late 2026 Version)
You open a blank .py file in your favourite IDE. You have a blank page in front of you. You start writing code.
This is how it used to be. And perhaps it’s how it still is for you. This is how I wrote computer programs just over half a year ago, too. But things started changing gradually for me earlier in 2026.
Last week, I ran a live course. In the first few minutes of the first live session, I opened a blank .py file in my IDE. There was nothing on my screen. Then I wrote code. Word by word. Line by line.
And about half an hour into this first session, it dawned on me: the last time I had gone through this process of opening a blank file and writing code from scratch was the previous time I had to run a live course a couple of months earlier.
My Agent and I
Every bit of code I had worked on in between these two live courses was written by my AI agent. I was very much involved in those processes, and I read and reviewed lots of code. But I didn’t write any.
Like many, I have mixed feelings about this. I’m getting a lot more done. But I miss the process of writing code, exploring options, putting a project together, function by function, class by class, module by module.
The Coding Gym
I’m going to force myself to write code. I’ll make time for it, as I enjoy it too much and I don’t want to lose the skills and fluency. But it will be like going for a run or lifting weights in the gym. I don’t run because I need to get from point A to point B quickly and I don’t lift weights because I need to move those weights from one place to another. I do those things to keep my body healthy and strong. I’ll code by hand to do the same thing for my mind.
A Real World Example
But let’s go back to how I code now, in late 2026. And let me give you an example of something I worked on last week and used successfully this weekend.
It’s not Python code. It’s a Google Sheets spreadsheet. But that doesn’t matter. The process is the same.
The Scenario
I’m a member of an athletics club (or ‘track and field’, depending on which flavour of English you speak). This weekend we had our club championships and I was tasked to take care of results and overall points. We award a shield for the best track performance and another for the best field performance.
The Pain Point
Last year I was asked to help out at the last minute (and by “last minute” I mean it almost literally, as it was 20 minutes before the meeting started). It was a nightmare. I wasn’t responsible for collating the results, that was done by someone else manually as if it was 1965. I just had to work out their performance points by tapping in lots of numbers into an online calculator using my phone. A laptop would have been easier, but I was only tasked with this job on the day and I only had a phone with me.
This year I had more notice, so I chose to make my life easier... and the whole process smoother.
The Problem
The problem we’re trying to solve…
…is not rocket science. Officials on the track or in the field write results on result sheets and they’re passed on to the result room by a runner. These sheets contain the name of the event, the athletes’ bib numbers, and their performances. So a result slip may look like this:
| Event: U16 Boys 100m-1 | |
| ---------------------- | --------------- |
| Athlete | Performance (s) |
| 23 | 11.5 |
| 47 | 11.8 |
| 13 | 11.9 |Here’s what happened last year:
My colleague had to cross-reference bib numbers with a printed-out sheet of registered athletes. She matched names and age groups, then she sorted results into separate sheets for each event and age group.
I had to look up each athlete’s date of birth from a spreadsheet I’d been given that morning (browsing spreadsheets on a phone screen is not fun), then tap their age, event, and performance into an online calculator, one by one, and write the resulting points on a separate sheet.
The points allow you to compare performances by athletes of different ages in different events, so that we could then find the single best track performance and the single best field performance.
I was already thinking, at last year’s event, of the Python program I could write to automate all of this.
The Solution (I would have never bothered with in 2025)
So, fast-forward to a couple of weeks ago. This time I had some time to come up with something. But I opted to create a Google Sheet instead of a Python program for two reasons:
It’s much easier for others in the club to use it and share it in the future.
My usual objection that it’s easier and more fun to write a Python program than create a complex spreadsheet full of linked tabs and formulae didn’t apply. Either way, it was my agent who was going to do the hard work.
And here’s the thing. If I was doing this last year, when I was still coding most things manually (and occasionally opening a ChatGPT window to ask a few things), I still wouldn’t have done this. I wouldn’t have had the time.
But this year was different. Sure, if anything I was busier this year than I was last year. But this year I had agents at my fingertips. Agents who know me and have been working alongside me for a while.
Here’s why I wouldn’t have bothered doing this myself:
The age group cut-off dates are different for younger age groups (31 August), older age groups (31 December), and Masters athletes (over 35s, where the age group is determined on the day of the competition.)
There are published points tables to compare performances in different events for senior athletes.
There are published points tables to compare performances in the same event for older athletes, above 30.
There are older tables to deal with age comparisons for under 30s
The system needs to take care of all this. None of it is too difficult to write in a Python program or to create one of those power-spreadsheets that link everything together. But it would have required some time and patience, which I didn’t have.
But here’s what I had time for…
I was using an agent I communicate with through Discord, which I also have on my phone. I have a good dictation tool on my phone, too. So in the past week, whenever I was preparing dinner, or waiting in line at the shops, or sitting on the sofa in the evening watching TV, I could have a chat with my agent to guide him (it?) to what I wanted. I knew what I wanted. I just didn’t have the time or desire to do it myself.
And after a few days of these on-and-off conversations, I had a Google Sheet with 20 tabs, including all the points tables for the various scenarios, all the age group rules, all the registered athletes, who-knows-how-many formulae linking columns, rows, and tabs, and, importantly, just two tabs to input results, one for track events and one for field events.
You just enter the event session (selecting from a drop-down menu), the bib number, and the performance. And that’s it. The spreadsheet works out each age group results, the points for each athlete, it shows a live points leaderboard, and so on.
As I said, none of this is rocket science. I know I would have been able to create this spreadsheet by hand, or write Python code that does the same thing. But I would have needed more time. And I wouldn’t have been able to multitask as I did last week.
Oh, and one more thing: I also asked my agent to check the spreadsheet every 15 minutes during our club championships, read the latest results, and post them on a Telegram channel I could share with everyone at the track. So we had live results published online too. I’d never have bothered to set that up manually.
Programming Will Never Be The Same
The real-world example I described above doesn’t represent every programming task I work on with the help of my AI agents. This was a hobby-type project that I probably wouldn’t have worked on otherwise because I didn’t have the time. It wasn’t too difficult, but it would have been time-consuming to do by hand. The stakes weren’t high. Sure, we didn’t want to make mistakes when assigning medals and shields, but it was easy to spot obvious mistakes, and this wasn’t the Olympic Games, either!
I’ll write about other case studies soon, including ones where I was more closely involved in the nitty-gritty of the Python code even though I didn’t write any, nor make any changes to the code by hand.
Do you want to master Python and programming one article at a time, even in this age of AI? Then don’t miss out on the articles in The Club which are exclusive to premium subscribers here on The Python Coding Stack
Coming Soon… Exploring SOLID Through AI-Assisted Coding
I’m starting a series on the SOLID principles here on The Python Coding Stack soon. I’ll write posts about each of the five principles, and I’ll do it my way, the same style I always use when tackling Python topics.
I’ll also have a series of posts, which will include video, where I’ll work on a project from beginning to end using my AI agent and reviewing the code it writes. This project will meander through several OOP concepts, and you’ll be able to see the SOLID principles come in naturally into the project as solutions to problems we might encounter as the project grows.
Stay tuned.
How far are you in the traditional-coding-to-AI-coding arc? Leave a comment and let’s compare notes!
Join The Club, the exclusive area for paid subscribers for more Python posts, videos, a members’ forum, and more.
You can also support this publication by making a one-off contribution of any amount you wish.
For more Python resources, you can also visit Real Python—you may even stumble on one of my own articles or courses there!
Also, are you interested in technical writing? You’d like to make your own writing more narrative, more engaging, more memorable? Have a look at Breaking the Rules.
And you can find out more about me at stephengruppetta.com
September 06, 2026 08:25 PM UTC
September 05, 2026
Graham Dumpleton
Beyond callables in wrapture
Every example in this series so far has wrapped a call. A binding named a method, and what flowed through the call was recorded or changed. Plenty of what a test needs to control is not a call, though. An outcome stored in an attribute, an environment variable that must be set or missing, a settings dict that other modules imported by reference at import time, a formatter looked up in a registry, and a generator whose interesting behaviour is spread over its consumption. unittest.mock and pytest between them cover most of this with patch.dict, monkeypatch.setattr, monkeypatch.setenv and so on, one idiom per shape. wrapture spells all of them as bindings, which buys the same lifecycle everywhere, and in a couple of places lets the binding observe as well as hold.
Attribute bindings
A binding on a class attribute which is not a callable is detected as attribute mode, and instead of on_call it has on_get, on_set and on_delete, one channel per operation. Under the covers it installs a data descriptor on the class, wrapping whatever was there before, so a property's getter still runs and writes still land in the instance dictionary. Take a model with a status:
class Model:
status = "draft"
def publish(self):
self.status = "published"
def archive(self):
self.status = "archived"
Inside a timeline, reads and writes record as get and set events on the same tape as everything else:
status = wrapture.binding(Model, "status")
with wrapture.timeline(status) as tape:
model = Model()
model.status
model.publish()
model.status
print(tape.tree())
status.events.of_kind("set").with_value("published").assert_once()
get __main__:Model.status -> 'draft'
set __main__:Model.status = 'published'
get __main__:Model.status -> 'published'
That assertion says publish() wrote the status exactly once, without the test knowing anything about how publish() works inside. A get event records the value read in result, the same field a call's return value uses, and a set event records the value written in value.
The channels carry the same kinds of verb as on_call. on_get.returns(value) answers a read without touching the real attribute, on_set.rejects() makes a write an AttributeError, on_set.validates(check) checks a written value and lets it through, and decorates() takes full control with the real operation handed in as a function. A guard on state transitions, which needs the current value as well as the new one, is a decorates():
ALLOWED = {("draft", "published"), ("published", "archived")}
def guard(write, instance, value):
current = instance.status
if (current, value) not in ALLOWED:
raise ValueError(f"cannot move from {current} to {value}")
write(value)
status.on_set.decorates(guard)
With that applied, publish() on a fresh model works, a second publish() raises cannot move from published to published, and archive() then works. The real write happens through write(value) when the guard allows it. Attribute channels have phases too, so on_get.returns_from([...]) can read one way for two reads and another afterwards.
Two details come up as soon as this is used on real code. An attribute assigned in __init__ rather than defined on the class does not exist when the binding is created, so the binding takes missing_ok=True, and the write made in __init__ is then recorded like any other. And when the attribute is a property whose getter does work, the get event is the parent of whatever that work recorded, which is exactly the question a lazy-loading bug turns on:
class Account:
def __init__(self):
self._balance = None
def load(self):
return 42
@property
def balance(self):
if self._balance is None:
self._balance = self.load()
return self._balance
get __main__:Account.balance -> 42
__main__:Account.load() -> 42
get __main__:Account.balance -> 42
The first read triggered the load and the second was served from the cache, which is what the property was written to do, and a test can now assert it.
One limit follows from the mechanism. A descriptor on a class fires for access through instances, so Model.status read off the class itself returns the descriptor without recording, and a class-level write replaces the descriptor outright, which the binding reports by going inactive rather than pretending. The known limitations page has the details.
Module attributes
A module's plain data is detected as attribute mode too, so a constant or a flag on a module gets the same three channels. A module cannot take a descriptor directly, so while a binding on it is applied the module is given a private subclass of its type with the descriptor installed there, and the original type comes back when the last binding is removed. isinstance(module, ModuleType) and inspect.ismodule() are unaffected, and the class is named module so reprs read the same.
What is intercepted is access through the module object. Code that did from config import TIMEOUT at import time holds the value already, and reads through vars(config) bypass the descriptor, which is the same caveat that applies to patching a module attribute with mock.
Value bindings
Often a test does not want to observe anything. It wants an environment variable set, a settings key changed, or a module constant lowered, for the duration of the test and then put back. That is a value binding: name the owner positionally, name the slot with attr= for an attribute or item= for a mapping entry, and say what it should hold. The pricing function below reads its configuration from all the usual places:
config.SETTINGS = {"currency": "USD", "tax_rate": 0.2}
config.TIMEOUT = 30.0
config.FORMATTERS = {"plain": lambda total: f"total={total:.2f}"}
def price(amount, style="plain"):
if "API_KEY" not in os.environ:
raise RuntimeError("API_KEY is not configured")
total = amount * (1 + config.SETTINGS["tax_rate"])
formatter = config.FORMATTERS[style]
return f"[{config.SETTINGS['currency']} within {config.TIMEOUT}s] " + formatter(total)
An environment variable is one entry of os.environ, so it is item=. overrides() holds the value while applied, and on exit the prior state comes back, whether the variable existed before or not:
api_key = wrapture.binding(os.environ, item="API_KEY")
with api_key.overrides("sk_test"):
print(price(100))
print("API_KEY" in os.environ)
[USD within 30.0s] total=120.00
False
The other direction is hides(), under which the slot is absent, which is how the missing-configuration branch gets tested even on a machine where the variable is set. overrides(None) cannot say that, since None is a value that is there. A module constant is the same shape with attr=, and the module can be named by import path so the test needs no import of its own:
with wrapture.binding("config", attr="TIMEOUT").overrides(0.5), api_key.overrides("sk_test"):
print(price(100))
[USD within 0.5s] total=120.00
A value binding holds a value and observes nothing. It has no channels, no events and no phases, and it says so if you ask for them. The two spellings differ by exactly that: binding("config", attr="TIMEOUT") holds, and binding("config", "TIMEOUT") intercepts. When the question shifts from "hold this value" to "does the retry path re-read the timeout, or did it cache it", the same location upgrades to the interception form and each read becomes an event:
timeout = wrapture.binding("config", "TIMEOUT")
timeout.on_get.returns(0.5)
with timeout, wrapture.timeline() as tape, api_key.overrides("sk_test"):
price(100)
price(100)
print([event.kind for event in tape.for_binding(timeout)])
['get', 'get']
Two calls, two reads. price() reads the timeout every time, and the tape proves it.
Everything around bindings applies to value bindings. They are context managers, they can be suspended and resumed, active reports whether the slot still holds what the binding put there so a teardown can see that something else overwrote it, and the pytest plugin's leak sweep reports one left applied. In the fixture shape one binding is applied holding nothing and each test says what the slot should be, api_key.overrides("sk_test") in one test and api_key.hides() in the next.
Mapping bindings
The settings dict has a complication. Other modules did from config import SETTINGS at import time, so they hold the same dict by reference, and a test that replaces config.SETTINGS with a new dict strands them with the old one. mode="mapping" on the location mutates the one dict in place and never replaces it, so every holder sees the test's content, and the original entries come back on exit in their original order:
SETTINGS = config.SETTINGS # a holder, as another module would have
settings = wrapture.binding(config, "SETTINGS", mode="mapping")
with settings.updates({"tax_rate": 0.0}), api_key.overrides("sk_test"):
print(price(100))
with settings.overrides({"currency": "EUR", "tax_rate": 0.1}), api_key.overrides("sk_test"):
print(price(100))
print(SETTINGS, SETTINGS is config.SETTINGS)
[USD within 30.0s] total=100.00
[EUR within 30.0s] total=110.00
{'currency': 'USD', 'tax_rate': 0.2} True
updates() merges the named keys over what is there, which is patch.dict's default, and overrides() makes the given entries the whole content, which is patch.dict(..., clear=True). Both took effect through the holder's reference and both restored it. Three dict spellings exist for three different intentions: item= for one entry changed or absent, attr= to make config.SETTINGS a different object with holders of the old one unaffected, and mode="mapping" for the one dict to hold these entries for every holder.
Bindings group, and a group applies and removes atomically, so a test that needs several of these pinned at once does it in one declaration, and as a fixture the group is a with around a yield:
pinned = wrapture.bindings(
api_key=wrapture.binding(os.environ, item="API_KEY").overrides("sk_test"),
settings=wrapture.binding(config, "SETTINGS", mode="mapping").overrides({"currency": "EUR", "tax_rate": 0.0}),
timeout=wrapture.binding("config", attr="TIMEOUT").overrides(0.5),
)
with pinned:
print(price(100))
[EUR within 0.5s] total=100.00
A callable held in a mapping
The formatter registry is configuration too, a callable in a dict. A value binding could swap the entry wholesale, but naming the entry with mode="callable" wraps it instead. The stand-in is installed in the slot, records like any bound callable, has phases like any bound callable, and the original entry comes back on removal:
loud = wrapture.binding(config.FORMATTERS, item="plain", mode="callable")
loud.on_call.transforms_result(str.upper)
with loud, api_key.overrides("sk_test"):
print(price(100))
print(config.FORMATTERS["plain"](120.0))
[USD within 30.0s] TOTAL=120.00
total=120.00
The real formatter ran and its result was adjusted on the way out. This reaches a handler in a dispatch table with the whole call vocabulary, which is something that previously needed the callable to be pulled out and wrapped by hand.
Generators and iteration
A callable that returns a generator produces its values later, one at a time, as the caller iterates. That changes both what recording means and what behaviour can do. Take a paginated catalogue and two consumers, one that reads to the end and one that stops as soon as it finds what it wants:
class Catalogue:
def __init__(self, records, page_size=2):
self.records = records
self.page_size = page_size
def pages(self, cursor=0):
while cursor < len(self.records):
yield {"cursor": cursor, "items": self.records[cursor:cursor + self.page_size]}
cursor += self.page_size
def collect_ids(pages):
ids = []
for page in pages:
ids.extend(item["id"] for item in page["items"])
return ids
def first_match(pages, predicate):
for page in pages:
for item in page["items"]:
if predicate(item):
return item
return None
A test that hands the consumer a canned list of pages proves it can add up ids and nothing else. A list is never lazy, cannot be abandoned, and cannot fail between items, so the properties a streaming consumer is written to have are exactly the ones such a test cannot check.
Binding the generator method records one event covering the whole iteration, not one per page, and the event's items field counts what was pulled through it. Reading to the end fills in result with the generator's return value, None here:
pages = wrapture.binding(Catalogue, "pages")
with wrapture.timeline(pages) as tape:
collect_ids(catalogue.pages())
event = pages.events.first
print(event.items, event.result)
3 None
Stopping early looks different. first_match() finds id 3 on the second page and returns, dropping the generator before it is exhausted. The event closes with the item count reached and no result at all, wrapture.MISSING rather than None, and no -> in the tree, which is the honest signal that the iteration never finished:
with wrapture.timeline(pages) as tape:
first_match(catalogue.pages(), lambda item: item["id"] == 3)
event = pages.events.first
print(event.items, event.result is wrapture.MISSING)
2 True
That already answers "how far did it read" and "did it finish" without touching the consumer. Item values are deliberately not captured on the tape, since a long stream would retain every item and no policy can guess which ones matter. When a test wants to see the items, or react to them, it says so with an iterator proxy. iterator() creates a factory with no target, behaviour is configured on its channels, and calling the factory with a generator returns a wrapped generator applying that behaviour. Since the factory takes an iterator and returns one it slots straight into the binding's transforms_result():
cursors = []
outcomes = []
watch = wrapture.iterator()
watch.on_item.validates_item(lambda page: cursors.append(page["cursor"]))
watch.on_finish.validates(lambda value: outcomes.append(("finished", value)))
watch.on_abandon.notifies(lambda: outcomes.append(("abandoned", None)))
pages.on_call.transforms_result(watch)
with pages:
collect_ids(catalogue.pages())
print(cursors, outcomes)
cursors.clear(); outcomes.clear()
with pages:
first_match(catalogue.pages(), lambda item: item["id"] == 3)
print(cursors, outcomes)
[0, 2, 4] [('finished', None)]
[0, 2] [('abandoned', None)]
on_abandon fires when a started, unexhausted generator is closed, whether explicitly or because the consumer dropped it and the garbage collector closed it. That is the question nothing else can see asked: the loop that stopped early, the generator left half-consumed. The proxy also has on_error for an iteration that raised, and on_item.transforms_item() to rewrite each item on its way through.
An item stage that raises fails the iteration at that point, as if the generator itself had raised while producing that page, which is how to test what a consumer does when page two fails to arrive:
def fail_at(position, exc):
seen = 0
def check(page):
nonlocal seen
seen += 1
if seen == position:
raise exc
return check
flaky = wrapture.iterator()
flaky.on_item.validates_item(fail_at(2, OSError("page 2 failed")))
pages.on_call.transforms_result(flaky)
With that applied, collect_ids() receives the first page and then an OSError on the second, and a consumer written to cope with that can be tested doing so.
One lifecycle for all of it
The thread through everything here is that whichever shape a patch takes, it is a binding, and everything that applies to a binding applies to it. It is a context manager and it has a decorator form. It groups with other bindings and the group applies and removes as one unit. It can be suspended and resumed, it knows whether it is still in place, and the pytest plugin's leak sweep reports it if a test forgets to remove it. Where the shape allows it, the same binding that holds a value can be upgraded to one that sees who reads it, and a callable pulled out of a dict gets the same phases and recording as one on a class.
The monkey patching guide is the full reference for every binding mode, and the worked examples on pinning configuration, checking that resources are released and testing generators and streamed results each take one of the questions above further than a blog post has room for.
September 05, 2026 12:00 AM UTC
Armin Ronacher
Latent Powers
A few weeks ago I felt like it would be fun to see if I can make one of those cheap Chinese CarPlay dongles run something other than the stock firmware. The idea was that rather than just forwarding CarPlay, why not do something more interesting with them? They all work quite similarly: they act as bridges between your car and the phone. From there they deal with video and audio streams and pass some other data through. Most of them also bring up a custom UI for pairing and have a web interface that your phone can reach for updates.
Long story short: I had a conversation with Fable and Sol via Pi about what could be done with such a dongle or whether I should use a Raspberry Pi instead if I wanted to do my own thing there. I figured it might be quite fun to run my own code while still allowing regular CarPlay to pass through.
Through working with the LLM I learned about CatPlay, which is a Rust reimplementation of the CarPlay protocol that can run on Carlinkit devices. In particular, it can run on the Carlinkit Mini Ultra, which I figured would be easy enough to buy. I do have a few CarPlay adapters around, but I did not have that particular model, so I bought one on Amazon. Twenty-four hours later, I had a device in my hand that was branded as a Carlinkit Mini Ultra, but instead of being the Ingenic device that the original author used, it turned out to be something else.
This is normally where the story would stop. However, it’s 2026. Armed with a bit of knowledge about how these systems work, I managed to have some fruitful discussions with Kimi K3 and Sol and figure out how flash the device and in turn, how to make CatPlay compile for that SoC.
I guess that hacking these USB devices is not necessarily hard, but it’s laborious and you can easily end up bricking your devices. It also just sucks because sometimes you need to work with someone else’s code that does not itself run on your machine. In the past, I would abandon many such projects for lack of tenacity. But my clanker is tenacious.
But so are all of our clankers. Some of the projects we’re now attempting are happening because of conversations we have with them. In this case I did not find or decide on CatPlay, the model did. It was not the only suggestion, but it became the best starting point after discarding others.
And I discover this more and more. Particularly when we have solitary interactions with these models, some of us “independently” decide to work on similar projects. When I talked with an acquaintance about CarPlay he also mentioned recently that he decided to try something similar because he too wanted to see if he can get his own agent be hooked up with the car. And guess what: he too learned about the CarPlay hacking community, and that it’s an option, from the models and roughly around the same time.
It really got me thinking about how this could create situations in which completely independent people end up building things they believe are their own ideas. Yet they were inspired or pushed towards doing something by a conversation with an LLM — a conversation that someone else also had. What if we took paths, because those were the paths that were more likely with current generation models? There is a running joke in the AI builder community right now that we’re all working on the same things, and in many ways it feels like we are. That might be because those things are obvious, or it might be partly because we all use the same models with the same capabilities.
A few months ago, I first saw Lucas Meijer share the idea to make a model in Pi produce HTML reports rather than Markdown. I thought that was pretty unique. Except, well turns out the models are probably trained more and more for that (e.g. Claude Artifacts), and now it has become for many the default choice for sharing reports.
How much of what we build comes from eliciting the same latent capabilities from the same models? Did the models make us prompt them that way? Was it because we shared ideas on Twitter and other communities that inspired us? Or is it all unrelated?
There is something powerful and strange about how LLMs diffuse knowledge and capabilities, while perhaps also nudging us all simultaniously and independently toward building the same things.
September 05, 2026 12:00 AM UTC
September 04, 2026
Python Morsels
Creating temporary files in Python
How to create temporary files and directories in Python using the tempfile module's NamedTemporaryFile and TemporaryDirectory.
Making a temporary file
To make a temporary file in Python, you can use the NamedTemporaryFile context manager from the tempfile module in Python's standard library:
from tempfile import NamedTemporaryFile
with NamedTemporaryFile(mode="wt") as file:
file.write("Temporary text.\n")
print(f"The filename is {file.name}")
Note this context manager will delete the file as soon as it exits, which may be a problem if we actually want to use the file after the context manager has exited:
>>> from tempfile import NamedTemporaryFile
>>> with NamedTemporaryFile(mode="wt") as file:
... file.write("Temporary text.\n")
... filename = file.name
...
16
>>> open(filename).read()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
open(filename).read()
~~~~^^^^^^^^^^
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmph52sw7ra'
Pretty much every time I make a temporary file, I need to close the file without actually deleting it so that I can then pass the filename off to other code that actually uses the file.
Leaving temporary files open
If you'd like to make …
Read the full article: https://www.pythonmorsels.com/temporary-files/
September 04, 2026 02:30 PM UTC
Python Anywhere
Annual plans, PostgreSQL 15, and easier database management
We deployed our latest system update to our EU-based system on 23 June 2026 and to our US-based system on 28 July 2026.
People creating an account or upgrading from a free account can now choose an annual plan. PostgreSQL 15 is available, and the Databases page now makes it easier to restart PostgreSQL servers and keep track of MySQL storage limits.
September 04, 2026 10:00 AM UTC
Talk Python to Me
#561: TonIO, a Multi-threaded Async Runtime for Python
How many cores does your machine have, 10, 18? Your async Python code uses just one of them. That isn't a bug in asyncio. That's the design, and optimizing event loops to be faster by 20% doesn't change it. So Giovanni Barillari started over. Joe is the creator of Granian, the Rust-based server that powers Talk Python. His new project is TonIO, an async runtime written from scratch for free-threaded Python. Real threads, a handful of primitives instead of asyncio's pile of them, and it flat out refuses to start if the GIL is on.<br/> <br/> <strong>Episode sponsors</strong><br/> <br/> <a href='https://talkpython.fm/sentry'>Sentry Error Monitoring, Code talkpython26</a><br> <a href='https://talkpython.fm/devopsbook'>Python in Production</a><br> <a href='https://talkpython.fm/training'>Talk Python Courses</a><br/> <br/> <h2 class="links-heading mb-4">Links from the show</h2> <div><strong>Guest</strong><br/> <strong>Giovanni Barillari</strong>: <a href="https://github.com/gi0baro?featured_on=talkpython" target="_blank" >github.com</a><br/> <br/> <strong>Granian</strong>: <a href="https://github.com/emmett-framework/granian?featured_on=talkpython" target="_blank" >github.com</a><br/> <strong>Hyper</strong>: <a href="https://github.com/hyperium/hyper?featured_on=talkpython" target="_blank" >github.com</a><br/> <strong>Free threaded Python</strong>: <a href="https://docs.python.org/3/howto/free-threading-python.html#freethreading-python-howto" target="_blank" >docs.python.org</a><br/> <strong>Sort of</strong>: <a href="https://labs.quansight.org/blog/free-threaded-one-year-recap?featured_on=talkpython" target="_blank" >labs.quansight.org</a><br/> <strong>did a whole course</strong>: <a href="https://training.talkpython.fm/courses/python-concurrency-deep-dive" target="_blank" >training.talkpython.fm</a><br/> <strong>uvloop</strong>: <a href="https://github.com/magicstack/uvloop?featured_on=talkpython" target="_blank" >github.com</a><br/> <strong>rloop</strong>: <a href="https://github.com/gi0baro/rloop?featured_on=talkpython" target="_blank" >github.com</a><br/> <strong>TonIO</strong>: <a href="https://github.com/gi0baro/tonio?featured_on=talkpython" target="_blank" >github.com</a><br/> <strong>your EuroPython 2026 talk</strong>: <a href="https://www.youtube.com/watch?v=3GwyadhJZBQ&list=PLQHk3YPV3Dgk&index=4&t=753s" target="_blank" >www.youtube.com</a><br/> <strong>Michael's Cutting Python Web App Memory Over 31% Article</strong>: <a href="https://mkennedy.codes/posts/cutting-python-web-app-memory-over-31-percent/?featured_on=talkpython" target="_blank" >mkennedy.codes</a><br/> <br/> <strong>Watch this episode on YouTube</strong>: <a href="https://www.youtube.com/watch?v=lmG0ocDTsZo" target="_blank" >youtube.com</a><br/> <strong>Episode #561 deep-dive</strong>: <a href="https://talkpython.fm/episodes/show/561/tonio-a-multi-threaded-async-runtime-for-python#takeaways-anchor" target="_blank" >talkpython.fm/561</a><br/> <strong>Episode transcripts</strong>: <a href="https://talkpython.fm/episodes/transcript/561/tonio-a-multi-threaded-async-runtime-for-python" target="_blank" >talkpython.fm</a><br/> <br/> <strong>Theme Song: Developer Rap</strong><br/> <strong>🥁 Served in a Flask 🎸</strong>: <a href="https://talkpython.fm/flasksong" target="_blank" >talkpython.fm/flasksong</a><br/> <br/> <strong>---== Don't be a stranger ==---</strong><br/> <strong>YouTube</strong>: <a href="https://talkpython.fm/youtube" target="_blank" ><i class="fa-brands fa-youtube"></i> youtube.com/@talkpython</a><br/> <br/> <strong>Bluesky</strong>: <a href="https://bsky.app/profile/talkpython.fm" target="_blank" >@talkpython.fm</a><br/> <strong>Mastodon</strong>: <a href="https://fosstodon.org/web/@talkpython" target="_blank" ><i class="fa-brands fa-mastodon"></i> @talkpython@fosstodon.org</a><br/> <strong>X.com</strong>: <a href="https://x.com/talkpython" target="_blank" ><i class="fa-brands fa-twitter"></i> @talkpython</a><br/> <br/> <strong>Michael on Bluesky</strong>: <a href="https://bsky.app/profile/mkennedy.codes?featured_on=talkpython" target="_blank" >@mkennedy.codes</a><br/> <strong>Michael on Mastodon</strong>: <a href="https://fosstodon.org/web/@mkennedy" target="_blank" ><i class="fa-brands fa-mastodon"></i> @mkennedy@fosstodon.org</a><br/> <strong>Michael on X.com</strong>: <a href="https://x.com/mkennedy?featured_on=talkpython" target="_blank" ><i class="fa-brands fa-twitter"></i> @mkennedy</a><br/></div>
September 04, 2026 07:33 AM UTC
Trey Hunner
Python Morsels now has spaced repetition
Nearly every book I’ve read on teaching and education over the past decade has talked about the value of spaced repetition. For much of that time, spaced repetition has been something I would recommend learners do, but it wasn’t something I helped anyone do… until now.
Python Morsels now has Daily Recall: a spaced repetition system for Python programmers.
We learn by recalling, not by reading
The most effective learning techniques all rely on active recall: trying to remember something without looking it up, whether with flash cards, by explaining an idea in your own words, or by doing a task that requires it.
We don’t learn by putting information into our heads. We learn by retrieving information from our heads. That’s why Python Morsels has always been built around exercises rather than videos: writing code is the most useful form of recall for a Python programmer.
But not everything worth remembering warrants an entire Python exercise.
- Which
uvcommand runs a tool without installing it? - What is the time complexity of various list operations?
- Which string method can remove a substring from the end of a string?
A 20-minute exercise is overkill for practicing something that small. But practicing it just once isn’t enough either. A quick question, asked again just before you’d forget, is a much better fit.
Spaced repetition beats the forgetting curve
Many of the things I learned in school are long gone, especially the ones I haven’t thought about even once in years. I think I could explain photosynthesis in 9th grade. I can’t today.
This is explained by the forgetting curve: we forget what we don’t recall, and the rate of forgetting is somewhat predictable. Spaced repetition is about using active recall to beat the forgetting curve. Instead of recalling an idea over and over right after learning it, you wait until it has started to fade, and then try to recall it. Each successful recall earns a longer wait before the next one: minutes at first, then hours, then days, and eventually weeks and months.
The tricky part is the timing: when should you try to recall each thing? That’s where an algorithm helps. A spaced repetition system tracks every idea you’re trying to remember and prompts you to recall each one right before you’d forget it.
What Daily Recall does
Back in April, just before Earth Day, I made Whereabouts.Earth to help me learn the name and location of every country in the world. When I started, I could name about 90 of the 197 countries on a map. By mid-June, with about 10 minutes of practice a day, I knew all of them.
Daily Recall is the same idea, but for Python. You pick the packs you want to practice, and each day it asks you a few questions from them. Answer a question correctly and it’ll be a while before you see that one again. Miss it and you’ll see it again soon.
Daily Recall uses FSRS for scheduling, which is one of the most effective spaced repetition algorithms (it’s an option in the Anki flash card app).
If you have trouble remembering which operations on different data structures are fast and which are slow, there’s a Time Complexity pack for that.
If you’re struggling to remember the many different subcommands that uv supports, there’s a uv pack for that.
There are also packs on string methods, built-in functions, dictionaries, iterable unpacking, f-strings, pytest, and what’s new in Python 3.13 and 3.14.
I’m hoping to release about one new pack each week over the next many months.
Daily Recall also works well on a phone because recall questions don’t require typing a bunch of code. So you can replace 5 minutes of your daily doomscrolling with 5 minutes of extra Python learning.
Early users who practiced about 5 minutes a day ended their first month with 25 to 50 new things they could still recall weeks after last seeing them.
You can use Daily Recall for free
Most Daily Recall packs are free, and the rest are included with the All Access plan. To get started, create a free Python Morsels account, pick a pack or two, and answer a few questions.
And whether or not you ever use Daily Recall, I’d recommend spaced repetition. If you’d rather write your own flash cards, on Python or anything else, Anki works well too. The next time you learn something new in Python, don’t just read it a second time. Try to recall it tomorrow, and then again next week.
September 04, 2026 12:30 AM UTC
Bob Belderbos
Why does Alembic need an import you never use?
A coaching session this week surfaced a good question: Alembic autogenerate only works if you import your models first, even though you never reference them.
Ruff flags the import as unused.
So why is it there?
Instead of just adding # noqa: F401, I stopped and followed the call path into SQLModel and SQLAlchemy. The answer is a nice example of Python metaprogramming: importing the module executes the class definition, and the class definition registers its table with SQLAlchemy's metadata.
The unused import that isn't unused
Here's the line every SQLModel + Alembic setup has in env.py:
from tips.models import Tip, User # noqa: F401
target_metadata = SQLModel.metadata
(That's a real line from this project's env.py.)
SQLModel.metadata is a single shared MetaData object. A table only gets registered in its tables collection as a side effect of the class being defined.
If tips.models never gets imported, the class Tip(...) statement never executes, so no table is registered. Alembic then sees empty model metadata when it runs autogenerate. Existing database tables aren't present in that metadata, so Alembic can propose dropping them.
So the import isn't there for the names Tip and User. It's there to run the module so the class definitions fire. That's why # noqa: F401 is justified.
This is a common confusion for people new to Python, importing an object actually reads and executes the module, and the side effects of that execution are often more important than the names it exports. (For a related trap, where when an object gets created changes its behavior, see Two Python Scoping Bugs.)
A better way to write it so we can include any future models without having to list them all is:
import tips.models # noqa: F401
The interesting question is what "the class definition fires" actually means. Nothing in your model class calls metadata.add_table(). So who does?
Following table=True into the source
When you write class Tip(SQLModel, table=True), the table=True is a class keyword argument. It gets routed to SQLModelMetaclass. In sqlmodel 0.0.39, main.py reads it back out and stashes it on the model config:
# sqlmodel/main.py — SQLModelMetaclass.__new__
config_table = get_config("table")
if config_table is True:
new_cls.model_config["table"] = config_table
...
Then __init__ on the same metaclass checks that flag and, only when it's set, hands the class over to SQLAlchemy's declarative machinery:
# sqlmodel/main.py — SQLModelMetaclass.__init__
base_is_table = any(is_table_model_class(base) for base in bases)
if is_table_model_class(cls) and not base_is_table:
... # build columns and relationships from the model's fields
DeclarativeMeta.__init__(cls, classname, bases, dict_, **kw)
else:
ModelMetaclass.__init__(cls, classname, bases, dict_, **kw)
(is_table_model_class is just the check that model_config["table"] is set.)
That if/else is the key fork created by table=True. With it, SQLModel takes the class down SQLAlchemy's declarative path. Without it, you get a SQLModel/Pydantic model rather than a mapped table model, so no SQLAlchemy Table is registered in SQLModel.metadata.
From DeclarativeMeta.__init__ the trail runs straight down into SQLAlchemy:
# sqlalchemy/orm/decl_api.py
if not cls.__dict__.get("__abstract__", False):
_as_declarative(reg, cls, dict_)
_as_declarative scans the class, builds a Table object from the columns, and that Table registers itself:
# sqlalchemy/sql/schema.py — Table.__new__
metadata._add_table(name, schema, table)# sqlalchemy/sql/schema.py — MetaData._add_table
def _add_table(self, name, schema, table):
key = _get_table_key(name, schema)
self.tables._insert_item(key, table)
There it is. self.tables._insert_item(...) is the exact moment your model becomes an entry in SQLModel.metadata.tables, and it runs during class definition, triggered by importing the module. That's the side effect Alembic depends on.
Seeing it happen
You don't have to trust the call path. Registration is a side effect of the class statement, so a plain REPL lets you watch the registry grow in real time:
>>> from sqlmodel import SQLModel, Field
>>> print(SQLModel.metadata.tables)
FacadeDict({})
>>> class Tip(SQLModel, table=True):
... id: int | None = Field(default=None, primary_key=True)
... text: str
...
>>> print(SQLModel.metadata.tables)
FacadeDict({'tip': Table('tip', MetaData(), Column('id', Integer(), table=<tip>, primary_key=True, nullable=False), Column('text', AutoString(), table=<tip>, nullable=False), schema=None)})
No create_engine, no create_all, no import of your models module. Just defining the class populated the shared metadata. Drop the table=True and run it again: the dict stays empty, because now Tip is a SQLModel/Pydantic model rather than a mapped table model.
That proves the class statement does the registering. To tie it back to the opening question, put the same class in a module and let the import fire it:
# models.py
from sqlmodel import SQLModel, Field
class Tip(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
text: str>>> from sqlmodel import SQLModel
>>> print(SQLModel.metadata.tables)
FacadeDict({})
>>> import models # the "unused" import
>>> print(SQLModel.metadata.tables)
FacadeDict({'tip': Table('tip', ...)})
You never touch models.Tip after importing it, yet the registry filled up. That's the exact env.py situation: the import runs the module, the module runs the class definition, and the class definition registers the table.
Is this metaprogramming?
SQLModelMetaclass inherits from both Pydantic's ModelMetaclass and SQLAlchemy's DeclarativeMeta.
This is the same mechanism behind a lot of Python you use daily. If you've ever wondered how a class declaration can acquire behavior you never explicitly wrote, the answer is often a metaclass or __init_subclass__ doing work at definition time.
A metaclass is code that runs when a class is defined. Here it inspects a keyword, processes the model's fields, and delegates table-model construction to SQLAlchemy.
Your field annotations get read at that moment and turned into Column objects.
The practical payoff: next time autogenerate produces an empty migration or wants to drop all your tables, you'll know the cause is registration order, not Alembic being broken. Something imported the models too late, or not at all, and metadata.tables was empty when Alembic read it.
Looking at libraries
Reading a library's source to answer "why is this import here" is a great way to learn the library and Python itself. You don't have to understand every line. Follow the call path of a single feature and you'll often see how the pieces fit together.
It's also about not taking things at face value. A noqa means you're deliberately silencing something a linter flagged. Here it was warranted, but only after understanding why the import exists.
AI can follow this call path for you in seconds. That's useful.
But the valuable skill isn't memorizing that Table.__new__ calls _add_table. It's having the instinct to stop when something looks strange and ask: why is this here?
As AI writes more of the code, this habit matters now more than ever I think. If you don't understand the code, you can't maintain / improve it over time.
September 04, 2026 12:00 AM UTC
Graham Dumpleton
Phased behaviour in wrapture
Most of what a test configures on a patch holds until the test changes it. Retry logic is the classic case where that is not enough: the code under test keeps calling, and the test needs the behaviour to change on its own as it does. Fail twice and then succeed. Hand out a sequence of canned responses. Run the real thing until it breaks and then fail fast. unittest.mock handles the first two of those with a list passed as side_effect, consumed one entry per call. wrapture models the same idea as phases, and this post is about what that buys you beyond the list.
The code under test
A client that fetches a URL, and a function that retries on a timeout:
class Client:
def fetch(self, url):
if "bad" in url:
raise ConnectionError(f"cannot reach {url}")
return {"url": url, "status": 200}
def fetch_with_retry(client, url, attempts=3):
for attempt in range(1, attempts + 1):
try:
return client.fetch(url)
except TimeoutError:
if attempt == attempts:
raise
With mock the retry test is a side_effect list, and it works fine:
with patch.object(Client, "fetch", side_effect=[TimeoutError("busy"), TimeoutError("busy"), {"url": "/x", "status": 200}]):
assert fetch_with_retry(Client(), "/x") == {"url": "/x", "status": 200}
What the list cannot say is "and then run the real code". Every entry is a fabricated outcome, so the third call is a canned dictionary rather than the real fetch(), and the test proves the loop retries but not that the real method is what it eventually reaches.
Phases
In wrapture the behaviour configured on on_call is phase 0, and then() adds the phase that takes over from it, with the argument saying when the hand-over happens. Each phase is a complete behaviour of its own with the full vocabulary, and nothing is inherited between them, so a phase with no terminal runs the real operation:
fetch = wrapture.binding(Client, "fetch")
fetch.on_call.raises(TimeoutError("busy"))
recovered = fetch.on_call.then(after=2)
recovered.passes_through()
The first two calls raise, and every call after that is real. Stating passes_through() on a fresh phase is optional, since that is what an empty phase does anyway, but worth writing when running the real thing is the point of the phase. Recording it shows the hand-over, and the tape marks which outcomes were injected and which were real:
with wrapture.timeline(fetch) as tape:
print(fetch_with_retry(Client(), "/orders"))
print(tape.tree())
{'url': '/orders', 'status': 200}
__main__:Client.fetch(url='/orders') !! TimeoutError (injected)
__main__:Client.fetch(url='/orders') !! TimeoutError (injected)
__main__:Client.fetch(url='/orders') -> {'url': '/orders', 'status': 200}
Each event carries the index of the phase that handled it, so the recording can be filtered by regime, and the binding knows which phase it is in:
fetch.events.in_phase(0).assert_times(2)
fetch.events.in_phase(1).assert_once()
assert fetch.phase == 1
binding.phase is the index of the phase currently active, and in_phase() filters the recorded events to those a given phase handled. The two answer different questions, since a phase can be entered and left without handling a call. Phases restart at 0 on every apply(), so a binding handed to timeline() starts its script afresh in each test that uses it.
The give-up path is the same binding with a bigger count. With then(after=3) all three attempts raise, fetch_with_retry() re-raises the last one, and the tape shows three injected failures and no real call.
The verbs on a phase return the phase, so a phase can be configured in one chain, then(after=1).validates_args(check).returns(b). Holding it in a variable named for what the phase is, and configuring it line by line as with on_call, usually reads better, and it is the style I would use in a test.
Ending a phase on a condition
A count is one of three ways a phase can end. then(until=fn) ends the phase once fn(event) is true for a call it handled. The event is the same one a timeline would record, seen as the caller saw it, so the condition can look at the arguments, the result, or whether the call raised. That is enough to build a circuit breaker: run the real call until one fails, then fail fast without touching the remote at all.
class CircuitOpen(Exception):
pass
def failed(event):
return event.exception is not None
fetch = wrapture.binding(Client, "fetch")
fetch.on_call.passes_through()
tripped = fetch.on_call.then(until=failed)
tripped.raises(CircuitOpen("circuit open"))
Fetch two good URLs, one bad one that the real fetch() rejects, and then another good one:
__main__:Client.fetch(url='/a') -> {'url': '/a', 'status': 200}
__main__:Client.fetch(url='/b') -> {'url': '/b', 'status': 200}
__main__:Client.fetch(url='/bad') !! ConnectionError
__main__:Client.fetch(url='/c') !! CircuitOpen (injected)
The ConnectionError is real, raised by the real method for a real reason, and the CircuitOpen after it is the binding's. A side_effect list has no way to express a phase whose boundary depends on what the real code did.
Sequences
For "return the next value on each call" a phase per value would be tiresome, so returns_from(iterable) is a terminal that draws successive values, one per call, lazily. A generator or itertools.cycle() works. When the sequence runs out the phase ends and the call that found it empty is handled by the successor, so a bare then() after a sequence means "when it is exhausted". A polling loop is the natural example:
class Job:
def status(self):
return "done"
def wait_for(job, polls=5):
for _ in range(polls):
if job.status() == "done":
return True
return False
status = wrapture.binding(Job, "status")
status.on_call.returns_from(["queued", "running", "running"])
settled = status.on_call.then()
settled.returns("done")
__main__:Job.status() -> 'queued' (injected)
__main__:Job.status() -> 'running' (injected)
__main__:Job.status() -> 'running' (injected)
__main__:Job.status() -> 'done' (injected)
This is the closest thing to mock's side_effect list, and the deliberate difference is that values and exceptions are kept apart. side_effect=[a, b, Err] becomes returns_from([a, b]) followed by a phase that raises(Err), which is more lines for the same three outcomes but each phase says what it is. Running out with no successor is a loud SequenceExhaustedError at the call site rather than a StopIteration leaking out of the code under test, and the message says to add a phase with then() or supply an endless sequence.
A known sequence of "random" numbers is another use, making code that jitters or samples deterministic without seeding tricks: binding(random, "random").on_call.returns_from([0.1, 0.9, 0.5]).
Advancing from outside
The third way a phase ends is that something other than this binding's own calls decides it should. A bare then() with no condition ends only when the test calls binding.advance(), which also works whatever the exit condition, so a test can force the next phase early. The simplest use is a test that sits between calls:
remote = wrapture.binding(Client, "fetch")
remote.on_call.raises(ConnectionError("down"))
remote.on_call.then().passes_through()
with remote:
client = Client()
with pytest.raises(ConnectionError):
client.fetch("/x")
remote.advance()
assert client.fetch("/x")["status"] == 200
The more interesting use is when the trigger lives in a different binding. Here the remote stays down until a health check, itself a binding, reports it healthy, and the health check's own result stage advances the remote:
class Monitor:
def check(self):
return "healthy"
remote = wrapture.binding(Client, "fetch")
remote.on_call.raises(ConnectionError("down"))
online = remote.on_call.then()
online.passes_through()
health = wrapture.binding(Monitor, "check")
health.on_call.returns_from(["unhealthy", "unhealthy", "healthy"])
health.on_call.then().returns("healthy")
def note_recovery(result):
if result == "healthy":
remote.advance()
health.on_call.validates_result(note_recovery)
Run code that polls the monitor and tries the client each time round, and the tape shows the two scripts interleaving:
__main__:Monitor.check() -> 'unhealthy' (injected)
__main__:Client.fetch(url='/x') !! ConnectionError (injected)
__main__:Monitor.check() -> 'unhealthy' (injected)
__main__:Client.fetch(url='/x') !! ConnectionError (injected)
__main__:Monitor.check() -> 'healthy' (injected)
__main__:Client.fetch(url='/x') -> {'url': '/x', 'status': 200}
Note that a stage such as validates_result() belongs to the phase it was configured on, which follows from phases inheriting nothing from each other. That is why "healthy" is the last value of the phase 0 sequence above rather than the value the successor phase returns; if the stage were on phase 0 and the triggering value only ever came from phase 1, the recovery would never be noticed. A stage that should run in every phase is configured in every phase. When the condition is visible in the binding's own calls, then(until=...) says it more directly than a stage calling advance(), and is the form to reach for first.
Where phases fit in a test
Phases are for behaviour that must change within one call of the code under test, as it happens with a retry loop, a breaker, or a polling wait. A test that sits between calls does not need them; it reconfigures the binding in place, on_call.returns(...) again, and carries on. That is why the decorator form deliberately leaves then() out of its chain: how behaviour changes over time is the test's script, and it is configured in the body through the injected handle, where the phase markers can be given names.
The attribute channels have phases too, on_get in particular has returns_from(), so a module constant can read one way for two reads and then another, which I will come back to in the next post. And passes_through() on a base namespace clears phase 0 only; to drop the whole chain and start again, on_call.reset() is the tool.
What's next
Everything in this series so far has been about calls. The next post is about everything a binding can name that is not a call: attribute reads and writes, a value held in a slot for the duration of a test, the whole content of a settings dict, and what happens item by item as a generator is consumed.
September 04, 2026 12:00 AM UTC
September 03, 2026
Jaime Buelta
The Many Challenges in Integrating Information for AI Agents
Recently I’ve been thinking quite a lot about information availability for agents, and the fact that this is a very difficult and potentially irresoluble problem. Let me try to explain myself. I talked before about a mental model on differentiating between the LLM models and the tools that access those models. I think that now that’s clearer as we are using more and more agents. We understand that we can use Claude Code with different models (like Sonnet or Opus) that change the capacity of the agent, but not its capabilities. The... Read More
September 03, 2026 07:12 AM UTC
Graham Dumpleton
Recording calls with wrapture
In unit testing with wrapture the tests leaned on a timeline and a tape to assert on what happened, and I skipped over what those actually are. This post is about the recording side of wrapture: what gets recorded, what one event holds, how a test reads the record back, and the whole-tape views that answer questions about the flow between calls rather than about any one of them.
The example is a resource leak, because it is the kind of bug the recording model was made for. Code that acquires a connection has to release it on every path out: the normal return, the early return, and the exception. The path that forgets is the one nobody looks at, and it does not fail. Nothing raises, nothing returns the wrong value, the test passes, and the pool runs dry a week later in production. The failure is an absence, and asserting on an absence needs a record of what did happen, on the real objects, including objects minted mid-call that the test never held.
The code under test
A stand-in for any pooled resource. Database.connect() mints a Connection, and a connection answers queries until close() sets its closed flag:
class Connection:
def __init__(self, number):
self.number = number
self.closed = False
def execute(self, sql):
if self.closed:
raise RuntimeError("connection is closed")
return [(1, "widget")] if "id = 1" in sql else []
def close(self):
self.closed = True
def __repr__(self):
return f"<Connection {self.number}>"
class Database:
def __init__(self):
self.issued = 0
def connect(self):
self.issued += 1
return Connection(self.issued)
The repository is where the bug lives. count() releases in a finally, so it is safe on every path. find() releases only when a row was found; the not-found early return leaks its connection:
class Repository:
def __init__(self, database):
self.database = database
def count(self, table):
connection = self.database.connect()
try:
return len(connection.execute(f"SELECT * FROM {table}"))
finally:
connection.close()
def find(self, table, key):
connection = self.database.connect()
rows = connection.execute(f"SELECT * FROM {table} WHERE id = {key}")
if not rows:
return None
connection.close()
return rows[0]
def report(repository, keys):
found = [repository.find("products", key) for key in keys]
return repository.count("products"), [row for row in found if row]
Running report(Repository(Database()), [1, 2]) returns (0, [(1, 'widget')]), which is correct. Nothing about that result says a connection was left open.
The usual way to test this is a hand-written fake Database whose connect() appends to a list, with connections that flip a flag, and a test that walks the list. It works, but it tests a substitute. The real classes never run, the fake has to be kept in step with them, and every acquiring class in the codebase needs its own. The record you want is of the real calls.
The timeline and the tape
Bind connect on Database and close on Connection, and record both onto one tape. Neither binding has any behaviour configured, so they observe and nothing else:
connect = wrapture.binding(Database, "connect")
close = wrapture.binding(Connection, "close")
with wrapture.timeline(connect, close) as tape:
report(Repository(Database()), [1, 2])
print(tape.tree())
__main__:Database.connect() -> <Connection 1>
__main__:Connection.close() -> None
__main__:Database.connect() -> <Connection 2>
__main__:Database.connect() -> <Connection 3>
__main__:Connection.close() -> None
Three acquisitions, two releases, and reading down the tape you can already see which one has no partner.
The two words are two views of one thing. The timeline is the scope: with wrapture.timeline(...) opens it, the bindings handed to it are applied on entry and removed on exit, and while it is open every call through every applied binding records an event. The tape is what the scope holds. Bindings applied by other means, a fixture or an outer with, record onto an open tape as well, and a binding applied with no timeline open records nothing and costs almost nothing beyond wrapt's own dispatch, so leaving bindings applied and only occasionally recording is a supported pattern rather than a mistake.
Notice that close is bound on the Connection class, not on any connection object. The connections do not exist when the test starts; connect() mints them mid-call. A binding on the class wraps the method for every instance, present and future, which is exactly what covers objects a factory hands out. A mock injected through a seam cannot see those objects at all.
What one event holds
Each call through a binding inside the scope records one event, and an event is a good deal richer than a mock's call record. The fields a test typically reads are path, the fully qualified location in module:qualname form; instance, the object the method was called on; arguments, the call normalised against the real signature with defaults applied, so charge(500) and charge(amount=500) record identically; result, the real return value, or exception when the call raised instead; and seq, parent_id and depth, which place the event in the call tree. There are timings too, started and duration, with recording's own bookkeeping excluded from the figure.
Because the values are real, they can be compared across events. A connect event's result is the connection it minted, and a close event's instance is the connection it was called on, so the leaked connections are the difference between the two sets:
with wrapture.timeline(connect, close):
report(Repository(Database()), [1, 2])
acquired = {event.result for event in connect.events}
released = {event.instance for event in close.events}
print(acquired - released)
{<Connection 2>}
That is the whole question answered, and it needed nothing from the repository. Events record what actually flowed, behaviour included: a call stubbed with returns() records the stubbed result, a failure injected with raises() records that exception, and when transforms_args() rewrote the arguments the event keeps both the arguments as the caller sent them and the ones the real method received, which no substitution-based tool can record because replacing a function discards what it would have been called with.
Filters narrow, assertions conclude
A binding's events property is a filterable view over the tape for that one binding, and it works inside the with block after the code under test has run. One naming rule holds across the whole package: a method whose name starts with assert_ raises immediately, one starting with expect_ declares and is checked when the scope closes, and everything else returns data. A mistyped assertion name is therefore an AttributeError rather than the silent pass mock's assert_calld_once was famous for.
Filters chain and never raise. with_args(amount=500) keeps calls whose normalised arguments include the given values, with_instance(obj) keeps calls made on exactly that object by identity, raising(TimeoutError) keeps calls that raised, returning(value) keeps calls that returned it, and matching(predicate) is the escape hatch. Assertions then conclude: assert_never(), assert_once(), assert_times(n), assert_at_least(n) and assert_at_most(n). Each returns the log on success so a passing assertion can keep chaining, and each prints the events it looked at on failure. Asserting three closes when there were two gives:
AssertionError: expected exactly 3 event(s), got 2
<EventLog __main__:Connection.close: 2 event(s)>
__main__:Connection.close()
__main__:Connection.close()
An assertion is written where it runs. An expectation is the same claim declared on the binding up front, before the run, and verified when the timeline exits:
close = wrapture.binding(Connection, "close").expect_times(3)
with wrapture.timeline(connect, close):
report(Repository(Database()), [1, 2])
ExpectationNotMetError: declared expectation on __main__:Connection.close not met: expected exactly 3 event(s), got 2
<EventLog __main__:Connection.close: 2 event(s)>
__main__:Connection.close()
__main__:Connection.close()
ExpectationNotMetError derives from AssertionError, so test frameworks report it as a failure. Expectations read as a contract at the top of the test with the body free of bookkeeping, and an expectation with nothing recording is an error rather than a pass. Verification is skipped when the block itself raised, since the in-flight failure is the real cause and a verification error on top would bury it.
The tree names the culprit
Counting says something leaked, and pairing says what. To say who, add the repository methods to the timeline. The tape then nests each acquire and release under the method that made it, and tape.children_of() walks the tree, so a root whose children include a connect but no close names itself:
find = wrapture.binding(Repository, "find")
count = wrapture.binding(Repository, "count")
with wrapture.timeline(find, count, connect, close) as tape:
report(Repository(Database()), [1, 2])
print(tape.tree())
for caller in tape.roots():
paths = [child.path for child in tape.children_of(caller)]
if "__main__:Connection.close" not in paths:
print("leaked by", caller)
__main__:Repository.find(table='products', key=1) -> (1, 'widget')
__main__:Database.connect() -> <Connection 1>
__main__:Connection.close() -> None
__main__:Repository.find(table='products', key=2) -> None
__main__:Database.connect() -> <Connection 2>
__main__:Repository.count(table='products') -> 0
__main__:Database.connect() -> <Connection 3>
__main__:Connection.close() -> None
leaked by __main__:Repository.find(table='products', key=2)
The tree shows the bug as it happened. find() with a key that matched released its connection, find() with a key that did not match never called close(), and count() released on the way out of its finally.
When the method is long, or acquires in several places, you want the line rather than the method. Stack capture on the acquire binding records the calling frame with each event, priced per binding so only the acquire pays for it:
connect = wrapture.binding(Database, "connect", stack="caller")
with wrapture.timeline(connect, close):
report(Repository(Database()), [1, 2])
released = {event.instance for event in close.events}
for event in connect.events:
if event.result not in released:
frame = wrapture.stack_frames(event.stack)[0]
print(f"{event.result} acquired at line {frame.lineno} in {frame.function}, never released")
<Connection 2> acquired at line 40 in Repository.find, never released
Order across bindings
Per-binding logs answer questions about one call site; the tape answers questions about the flow between them. tape.assert_order(connect, close) is a subsequence check across any bindings: other events may appear before, between and after, and only the relative order of the named bindings' events matters. A step can also be a filtered log, which is how to say which call, so tape.assert_order(charge.events.raising(TimeoutError), refund) reads as "the refund came after the charge that timed out". consecutive=True requires the steps to match a consecutive run with nothing of those bindings' in between, and exact=True requires those bindings' events to be exactly the steps, which are mock's assert_has_calls and mock_calls == respectively, except that they work across bindings instead of within one mock.
On failure the message names where the walk stalled and prints the actual timeline, which reads far better than a list diff. Asserting a close before a connect on a run that only leaked:
AssertionError: expected order not satisfied; stalled waiting for __main__:Database.connect (position 2 of 2)
actual timeline:
__main__:Repository.find(table='products', key=2)
__main__:Database.connect()
__main__:Repository.count(table='products')
__main__:Database.connect()
__main__:Connection.close()
Scoping instead of resetting
A tape is never cleared. Where a mock suite reaches for reset_mock() to discard setup calls before the act step, wrapture opens the timeline around the part that counts. Timelines nest, and an inner timeline() with no arguments records only what happens inside it while the outer one keeps the whole run:
with wrapture.timeline(connect, close) as whole:
repository = Repository(Database())
repository.count("products") # lands on `whole` only
with wrapture.timeline() as act:
repository.find("products", 1)
connect.events.assert_once() # the act step alone
Inside the inner block connect.events reads the innermost tape, so the count is one even though the outer tape holds four events. The same scoping is how a phased test keeps each phase's counts separate, one timeline per phase, with the same bindings applied on entry and removed on exit each time. The second phase can then state assert_never() outright, where one cumulative tape could only say the count is still one.
Messages and phases as events
Calls are not the only thing that records. An attribute binding records reads and writes of an attribute as get and set events on the same tape, which for this example means the closed flag can be watched directly rather than inferred from close() being called. That is a subject for a later post. Two other event producers are worth knowing about now, because they change what a test can pin an assertion to.
The first is log capture. capture_logs() records standard library logging onto the tape as events of kind "log", selected by logger name pattern and level, and it applies like a binding so timeline() accepts it alongside them. Give the repository a warning when nothing is found, and the message lands inside the call that logged it:
logs = wrapture.capture_logs("myapp.*")
with wrapture.timeline(find, connect, close, logs) as tape:
report(Repository(Database()), [1, 2])
print(tape.tree())
warning = logs.events.at_level("WARNING").with_message("*no row*").assert_once().first
assert tape.parent_of(warning) is find.events.with_args(key=2).first
__main__:Repository.find(table='products', key=1) -> (1, 'widget')
__main__:Database.connect() -> <Connection 1>
__main__:Connection.close() -> None
__main__:Repository.find(table='products', key=2) -> None
__main__:Database.connect() -> <Connection 2>
log myapp.repo WARNING 'no row in products with id 2'
__main__:Database.connect() -> <Connection 3>
__main__:Connection.close() -> None
That last assertion is the one pytest's caplog has no words for: the warning was logged by this call, not merely somewhere during the test. Capture sits at Logger.handle, so it hears each record once on the logger that emitted it, before propagation and regardless of handler configuration, and nothing the application configured is touched.
The second is a block. wrapture.block(name) is a context manager the code, or the test, uses to declare a stretch of code as one event, with everything recorded inside it nested underneath. In a test body it names the phases of an integration test so that "the events during the second request" stops being an exercise in parent-chasing:
with wrapture.timeline(connect, close) as tape:
repository = Repository(Database())
with wrapture.block("lookups"):
repository.find("products", 1)
repository.find("products", 2)
with wrapture.block("summary"):
repository.count("products")
lookups = tape.blocks("lookups").assert_once().first
tape.within(lookups).for_binding(close).assert_once()
block: lookups
__main__:Database.connect() -> <Connection 1>
__main__:Connection.close() -> None
__main__:Database.connect() -> <Connection 2>
block: summary
__main__:Database.connect() -> <Connection 3>
__main__:Connection.close() -> None
tape.within(event) scopes the whole query surface to one block's contents, so an ordering assertion on the view never sees an event outside it. In application code the same marker is inert when nothing is listening, so it can stay in production code permanently, which is what makes the same block a span when the events are going to a tracing backend rather than a test.
As a pytest test
In a test the pairing becomes the assertion, and the failure message carries the leaked connections and where each was acquired. close is given a declared expectation of at least one call, so a path that acquires nothing at all cannot pass by accident:
def test_find_releases_its_connection():
connect = wrapture.binding(Database, "connect", stack="caller")
close = wrapture.binding(Connection, "close").expect_at_least(1)
with wrapture.timeline(connect, close):
report(Repository(Database()), [1, 2])
released = {event.instance for event in close.events}
leaked = [
(event.result, wrapture.stack_frames(event.stack)[0])
for event in connect.events
if event.result not in released
]
assert not leaked, f"connections left open: {leaked}"
The test fails today, naming <Connection 2> and the frame inside find(). Fix the early return with a finally and it passes. With the pytest plugin enabled the tape's tree is attached to the failure report as well, so the output shows what ran rather than only the assertion that tripped.
What's next
Everything in this post recorded real calls with the bindings doing nothing but watch. The next post is about the other direction, changing what a call does, and specifically about behaviour that changes over time as the code under test keeps calling, which is what retry logic and circuit breakers need from a test.
September 03, 2026 12:00 AM UTC
September 02, 2026
Django Weblog
Django bugfix release issued: 6.1.1
Today we've issued the 6.1.1 bugfix release.
The release package and checksums are available from our downloads page, as well as from the Python Package Index.
The PGP key ID used for this release is Jacob Walls: 131403F4D16D8DC7


