codesnips
python 159 lines · 4 tabs

In-Process Threaded Background Job Queue for Sending Emails Without Redis

Shared by codesnips Jul 2026
4 tabs 
import queue
import time
from dataclasses import dataclass, field, replace
from typing import Any, Dict, Tuple


@dataclass(frozen=True)
class Job:
    task_name: str
    args: Tuple[Any, ...] = ()
    kwargs: Dict[str, Any] = field(default_factory=dict)
    attempts: int = 0
    enqueued_at: float = field(default_factory=time.time)

    def retried(self) -> "Job":
        return replace(self, attempts=self.attempts + 1)


class JobQueue:
    def __init__(self, maxsize: int = 1000):
        self._q: "queue.Queue[Job]" = queue.Queue(maxsize=maxsize)

    def enqueue(self, task_name, *args, **kwargs) -> None:
        job = Job(task_name=task_name, args=args, kwargs=kwargs)
        self._q.put(job)

    def put(self, job: Job) -> None:
        self._q.put(job)

    def get(self, timeout=None) -> Job:
        return self._q.get(timeout=timeout)

    def task_done(self) -> None:
        self._q.task_done()

    def join(self) -> None:
        self._q.join()
4 files · python Explain with highlit

This snippet shows a minimal, dependency-free background job system built on Python's standard library, useful when a project needs deferred work like sending emails but does not want to run Redis, RabbitMQ, or Celery. It trades durability for simplicity: jobs live in an in-memory queue.Queue, so anything unprocessed at shutdown is lost. That trade-off is acceptable for low-stakes side effects (welcome emails, cache warming) on a single process, but not for money-critical work.

The job_queue tab defines the primitives. A Job is a frozen dataclass carrying a callable name, its args/kwargs, and an attempts counter that supports retries. JobQueue wraps a bounded queue.Queue; enqueue builds a Job and calls put, while get and task_done mirror the standard queue API so the worker can cooperate with join(). Bounding the queue provides natural backpressure — if producers outrun consumers, enqueue blocks instead of exhausting memory.

The worker tab is where concurrency lives. Worker spawns a pool of daemon threading.Threads, each running _run, which loops on queue.get(timeout=...). Using a timeout rather than a blocking get lets the loop periodically check _stopping, an threading.Event, so shutdown is responsive. When a task raises, _handle_failure re-enqueues it with an incremented attempts until max_retries is hit, then logs and drops it. stop sets the event and joins the threads, and drain calls the queue's join() so callers can wait for in-flight work — the key to a graceful shutdown that does not strand half-sent batches.

The email_tasks tab registers the actual work. send_welcome_email is an ordinary function looked up by name from a TASK_REGISTRY; keeping tasks name-addressable (rather than pickling closures) keeps the queue serialization-friendly if it is later swapped for a real broker. The app tab wires everything together: it constructs the Worker, exposes enqueue_email, and registers an atexit hook plus a SIGTERM handler so a container stop drains pending jobs before exit. GIL contention makes this best for I/O-bound work like SMTP calls, exactly the email case shown here.

Related snips

python 
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

python auditing host-security
by Kai Nakamura 1 tab
python 
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

django python models
by Priya Sharma 2 tabs
ruby 
require "csv"

class PeopleCsvStream
  include Enumerable

  HEADERS = %w[id full_name email signed_up_at plan].freeze

Resilient CSV Export as a Streamed Response

rails performance streaming
by codesnips 3 tabs
rust 
use crossbeam::channel::unbounded;
use std::thread;

fn main() {
    let (tx, rx) = unbounded();

Crossbeam for advanced concurrent data structures

rust concurrency lock-free
by Marcus Chen 1 tab
javascript 
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
  const success = true;

  setTimeout(() => {
    if (success) {

Promises and async/await patterns for asynchronous JavaScript

javascript promises async-await
by Alex Chang 1 tab
rust 
use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

Channels (mpsc) for message passing between threads

rust concurrency channels
by Marcus Chen 1 tab

Share this code

Here's the card — post it anywhere.

In-Process Threaded Background Job Queue for Sending Emails Without Redis — share card
Link copied