Modules System Overview

Django-CFG provides a powerful modular system that automatically configures Django services based on your configuration. Modules eliminate boilerplate code and provide production-ready functionality out of the box.

What are Modules?

Modules are auto-configuring services that:

🔧 Auto-discover configuration from your Django-CFG settings
⚡ Zero-setup integration - just import and use
🛡️ Production-ready with error handling and logging
🔌 Extensible - create your own modules easily
📊 Observable - built-in monitoring and health checks

Architecture

Base Module System

All modules inherit from BaseCfgModule:


from django_cfg.modules.base import BaseCfgModule
 
class MyCustomModule(BaseCfgModule):
    def __init__(self):
        super().__init__()
        self.config = self.get_config()  # Auto-discovers Django-CFG config
        
    def my_method(self):
        # Access configuration automatically
        if hasattr(self.config, 'my_service'):
            return self.config.my_service.api_key

Configuration Discovery

Modules automatically find your configuration:


# api/config.py
class MyProjectConfig(DjangoConfig):
    email: EmailConfig = EmailConfig(
        host=env.email.host,
        port=env.email.port
    )
 
# Anywhere in your code
from django_cfg import DjangoEmailService
 
email = DjangoEmailService()  # Automatically uses your EmailConfig
email.send_simple("Test", "Hello!", ["[email protected]"])

Built-in Modules

📧 Communication

Email Module (`DjangoEmailService`)

Auto-configuring email service with multiple backends:


from django_cfg import DjangoEmailService
 
email = DjangoEmailService()
 
# Simple email
email.send_simple(
    subject="Welcome!",
    message="Hello World!",
    recipient_list=["[email protected]"]
)
 
# Template-based email
email.send_template(
    template_name="welcome_email.html",
    context={"user_name": "John"},
    recipient_list=["[email protected]"],
    subject="Welcome to our platform!"
)
 
# Bulk email
email.send_bulk(
    recipients=[
        {"email": "[email protected]", "name": "User 1"},
        {"email": "[email protected]", "name": "User 2"}
    ],
    subject="Newsletter",
    message="Hello!"
)

Features:

✅ Multiple backends (SMTP, etc.)
✅ Template rendering with Django templates
✅ Bulk email support
✅ Attachment handling
✅ HTML/text alternatives
✅ Error handling and retries

Telegram Module (`DjangoTelegram`)

Bot integration and notifications:


from django_cfg import DjangoTelegram, send_telegram_message
 
telegram = DjangoTelegram()
 
# Send notifications
telegram.send_message(
    chat_id="-1001234567890",
    text="🚀 Deployment completed successfully!"
)
 
# Send with formatting
telegram.send_message(
    chat_id="@my_channel",
    text="*Error Alert*\n`Database connection failed`",
    parse_mode="Markdown"
)
 
# Send files
telegram.send_document(
    chat_id="123456789",
    document=open("report.pdf", "rb"),
    caption="Monthly report"
)

Features: Message sending and formatting, file and media uploads, bot command handling, webhook integration, channel/group management, and error notifications.

⚙️ Operations

Health Check Module

Django-CFG provides health check views:


# In your urls.py
from django_cfg.modules.django_health import HealthCheckView, SimpleHealthView
 
urlpatterns = [
    path('health/', HealthCheckView.as_view(), name='health-check'),
    path('health/simple/', SimpleHealthView.as_view(), name='health-simple'),
]

Built-in Checks:

✅ Database connectivity
✅ Cache availability (Redis/Memcached)
✅ Disk space and memory
✅ Returns JSON health status

Logger Module (`DjangoLogger`)

Enhanced logging with structured output:


from django_cfg import DjangoLogger, get_logger
 
logger = get_logger("my_app")
 
# Structured logging
logger.info("User login", extra={
    "user_id": 123,
    "ip_address": "192.168.1.1",
    "user_agent": "Chrome/91.0"
})
 
# Performance monitoring
with logger.timer("database_query"):
    results = MyModel.objects.filter(active=True)

Features:

✅ Structured JSON logging
✅ Automatic request correlation
✅ Performance timing
✅ Error aggregation
✅ Multiple output formats
✅ Log rotation and archival

Tasks Module (`DjangoTasks`)

Background task processing with Django-RQ:


from django_cfg.modules.django_tasks import DjangoTasks
 
tasks = DjangoTasks()
 
# Simple task
@tasks.task
def send_welcome_email(user_id):
    user = User.objects.get(id=user_id)
    return f"Email sent to {user.email}"
 
# Schedule task
task_id = tasks.enqueue(send_welcome_email, user_id=123)
 
# Periodic task
tasks.schedule_periodic(
    "cleanup_old_files",
    cron="0 2 * * *",  # Daily at 2 AM
    func=cleanup_old_files
)

Features: Async task execution, scheduled/delayed tasks, periodic/cron tasks, task monitoring and retries, priority queues, and dead letter queues.

Django-RQ Task Queue Module

Type-safe task scheduling with Django-RQ for background job processing:


from django_cfg import DjangoRQConfig, RQQueueConfig, RQScheduleConfig
 
# In config.py
django_rq = DjangoRQConfig(
    enabled=True,
    redis_db=0,  # Required: unique per project (0-15)
    queues=[
        RQQueueConfig(queue="default"),
        RQQueueConfig(queue="high", default_timeout=3600),
    ],
    schedules=[
        # Sync data every 5 minutes
        RQScheduleConfig(
            func="myapp.tasks.sync_data",
            interval=300,  # seconds
            queue="default",
        ),
        # Daily cleanup at 2 AM
        RQScheduleConfig(
            func="myapp.tasks.cleanup_old_data",
            cron="0 2 * * *",
            queue="default",
        ),
    ],
)

Features: High performance (10K+ jobs/sec), cron & interval scheduling, built-in monitoring (Admin, REST API, Prometheus), job dependencies, retry logic, type-safe configuration, and production-ready error handling.

Learn more: Django-RQ Integration Documentation

🔧 Development Tools

Ngrok Module (`DjangoNgrok`)

Development tunneling for webhooks:


from django_cfg.modules.django_ngrok import DjangoNgrok
 
ngrok = DjangoNgrok()
 
# Start tunnel for development
tunnel_url = ngrok.start_tunnel(port=8000)
print(f"Public URL: {tunnel_url}")
 
# Configure webhooks automatically
ngrok.setup_webhook_urls({
    "stripe": "/webhooks/stripe/",
    "github": "/webhooks/github/"
})

Features: Automatic tunnel creation, webhook URL management, SSL certificate handling, custom domain support, tunnel monitoring, and development/production switching.

🚀 Advanced Modules

Currency Module (`CurrencyConverter`)

Multi-currency support with real-time rates and crypto:


from django_cfg.modules.django_currency import convert_currency, CurrencyConverter
 
# Quick conversions
usd_amount = convert_currency(100, "EUR", "USD")
btc_amount = convert_currency(50000, "USD", "BTC")
 
# Detailed conversion
converter = CurrencyConverter()
result = converter.convert(100, "USD", "EUR")
print(f"Rate: {result.rate.rate}, Source: {result.rate.source}")

LLM Module (`LLMClient`)

AI/LLM integration with multiple providers:


from django_cfg.modules.django_llm.llm.client import LLMClient
 
client = LLMClient()
 
# Chat completion
response = client.chat_completion(
    messages=[
        {"role": "system", "content": "You are a helpful Django expert"},
        {"role": "user", "content": "How do I optimize Django queries?"}
    ],
    model="gpt-4"
)

Note: See LLM Module Documentation for complete API reference.

🔌 Creating Custom Modules

Basic Custom Module


from django_cfg.modules.base import BaseCfgModule
from typing import Optional
 
class MyCustomService(BaseCfgModule):
    """Custom service module."""
    
    def __init__(self):
        super().__init__()
        self.config = self.get_config()
        self.my_config = getattr(self.config, 'my_service', None)
    
    def do_something(self) -> str:
        """Perform custom operation."""
        if not self.my_config:
            raise ValueError("MyService not configured")
        
        # Use configuration
        api_key = self.my_config.api_key
        base_url = self.my_config.base_url
        
        # Your service logic here
        return "Operation completed"
    
    def health_check(self) -> dict:
        """Health check for monitoring."""
        try:
            # Test your service
            result = self.do_something()
            return {
                "status": "healthy",
                "details": "Service is operational"
            }
        except Exception as e:
            return {
                "status": "unhealthy",
                "details": str(e)
            }

Configuration Integration


# api/config.py
from pydantic import BaseModel
 
class MyServiceConfig(BaseModel):
    api_key: str
    base_url: str = "https://api.myservice.com"
    timeout: int = 30
    retries: int = 3
 
class MyProjectConfig(DjangoConfig):
    my_service: MyServiceConfig = MyServiceConfig(
        api_key=env.my_service.api_key
    )

Best Practices

1. Use Type Hints


from typing import List, Dict, Optional
 
class MyService(BaseCfgModule):
    def process_items(self, items: List[Dict[str, str]]) -> Optional[str]:
        pass

2. Handle Configuration Gracefully


def __init__(self):
    super().__init__()
    self.config = self.get_config()
    
    # Graceful degradation
    if not self.config or not hasattr(self.config, 'my_service'):
        self.enabled = False
        return
    
    self.enabled = True
    self.service_config = self.config.my_service

3. Implement Health Checks


def health_check(self) -> dict:
    if not self.enabled:
        return {"status": "disabled", "details": "Service not configured"}
    
    try:
        # Test service connectivity
        self._test_connection()
        return {"status": "healthy"}
    except Exception as e:
        return {"status": "unhealthy", "details": str(e)}

4. Add Monitoring and Metrics


import time
from collections import defaultdict
 
class MonitoredService(BaseCfgModule):
    def __init__(self):
        super().__init__()
        self.metrics = defaultdict(int)
        self.response_times = []
    
    def tracked_operation(self, data):
        start_time = time.time()
        
        try:
            result = self._process(data)
            self.metrics["success"] += 1
            return result
        except Exception as e:
            self.metrics["error"] += 1
            raise
        finally:
            duration = time.time() - start_time
            self.response_times.append(duration)
            self.metrics["total_requests"] += 1

Integration with Django-CFG

Automatic Discovery

Modules automatically discover your Django-CFG configuration:


# No manual configuration needed!
email = DjangoEmailService()  # Finds EmailConfig automatically
health = DjangoHealthService()  # Finds all health-related configs
telegram = DjangoTelegramService()  # Finds TelegramConfig automatically

Configuration Validation


# api/config.py
class MyProjectConfig(DjangoConfig):
    email: EmailConfig = EmailConfig(...)
    
    def validate_modules(self):
        """Custom validation for modules."""
        # Ensure email is configured for production
        if self.is_prod and not self.email.host:
            raise ValueError("Email host required for production")
 
        # Validate Telegram config
        if hasattr(self, 'telegram') and self.telegram:
            if not self.telegram.bot_token:
                raise ValueError("Telegram bot token is required")

All Modules

Core Infrastructure

Module	Description	Optional
Cloudflare D1	Cloudflare D1 integration, auto user sync	Yes (cloudflare)
Monitor	Application performance monitoring, D1 events	Yes (cloudflare)
Logging	Structured logging with D1 persistence	No
Health Checks	Auto-configuring health check endpoints	No

Communication & Real-time

Module	Description	Optional
gRPC Server	Async gRPC with streaming, auth, D1 persistence	Yes (grpcio)
Centrifugo	Real-time WebSocket pub/sub via Centrifugo	Yes (cent)
Django-RQ	Redis Queue task processing and scheduling	Yes (rq)
Email	Auto-configuring email service	No
Telegram	Telegram bot notifications	Yes (httpx)

Admin & UI

Module	Description	Optional
Django Admin	Type-safe admin with 2234 Material icons	No
Unfold Theme	Modern admin interface theme	Yes (django-unfold)
DRF Theme	Tailwind theme for DRF Browsable API	No

Code Generation

Module	Description	Optional
Code Generation	Unified client generation (TS, Python, etc.)	No
FastAPI ORM	Generate FastAPI/SQLModel from Django models	No
OG Images	OG image generation with PicTex/Pillow	No

AI & LLM

Module	Description	Optional
LLM Integration	Multi-provider LLM client with caching	Yes
LLM Balance Monitoring	Provider balance monitoring with alerts	Yes

Data & Tools

Module	Description	Optional
Currency	Multi-currency conversion with MoneyField	No
Import/Export	Excel import/export integration	Yes (django-import-export)
Encryption	API payload encryption (AES)	No
File Cleanup	Auto file cleanup on model delete	No
Tailwind CSS	Tailwind layout templates and utilities	No

Networking

Module	Description	Optional
Ngrok Tunnels	Development tunneling for webhooks	Yes (pyngrok)