python
37 lines · 2 tabs
Priya Sharma
Jan 2026
2 tabs
from celery import shared_task
from django.core.mail import send_mail
from django.contrib.auth import get_user_model
User = get_user_model()
@shared_task(bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3}, retry_backoff=True)
def send_welcome_email_task(self, user_id):
"""Send welcome email asynchronously."""
try:
user = User.objects.get(id=user_id)
send_mail(
subject='Welcome!',
message=f'Hi {user.first_name}, thanks for joining!',
from_email='noreply@example.com',
recipient_list=[user.email],
)
except User.DoesNotExist:
# Don't retry if user was deleted
return
from django.contrib.auth import get_user_model
from rest_framework import generics
from .serializers import UserSerializer
from core.tasks import send_welcome_email_task
User = get_user_model()
class UserCreateView(generics.CreateAPIView):
queryset = User.objects.all()
serializer_class = UserSerializer
def perform_create(self, serializer):
user = serializer.save()
# Queue email task - returns immediately
send_welcome_email_task.delay(user.id)
2 files · python
Explain with highlit
I use Celery for any operation that might be slow or fail intermittently, like sending emails. By decorating with @shared_task, I make tasks reusable across different apps. I set bind=True to access task instance (useful for retries), and configure retry logic with autoretry_for and exponential backoff. The countdown parameter delays execution. I avoid passing complex objects to tasks—use IDs and fetch from DB instead. This prevents serialization issues and ensures fresh data. For critical tasks, I add monitoring and alerting on failures.
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
typescript
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
typescript
reliability
retry
by codesnips
2 tabs
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
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
python
from django.urls import path
from . import views
app_name = 'blog'
urlpatterns = [Django URL namespacing and reverse lookups
django
python
urls
by Priya Sharma
3 tabs
typescript
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
node
concurrency
async
by codesnips
2 tabs
Share this code
Here's the card — post it anywhere.
