Azure Communication Service Queue

Question

Azure Communication Service Queue

116 Views Asked by karol marszalek At 19 January 2024 at 21:25

Does anyone know how can i queue azure emails and send them in a queue?

Like for example I wanna to create a job to send emails at 10:00 at noon, so it should send that time, not instantly like Azure provides in their documentation. in python.

def _send(self, data: Dict[str, Any], recipients: List[CustomUser]) -> None:
    self._email_template.validate_data(data)
    self._initialize_message()
    relative_link = self._email_template.get_relative_link(data)
    context = self._email_template.get_template_context(data)
    context.update({"relative_link": relative_link})
    self._message.update({"recipients": {"to": self._get_recipients(recipients)}})
    self._message["content"].update({"html": self._render_template(context)})
    try:
        client = self._initialize_connection()
        response = client.begin_send(self._message)

    except Exception as ex:
        logger.error(f"Error occurred during email sending: {ex}")
        raise ex

i just use being_send, there is also async being_send, but its not solving my problem. Should i use kafka and rabbirmq for triggering messages from my app? I would prefer use something which offers Azure tbh. Sorry, i mean its my 1st time on stackoverflow, so I'm kinda shy lol

Original Q&A

There are 1 best solutions below

**Sampath** · Answer 1 · 2024-01-22T07:51:20.883000

The sample below is an Azure Functions Python script that uses the Azure Communication Services Email SDK to send an email using a timer trigger.

The function sets a timer trigger to run at the desired time for sending emails, and the timer interval is appropriately set for your use case.

import datetime
import logging
from azure.communication.email import EmailClient
import azure.functions as func

# Replace the following with your actual connection string
connection_string = "endpoint=https://<resource-name>.communication.azure.com/;accessKey=<Base64-Encoded-Key>"

# Initialize the EmailClient using the connection string
email_client = EmailClient.from_connection_string(connection_string)

def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()

    if mytimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function ran at %s', utc_timestamp)

    # Define the email message
    message = {
        "content": {
            "subject": "This is the subject",
            "plainText": "This is the body",
            "html": "<html><h1>This is the body</h1></html>"
        },
        "recipients": {
            "to": [
                {
                    "address": "[email protected]",
                    "displayName": "Customer Name"
                }
            ]
        },
        "senderAddress": "[email protected]"
    }

    # Send the email message and retrieve the result
    poller = email_client.begin_send(message)
    result = poller.result()

    # Print the result (status and details)
    logging.info("Email sent successfully!")
    logging.info(result)

Output: enter image description here

enter image description here

import datetime
import logging
from typing import Dict, Any, List
from azure.communication.email import EmailClient
import azure.functions as func

# Replace the following with your actual connection string
connection_string = "endpoint=https://<resource-name>.communication.azure.com/;accessKey=<Base64-Encoded-Key>"

# Initialize the EmailClient using the connection string
email_client = EmailClient.from_connection_string(connection_string)

# CustomUser class definition (replace with your actual implementation)
class CustomUser:
    def __init__(self, email: str, name: str):
        self.email = email
        self.name = name

class EmailSender:
    def __init__(self):
        # Add your email template initialization logic here
        pass

    def _initialize_connection(self):
        # Replace this with your actual logic to initialize the connection
        return EmailClient.from_connection_string(connection_string)

    def _send(self, data: Dict[str, Any], recipients: List[CustomUser]) -> None:
        # Your existing _send method implementation
        client = self._initialize_connection()
        context = {"key": "value"}  # Replace with your actual data
        message = {
            "content": {
                "subject": "This is the subject",
                "plainText": "This is the body",
                "html": "<html><h1>This is the body</h1></html>",  # Replace with your HTML template
            },
            "recipients": {"to": self._get_recipients(recipients)},
            "senderAddress": "[email protected]",
        }
        client.begin_send(message).result()

    def _get_recipients(self, recipients: List[CustomUser]) -> List[Dict[str, str]]:
        # Replace this with your logic to convert CustomUser instances to recipients dictionary
        return [{"address": user.email, "displayName": user.name} for user in recipients]

# Azure Function
def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc
    ).isoformat()

    if mytimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function ran at %s', utc_timestamp)

    # Your email data and recipients
    email_data = {"key1": "value1", "key2": "value2"}  # Replace with your actual data
    recipients_list = [CustomUser(email="[email protected]", name="Recipient Name")]

    # Create an instance of EmailSender
    email_sender = EmailSender()

    try:
        # Call the _send method to send the email
        email_sender._send(email_data, recipients_list)

        logging.info("Email sent successfully!")
    except Exception as ex:
        logging.error(f"Error occurred during email sending: {ex}")
        # Handle the exception as needed
        raise ex

Azure Communication Service Queue

There are 1 best solutions below

Related Questions in PYTHON

Related Questions in AZURE

Related Questions in AZURE-COMMUNICATION-SERVICES

Trending Questions

Popular # Hahtags

Popular Questions