Forem: Mehedi Bappi

Understanding Closures in Python

Mehedi Bappi — Sat, 30 Nov 2024 14:36:17 +0000

In Python, closures are an important concept that allows a function to "remember" the environment in which it was created, even after the function has finished executing. Closures allow us to implement stateful functions without using global variables or class instances.

In this post, we'll explore closures by implementing a simple counter using the nonlocal keyword. Let's dive into it!

What is a Closure?

A closure occurs when a nested function refers to a variable from its enclosing scope, allowing it to retain access to those variables even after the enclosing function has finished executing. Closures are particularly useful when you want to encapsulate a state or behaviour within a function.

The `nonlocal` Keyword

In Python, we use the nonlocal keyword to modify a variable in the nearest enclosing scope that is not global. Without the nonlocal keyword, an inner function cannot modify variables in its enclosing scope; it would instead create a new local variable. The nonlocal keyword resolves this by telling Python that we want to work with a variable from the enclosing scope.

Implementing a Counter with Closures

Let's create a simple counter function that uses closures to keep track of the count without relying on global variables or a class.

Step 1: Define the `make_counter` Function

We’ll create a function called make_counter, which will return an inner function increment. The inner function will increase a count variable each time it is called.

Step 2: Use `nonlocal` to Modify the Count Variable

To ensure the increment function modifies the count variable defined in the make_counter function's scope, we'll use the nonlocal keyword.

Here’s the implementation:

def make_counter():
    count = 0  # Variable in the enclosing scope

    def increment():
        nonlocal count  # Tell Python to modify the `count` from the enclosing scope
        count += 1  # Increment the counter
        return count  # Return the current count

    return increment  # Return the inner function, which forms the closure

Step 3: Using the Counter

Now that we have the make_counter function, we can create an instance of the counter and call it multiple times to see the counter increment.

counter = make_counter()

print(counter())  # Output: 1
print(counter())  # Output: 2
print(counter())  # Output: 3
print(counter())  # Output: 4
print(counter())  # Output: 5

Explanation

The count variable is initialized to 0 within the make_counter function.
Every time we call the increment function (which is returned by make_counter), it modifies the count variable in the enclosing scope using the nonlocal keyword. Thanks to the closure property, the increment function retains access to count even after the make_counter function finishes execution.

Why Use Closures?

Closures provide a powerful and elegant way to encapsulate state within functions. They are especially useful in scenarios where:

You want to keep track of a value without exposing it to the global scope.
You need a function to remember its previous state across multiple calls (like our counter-example).
You don’t want to use global variables or create classes for simple state management.

Closures can be used for more advanced use cases such as decorators, memoization, and callbacks.

Building a Flexible Notification System in Django: A Comprehensive Guide

Mehedi Bappi — Wed, 27 Nov 2024 16:43:46 +0000

Notifications are a key component of any modern web application, ensuring users are informed and engaged. A well-implemented notification system can handle multiple channels like in-app alerts, emails, and SMS while dynamically tailoring content for a seamless user experience. In this guide, we’ll walk you through creating a robust, scalable notification system in Django.

System Features

Our notification system is designed to provide:

Support for Multiple Channels: Notifications via in-app alerts, email, or SMS.
Dynamic Content Personalization: Templates with placeholders to generate personalized messages.
Event-Based Triggers: Trigger notifications based on specific system or user events.
Status Tracking: Monitor the delivery status for email and SMS notifications.
Admin and System Integration: Notifications can be triggered by administrators or system events.

Defining the Models

1. Notification Templates

Templates act as the backbone of our system, storing reusable content for notifications.

from django.db import models

class ChannelType(models.TextChoices):
    APP = 'APP', 'In-App Notification'
    SMS = 'SMS', 'SMS'
    EMAIL = 'EMAIL', 'Email'


class TriggeredByType(models.TextChoices):
    SYSTEM = 'SYSTEM', 'System Notification'
    ADMIN = 'ADMIN', 'Admin Notification'


class TriggerEvent(models.TextChoices):
    ENROLLMENT = 'ENROLLMENT', 'Enrollment'
    ANNOUNCEMENT = 'ANNOUNCEMENT', 'Announcement'
    PROMOTIONAL = 'PROMOTIONAL', 'Promotional'
    RESET_PASSWORD = 'RESET_PASSWORD', 'Reset Password'

class NotificationTemplate(models.Model):
    title = models.CharField(max_length=255)
    template = models.TextField(help_text='Use placeholders like {{username}} for personalization.')
    channel = models.CharField(max_length=20, choices=ChannelType.choices, default=ChannelType.APP)
    triggered_by = models.CharField(max_length=20, choices=TriggeredByType.choices, default=TriggeredByType.SYSTEM)
    trigger_event = models.CharField(max_length=50, choices=TriggerEvent.choices, help_text='Event that triggers this template.')
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

Key Features:

template: Text with placeholders for dynamic values like {{username}}.
channel: Specifies whether it’s an email, SMS, or in-app notification.
trigger_event: Associates the template with a specific event.

2. General Notifications

The Notification model links templates to users and stores any dynamic payload for personalization.

class Notification(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="notifications")
    content = models.ForeignKey(NotificationTemplate, on_delete=models.CASCADE, related_name="notifications")
    payload = models.JSONField(default=dict, help_text="Data to replace template placeholders.")
    is_read = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

3. Channel-Specific Models

To handle emails and SMS uniquely, we define specific models.

Email Notifications

This model manages email-specific data, such as dynamic message generation and delivery tracking.

class StatusType(models.TextChoices):
    PENDING = 'PENDING', 'Pending'
    SUCCESS = 'SUCCESS', 'Success'
    FAILED = 'FAILED', 'Failed'

class EmailNotification(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='email_notifications')
    content = models.ForeignKey(NotificationTemplate, on_delete=models.CASCADE, related_name='email_notifications')
    payload = models.JSONField(default=dict)
    status = models.CharField(max_length=20, choices=StatusType.choices, default=StatusType.PENDING)
    status_reason = models.TextField(null=True)

    @property
    def email_content(self):
        """
        Populate the template with dynamic data from the payload.
        """
        content = self.content.template
        for key, value in self.payload.items():
            content = re.sub(
                rf"{{{{\s*{key}\s*}}}}",
                str(value),
                content,
            )
        return content

SMS Notifications

Similar to email notifications, SMS-specific logic is implemented here.

class SMSNotification(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='sms_notifications')
    content = models.ForeignKey(NotificationTemplate, on_delete=models.CASCADE, related_name='sms_notifications')
    payload = models.JSONField(default=dict)
    status = models.CharField(max_length=20, choices=StatusType.choices, default=StatusType.PENDING)
    status_reason = models.TextField(null=True)

    @property
    def sms_content(self):
        """
        Populate the template with dynamic data from the payload.
        """
        content = self.content.template
        for key, value in self.payload.items():
            content = re.sub(
                rf"{{{{\s*{key}\s*}}}}",
                str(value),
                content,
            )
        return content

Admin Integration

To make managing notifications easier, we register the models in the Django admin panel.

from django.contrib import admin
from notifier.models import NotificationTemplate

@admin.register(NotificationTemplate)
class NotificationTemplateAdmin(admin.ModelAdmin):
    list_display = ['title', 'channel', 'triggered_by', 'trigger_event', 'is_active']
    list_filter = ['channel', 'triggered_by', 'is_active']
    search_fields = ['title', 'trigger_event']

Notification Service

We’ll implement a service layer to manage sending notifications through various channels.

Strategy Pattern

Using the Strategy Pattern, we’ll define classes for each notification channel.

from abc import ABC, abstractmethod
import logging

logger = logging.getLogger(__name__)

class NotificationStrategy(ABC):
    @abstractmethod
    def send(self, user, content, payload):
        pass


class AppNotificationStrategy(NotificationStrategy):
    def send(self, user, content, payload):
        notification = Notification.objects.create(user=user, content=content, payload=payload)
        logger.info(f"In-app notification sent to {user.email}")
        return notification


class EmailNotificationStrategy(NotificationStrategy):
    def send(self, user, content, payload):
        notification = EmailNotification.objects.create(user=user, content=content, payload=payload)
        try:
            self._send_email(user.email, content.title, notification.email_content)
            notification.status = "SUCCESS"
        except Exception as e:
            notification.status = "FAILED"
            notification.status_reason = str(e)
        notification.save()
        return notification

    def _send_email(self, to_email, subject, body):
        print(f"Sending email to {to_email} with subject {subject}")
        if "@" not in to_email:
            raise ValueError("Invalid email address")


class SMSNotificationStrategy(NotificationStrategy):
    def send(self, user, content, payload):
        notification = SMSNotification.objects.create(user=user, content=content, payload=payload)
        try:
            self._send_sms(user.phone_number, notification.sms_content)
            notification.status = "SUCCESS"
        except Exception as e:
            notification.status = "FAILED"
            notification.status_reason = str(e)
        notification.save()
        return notification

    def _send_sms(self, phone_number, message):
        print(f"Sending SMS to {phone_number}: {message}")
        if not phone_number.isdigit():
            raise ValueError("Invalid phone number")

Notification Service

This service ties everything together, selecting the appropriate strategy based on the notification channel.

class NotificationService:
    _strategies: dict[Type[ChannelType], Type[NotificationStrategy]] = {
        ChannelType.APP: AppNotificationStrategy,
        ChannelType.EMAIL: EmailNotificationStrategy,
        ChannelType.SMS: SMSNotificationStrategy,
    }

    @classmethod
    def get_strategy(cls, instance: NotificationTemplate) -> NotificationStrategy:
        try:
            channel = ChannelType[instance.channel]
            strategy = cls._strategies[channel]
        except KeyError:
            raise Exception(f"Unknown notification strategy {instance.channel}")

        return strategy()

    @classmethod
    def notify(
        cls,
        user: User,
        event: TriggerEvent,
        payload: dict,
    ):
        """
        Automatically create and send a system-triggered notification.

        Args:
            user: User instance.
            event: TriggerEvent type.
            payload: Dynamic `dict` data for the notification.

        Returns:
            Result of the notification strategy.
        """

        content, _ = NotificationTemplate.objects.get_or_create(
            trigger_event=event,
            triggered_by=TriggeredByType.SYSTEM,
            defaults={
                "title": 'Default Title',
                "template": "This is a system notification.",
            },
        )
        strategy = cls.get_strategy(instance=content)
        return strategy.send(user, content, payload)

Usage Example

Here’s how you can use the notification service:

from notifier.services import NotificationService, TriggerEvent

user = User.objects.get(email="user@example.com")
payload = {"username": "John Doe", "course": "Python Basics"}

NotificationService.notify(user, TriggerEvent.USER_ENROLLED, payload)

If you found this guide helpful and insightful, don’t forget to like and follow for more content like this. Your support motivates me to share more practical implementations and in-depth tutorials. Let’s keep building amazing applications together!

Forem: Mehedi Bappi

Understanding Closures in Python

What is a Closure?

The nonlocal Keyword

Implementing a Counter with Closures

Step 1: Define the make_counter Function

Step 2: Use nonlocal to Modify the Count Variable

Step 3: Using the Counter

Explanation

Why Use Closures?

Building a Flexible Notification System in Django: A Comprehensive Guide

System Features

Defining the Models

1. Notification Templates

2. General Notifications

3. Channel-Specific Models

Admin Integration

Notification Service

Strategy Pattern

Notification Service

Usage Example

The `nonlocal` Keyword

Step 1: Define the `make_counter` Function

Step 2: Use `nonlocal` to Modify the Count Variable