Automation That Actually Saves Time
Most "automation" articles talk about theory. This one covers real Python scripts I have built for clients — each saving 5-20 hours per week. These are patterns from my business automation services that you can adapt for your own needs.
1. CRM Data Synchronization
A common problem: your sales team updates the CRM, but your Django app has its own customer database. Keeping them in sync manually is a nightmare.
import requests
from celery import shared_task
from django.conf import settings
from customers.models import Customer
@shared_task
def sync_crm_contacts():
headers = {'Authorization': f'Bearer {settings.CRM_API_KEY}'}
response = requests.get(
f'{settings.CRM_URL}/api/contacts',
headers=headers,
params={'updated_since': get_last_sync_time()}
)
contacts = response.json()['data']
created, updated = 0, 0
for contact in contacts:
customer, was_created = Customer.objects.update_or_create(
crm_id=contact['id'],
defaults={
'name': contact['name'],
'email': contact['email'],
'phone': contact.get('phone', ''),
'company': contact.get('company', ''),
'deal_value': contact.get('deal_value', 0),
}
)
if was_created:
created += 1
else:
updated += 1
logger.info("CRM sync: %d created, %d updated", created, updated)
return {'created': created, 'updated': updated}
2. Google Shopping Feed Generation
E-commerce clients need product feeds for Google Shopping. Generating these manually is tedious — automate it:
from django.http import HttpResponse
from products.models import Product
def generate_google_feed(request):
products = Product.objects.filter(
is_active=True, stock__gt=0
).select_related('category')
items = []
for p in products:
items.append(f' <item>'
f'<g:id>{p.sku}</g:id>'
f'<g:title>{p.name}</g:title>'
f'<g:description>{p.description[:5000]}</g:description>'
f'<g:link>https://shop.example.com/products/{p.slug}/</g:link>'
f'<g:price>{p.price} EUR</g:price>'
f'<g:availability>in_stock</g:availability>'
f'<g:brand>{p.brand}</g:brand>'
f'</item>')
xml = ('<?xml version="1.0"?>'
'<rss version="2.0" xmlns:g="http://base.google.com/ns/1.0">'
'<channel><title>Products</title>'
+ ''.join(items) +
'</channel></rss>')
return HttpResponse(xml, content_type='application/xml')
3. Telegram Notification Automation
Real-time notifications for business events — new orders, low stock, system errors:
import json
from urllib.request import urlopen, Request
from django.conf import settings
def send_telegram(message, chat_id=None):
url = f'https://api.telegram.org/bot{settings.TELEGRAM_BOT_TOKEN}/sendMessage'
data = json.dumps({
'chat_id': chat_id or settings.TELEGRAM_CHAT_ID,
'text': message,
'parse_mode': 'HTML',
}).encode()
req = Request(url, data=data, headers={'Content-Type': 'application/json'})
urlopen(req, timeout=5)
# Usage in Django signals
from django.db.models.signals import post_save
from django.dispatch import receiver
@receiver(post_save, sender=Order)
def notify_new_order(sender, instance, created, **kwargs):
if created:
send_telegram(
f"New order #{instance.id}\n"
f"Customer: {instance.customer.name}\n"
f"Total: EUR {instance.total}\n"
f"Items: {instance.items.count()}"
)
4. Celery Task Patterns for Scheduled Jobs
from celery.schedules import crontab
app.conf.beat_schedule = {
'sync-crm-hourly': {
'task': 'tasks.crm_sync.sync_crm_contacts',
'schedule': crontab(minute=0), # Every hour
},
'generate-daily-report': {
'task': 'tasks.reports.daily_summary',
'schedule': crontab(hour=8, minute=0), # 8am daily
},
'check-low-stock': {
'task': 'tasks.inventory.check_stock_levels',
'schedule': crontab(minute='*/30'), # Every 30 minutes
},
'cleanup-old-sessions': {
'task': 'tasks.maintenance.cleanup_sessions',
'schedule': crontab(hour=3, minute=0), # 3am daily
},
}
5. Automated Report Generation
@shared_task
def daily_summary():
today = timezone.now().date()
orders = Order.objects.filter(created_at__date=today)
revenue = orders.aggregate(total=Sum('total'))['total'] or 0
new_customers = Customer.objects.filter(created_at__date=today).count()
message = (
f"Daily Summary for {today}\n\n"
f"Orders: {orders.count()}\n"
f"Revenue: EUR {revenue:.2f}\n"
f"New customers: {new_customers}\n"
)
send_telegram(message)
Monitoring Automated Processes
Automation without monitoring is a time bomb. Set up alerts for failures:
@shared_task(bind=True, max_retries=3)
def reliable_task(self, task_name):
try:
result = run_task(task_name)
TaskLog.objects.create(name=task_name, status='success', result=result)
except Exception as exc:
TaskLog.objects.create(name=task_name, status='failed', error=str(exc))
send_telegram(f"Task {task_name} failed: {exc}")
raise self.retry(exc=exc, countdown=60)
These patterns form the backbone of the automation systems I build. Each one replaces hours of manual work with a few lines of Python running on a schedule. See real examples in my CRM integration project and real estate platform with automated import. Ready to automate your business processes? Let's talk.