Service Integrations

from django_cfg import TelegramConfig
 
telegram: TelegramConfig | None = (
    TelegramConfig(
        bot_token=env.telegram.bot_token,
        chat_id=env.telegram.chat_id,
    )
    if env.telegram.bot_token and env.telegram.chat_id != 0
    else None
)

TELEGRAM__BOT_TOKEN="123456:ABC-DEF"
TELEGRAM__CHAT_ID=-1001234567890

from django_cfg import DjangoTelegram
 
telegram = DjangoTelegram()
 
# Send simple message (uses chat_id from config)
def send_alert(message):
    """Send alert to Telegram channel."""
    telegram.send_message(message=message)
 
# Send formatted message with parse mode
def send_system_alert(message, level="info"):
    """Send formatted system alert."""
    emoji = {"info": "ℹ️", "warning": "⚠️", "error": "🚨"}
 
    telegram.send_message(
        message=f"{emoji[level]} *System Alert*\n\n{message}",
        parse_mode="Markdown",
    )

from django_cfg import DjangoTelegram
 
telegram = DjangoTelegram()
 
# Notify new order
def notify_new_order(order):
    """Send new order notification."""
    message = f"""
🛒 *New Order Received*
 
Order ID: `{order.id}`
Customer: {order.user.get_full_name()}
Total: ${order.total}
Items: {order.items.count()}
    """
    telegram.send_message(message=message, parse_mode="Markdown")
 
# Notify order status change
def notify_order_status(order, old_status, new_status):
    """Notify when order status changes."""
    message = f"""
📦 *Order Status Updated*
 
Order ID: `{order.id}`
Status: {old_status} → *{new_status}*
Customer: {order.user.email}
    """
    telegram.send_message(message=message, parse_mode="Markdown")

from arq import ArqRedis
from django_cfg import DjangoEmailService
 
async def send_welcome_email_async(ctx, user_id):
    """Send welcome email in background."""
    from django.contrib.auth import get_user_model
 
    User = get_user_model()
    user = User.objects.get(id=user_id)
 
    email = DjangoEmailService()
    email.send_template(
        template_name="emails/welcome.html",
        context={"user_name": user.get_full_name()},
        recipient_list=[user.email],
        subject="Welcome!"
    )
 
# Usage
async def handle_user_registration(user):
    """Handle new user registration."""
    redis = await ArqRedis.create()
    await redis.enqueue_job('send_welcome_email_async', user.id)

from django_cfg import DjangoTelegram
import logging
 
logger = logging.getLogger(__name__)
 
def notify_error(error_message):
    """Send error notification with fallback."""
    try:
        # Try Telegram first
        telegram = DjangoTelegram()
        telegram.send_message(message=f"🚨 Error: {error_message}", fail_silently=True)
    except Exception as e:
        # Fallback to email
        logger.warning(f"Telegram notification failed: {e}")
        try:
            email = DjangoEmailService()
            email.send_simple(
                subject="System Error",
                message=error_message,
                recipient_list=["[email protected]"]
            )
        except Exception as e:
            # Log as last resort
            logger.error(f"All notifications failed: {e}")

from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def send_critical_email(recipient, subject, message):
    """Send email with automatic retries."""
    email = DjangoEmailService()
    return email.send_simple(
        subject=subject,
        message=message,
        recipient_list=[recipient]
    )

# ✅ Good: Async notification
notify_user.send(user.id)  # Non-blocking
 
# ❌ Bad: Synchronous notification
send_notification(user.id)  # Blocks request

# ✅ Good: Use fail_silently for non-critical notifications
telegram.send_message(message=message, fail_silently=True)
 
# ✅ Good: Explicit try/except
try:
    telegram.send_message(message=message)
except Exception as e:
    logger.warning(f"Telegram failed: {e}")
    # Continue without failing
 
# ❌ Bad: Unhandled failures
telegram.send_message(message=message)  # Crashes on error if unconfigured