prefect-email

import os from prefect import flow from prefect_email.credentials import EmailServerCredentials from prefect_email.tasks import email_send_message @flow(log_prints=True) def send_notification_email_flow(): # For production, it is highly recommended to store credentials # in a Prefect EmailServerCredentials Block for security and reusability. # Example: EmailServerCredentials(host=..., port=..., username=..., password=...).save('my-email-creds') # Then load: credentials = EmailServerCredentials.load('my-email-creds') # For this quickstart, we'll instantiate directly from environment variables. # Ensure these are set: SMTP_HOST, SMTP_PORT, SMTP_USERNAME, SMTP_PASSWORD, TO_EMAIL, FROM_EMAIL credentials = EmailServerCredentials( host=os.environ.get("SMTP_HOST", "smtp.mailtrap.io"), port=int(os.environ.get("SMTP_PORT", 2525)), username=os.environ.get("SMTP_USERNAME", "YOUR_USERNAME"), password=os.environ.get("SMTP_PASSWORD", "YOUR_PASSWORD"), use_tls=True # Often required for secure SMTP ) to_emails = os.environ.get("TO_EMAIL", "recipient@example.com").split(',') from_email = os.environ.get("FROM_EMAIL", "sender@example.com") subject = "Prefect Email Test Notification" body = "This is a test email sent from a Prefect flow using prefect-email." print(f"Attempting to send email from {from_email} to {to_emails}...") try: email_send_message( smtp_credentials=credentials, subject=subject, msg=body, email_from=from_email, email_to=to_emails ) print("Email sent successfully!") except Exception as e: print(f"Failed to send email: {e}") if __name__ == "__main__": # To run locally, set environment variables like: # export SMTP_HOST="smtp.mailtrap.io" # export SMTP_PORT="2525" # export SMTP_USERNAME="your_mailtrap_username" # export SMTP_PASSWORD="your_mailtrap_password" # export TO_EMAIL="your_email@example.com" # export FROM_EMAIL="prefect@example.com" # Then execute: python your_script_name.py send_notification_email_flow()