Python ve boto3 ile Amazon SES entegrasyonu

· Furkan Ergin · 4 dk okuma · Python , AWS , E-posta , boto3

Selamlar, bu yazıda Python uygulamalarından Amazon SES üzerinden e-posta göndermeyi konuşalım. Aslında konu basit görünür: send_email çağır, bitti. Ama işin içine SES sandbox'ı, bounce'lar, throttling ve attachment göndermek girince sandalye birden çetrefilliyor. Hadi adım adım, ben hangi tuzaklara takıldıysam onlardan başlayarak gidelim.

Neden SES?

Önce küçük bir bağı kurayım. SES (Simple Email Service) AWS'in işlem hacmine göre ödeme aldığı transactional ve marketing e-posta servisi. Ücret tarafı çok ucuz: 1000 e-posta için 0.10 dolar civarı. Sendgrid veya Mailgun'a göre daha az aksesuar ile geliyor ama temel ihtiyaçları fazlasıyla karşılıyor. Bence özellikle zaten AWS'te çalışıyorsanız, IAM tarafından kimlik doğrulamayı çözmüş olmanız tek başına yeterli sebep.

Kurulum ve kimlik bilgileri

boto3'ü kurmakla başlayalım:

pip install boto3

boto3 kimlik bilgilerini şu sırayla arar: ortam değişkenleri, ~/.aws/credentials dosyası, son olarak EC2/Lambda/ECS üzerindeki IAM rolü. Lokalde çalışıyorsanız aws configure ile yapılandırma dosyasını oluşturmak en hızlı yol. Production'da ise rica ediyorum, lütfen AWS_ACCESS_KEY_ID'yi koda gömmeyin; IAM role veya en azından AWS Secrets Manager kullanın.

Bir de çok kritik bir not: SES'i ilk açan herkes sandbox modundadır. Bu modda sadece kendi doğruladığınız adreslere mail gönderebilirsiniz. Production'a geçmek için AWS konsolundan küçük bir form doldurup çıkış istemeniz gerek. Ben bunu ilk seferinde gözden kaçırmıştım, neden test mailim arkadaşıma gitmiyor diye yarım saat uğraşmıştım. Böyle olmasın.

Basit bir mail gönderimi

En sade haliyle bir mail nasıl gider, bakalım:

import boto3
from botocore.exceptions import ClientError

ses = boto3.client('ses', region_name='eu-central-1')

def mail_gonder(alici, konu, metin):
    try:
        cevap = ses.send_email(
            Source='bildirim@ornekdomain.com',
            Destination={'ToAddresses': [alici]},
            Message={
                'Subject': {'Data': konu, 'Charset': 'UTF-8'},
                'Body': {'Text': {'Data': metin, 'Charset': 'UTF-8'}},
            },
        )
        return cevap['MessageId']
    except ClientError as hata:
        kod = hata.response['Error']['Code']
        print(f'SES hatasi ({kod}): {hata.response["Error"]["Message"]}')
        raise

Charset parametresini her yere UTF-8 koyduğuma dikkat edin. Türkçe karakterler için bu şart; aksi halde 'Merhaba arkadaşlar' gibi sıcak başlayan mailler 'Merhaba arkada?lar' olarak gidiyor ki bu çok kötü durur.

Production için minik bir servis sınıfı

Tek fonksiyon hızlı ama gerçek hayatta retry, log ve birkaç yardımcısı olan bir sınıfa dönmek isteyeceksiniz:

import json
import logging
import time
import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)

class MailServisi:
    YENIDENDENENECEK = {'Throttling', 'ServiceUnavailable', 'RequestTimeout'}

    def __init__(self, gonderen, region='eu-central-1', config_set=None):
        self.ses = boto3.client('ses', region_name=region)
        self.gonderen = gonderen
        self.config_set = config_set
        self.max_deneme = 3

    def gonder(self, alici, konu, text=None, html=None, etiketler=None):
        govde = {}
        if text:
            govde['Text'] = {'Data': text, 'Charset': 'UTF-8'}
        if html:
            govde['Html'] = {'Data': html, 'Charset': 'UTF-8'}
        if not govde:
            raise ValueError('text veya html zorunlu')

        params = {
            'Source': self.gonderen,
            'Destination': {'ToAddresses': [alici] if isinstance(alici, str) else list(alici)},
            'Message': {
                'Subject': {'Data': konu, 'Charset': 'UTF-8'},
                'Body': govde,
            },
        }
        if self.config_set:
            params['ConfigurationSetName'] = self.config_set
        if etiketler:
            params['Tags'] = [{'Name': k, 'Value': v} for k, v in etiketler.items()]

        return self._dene('send_email', params)

    def _dene(self, metot_adi, params):
        metot = getattr(self.ses, metot_adi)
        for deneme in range(self.max_deneme):
            try:
                cevap = metot(**params)
                logger.info('mail gitti: %s', cevap.get('MessageId'))
                return {'ok': True, 'id': cevap.get('MessageId')}
            except ClientError as hata:
                kod = hata.response['Error']['Code']
                if kod in self.YENIDENDENENECEK and deneme < self.max_deneme - 1:
                    bekle = (2 ** deneme) + 1
                    logger.warning('throttle, %ss bekleyip tekrar', bekle)
                    time.sleep(bekle)
                    continue
                logger.error('SES hatasi (%s): %s', kod, hata.response['Error']['Message'])
                return {'ok': False, 'kod': kod}

Burada önemli iki şey var: birincisi YENIDENDENENECEK setiyle hangi hataların yeniden deneneceğine karar veriyoruz, ikincisi exponential backoff. SES Throttling döndüğünde hemen tekrar atmak işi çözmez, kuyruğa bir nefes lazım.

Tags ve ConfigurationSet meselesi

Yukarıdaki etiketler parametresi gerçekten işe yarıyor. Configuration Set tanımlayıp CloudWatch'a bounce/complaint metriklerini bağladığınızda, bu etiketler dashboard'da kampanya/tip ayrımı yapmanızı sağlıyor. Şahsi kanaatim, en baştan EmailType diye bir tag koymak ileride 'şifre sıfırlamada bounce oranı neden 4 katı' gibi bir soruya cevap vermeyi çok kolaylaştırıyor.

Sık karşılaşılan tuzaklar

  • Sandbox'ı unutmak: Mail kimseye gitmiyor, kod da hata vermiyor; çünkü siz doğrulanmamış adrese atıyorsunuz. Konsoldan production access isteyin.
  • From adresi doğrulanmamış: SES Email address is not verified döner. Domain doğrulamak en sağlamı; tek tek adres doğrulamak sürdürülemez.
  • Region karıştırmak: Sandbox çıkışı region bazlı. eu-central-1'de çıkış aldım diye us-east-1'de aynı hakka sahip değilsiniz, baktığım her ekibin bir kere yediği kazık.
  • Throttling'i ezmek: Yeni hesabın saniyede 14 mail limitine takılırsınız. Toplu gönderimde time.sleep(1/14) ile yumuşatmak veya SQS aracılması sık kullanılan iki yol.
  • Bounce'ları görmezden gelmek: Bounce oranı %5'i geçince AWS hesabı askıya alabiliyor. SNS'e bounce/complaint topic bağlamak opsiyonel değil, mecburi.

Kapanış

Bu yazıda SES'i Python tarafından nasıl ehlileştireceğimize, sandbox tuzağına, throttling'e ve kücük ama derli toplu bir servis sınıfına baktık. Bence Django ve Flask için ayrı ayrı uzun setup'a girmeden önce böyle bir MailServisi sınıfını elde tutmak işi taşınabilir kılıyor; sonra istediğiniz framework'e çağrı yapacaksınız, o kadar. Umarım faydalı olur, görüşmek üzere.