import functools
import threading
def debounce(wait):
def decorator(func):
lock = threading.Lock()
state = {"timer": None}
def call(args, kwargs):
with lock:
state["timer"] = None
func(*args, **kwargs)
@functools.wraps(func)
def debounced(*args, **kwargs):
with lock:
if state["timer"] is not None:
state["timer"].cancel()
timer = threading.Timer(wait, call, args=(args, kwargs))
timer.daemon = True
state["timer"] = timer
timer.start()
def cancel():
with lock:
if state["timer"] is not None:
state["timer"].cancel()
state["timer"] = None
def flush(*args, **kwargs):
cancel()
func(*args, **kwargs)
debounced.cancel = cancel
debounced.flush = flush
return debounced
return decorator
import logging
from debounce import debounce
log = logging.getLogger("search")
class SearchController:
def __init__(self, api_client):
self.api = api_client
self.last_results = []
@debounce(0.3)
def on_keypress(self, query):
query = query.strip()
if not query:
return
log.info("querying backend for %r", query)
self.last_results = self.api.search(query)
def on_submit(self, query):
# user hit Enter: skip the wait and run right now
self.on_keypress.flush(self, query)
def on_blur(self):
# field lost focus with nothing pending we care about
self.on_keypress.cancel()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
controller = SearchController(api_client=FakeApi())
for chunk in ["l", "la", "lap", "lapt", "lapto", "laptop"]:
controller.on_keypress(controller, chunk)
controller.on_submit(controller, "laptop")
Debouncing collapses a burst of rapid calls into a single execution that runs only after the caller goes quiet for some interval. It is the standard tool for taming noisy triggers: a search box that fires on every keystroke, a file watcher that emits many events per save, or a resize handler that would otherwise flood a backend with requests. The idea is simple but the implementation has sharp edges around timing and thread safety, which is why a reusable decorator is worth building once and reusing everywhere.
The debounce decorator file implements the core pattern with threading.Timer. Each decorated function keeps a small piece of per-function state guarded by a threading.Lock. On every call, debounced cancels any pending Timer and schedules a fresh one for wait seconds later. Only when calls stop for the full wait window does the timer fire and actually invoke the wrapped function. This is trailing-edge debounce: the last set of arguments wins, which is exactly what a search-as-you-type feature needs. The lock matters because Timer callbacks run on separate threads, so cancelling and reassigning self._timer without synchronization would race.
Two helper methods are exposed on the wrapper via functools.wraps-preserved attributes: cancel() drops a pending call entirely, and flush() runs it immediately. These are important for shutdown paths and tests, where waiting on real wall-clock timers is undesirable. Note the pitfall the code sidesteps: return values are effectively discarded, since the real call happens later on a timer thread — debounce is for side effects, not for functions whose result the caller awaits inline.
The search box usage file shows the decorator in a realistic setting. SearchController.on_keypress is debounced at 0.3 seconds, so typing "laptop" issues one query instead of six. The flush and cancel hooks are wired to explicit submit and blur events. A developer reaches for this whenever an event source is faster than the work it triggers, trading a little latency for far fewer, more meaningful invocations.
Related snips
import os
import stat
for root, _dirs, files in os.walk('/etc'):
for name in files:
path = os.path.join(root, name)Python security audit script for exposed risky filesystem state
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
<form data-controller="query-sync" data-action="change->query-sync#apply">
<select name="status" class="rounded border p-2">
<option value="">Any</option>
<option value="open">Open</option>
<option value="closed">Closed</option>
</select>Filter UI that syncs query params via Stimulus (no front-end router)
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
Share this code
Here's the card — post it anywhere.
