mailgun template migration to different accounts

mailgun-migrate.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
#
# Copyright (C) 2024 Kalyan Mudumby <theinhumaneme@gmail.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
# ----------------------------------------------------------------------------
#    Code File: mailgun-migrate.py
#    Author: Kalyan Mudumby
#    Date: 17 December 2024
#    Description:
#           Automate export and import of mailgun templates within or across accounts
#
#    For questions or concerns, please contact `Kalyan Mudumby` at
#    `theinhumaneme@gmail.com`.
#
#   Built with Love ❤️ by Kalyan Mudumby 🚀
# ----------------------------------------------------------------------------


import json
import requests
import os
import logging
import click
import sys
import csv
import concurrent.futures
from typing import List, Dict, Any
import re
from urllib.parse import quote


# Source and destination Mailgun API keys and domains
# The Script uses Basic Auth Read More -> https://documentation.mailgun.com/docs/mailgun/api-reference/authentication/

MAILGUN_SOURCE_API_KEY: str = os.getenv("MAILGUN_SOURCE_API_KEY", "")
MAILGUN_SOURCE_DOMAIN: str = os.getenv("MAILGUN_SOURCE_DOMAIN", "")
MAILGUN_DESTINATION_API_KEY: str = os.getenv("MAILGUN_DESTINATION_API_KEY", "")
MAILGUN_DESTINATION_DOMAIN: str = os.getenv("MAILGUN_DESTINATION_DOMAIN", "")

TEMPLATES_FILE_JSON = "templates.json"
TEMPLATES_FILE_CSV = "templates.csv"

VARIABLE_RE = re.compile(r"\{\{\{?\s*([A-Za-z0-9_.]+)\s*\}?\}\}")

logging.basicConfig(format="%(asctime)s %(levelname)s %(message)s", level=logging.INFO)
logger = logging.getLogger(__name__)


def extract_template_variables(html: str) -> List[str]:
    """
    Find all unique Handlebars/Mustache variables in `html`.
    E.g. turns "<h1>{{ title }}</h1>{{#if user}}Hello {{user.name}}{{/if}}"
    into ["title", "user", "user.name"].
    """
    names = VARIABLE_RE.findall(html)
    seen: Dict[str, bool] = {}
    for name in names:
        seen.setdefault(name, True)
    return list(seen.keys())


def download_templates(
    session: requests.Session, output_format: str = "json", workers: int = 2
) -> None:
    """
    Export all templates from source Mailgun account.
    Fetch metadata, then in parallel fetch HTML content for both JSON and CSV formats.
    """
    logger.info("Downloading Templates")
    items: list[dict[str, str]] = []
    url: str = f"https://api.mailgun.net/v3/{MAILGUN_SOURCE_DOMAIN}/templates"

    visited: set[str] = set()

    while url:
        if url in visited:
            logger.warning(f"Pagination URL repeated ({url}); stopping.")
            break
        visited.add(url)

        resp = session.get(url, auth=("api", MAILGUN_SOURCE_API_KEY), timeout=(5, 30))
        resp.raise_for_status()
        data: Dict[str, Any] = resp.json()

        page_items = data.get("items", [])
        if not page_items:
            break
        items.extend(page_items)

        next_url = data.get("paging", {}).get("next", "")
        if not next_url or next_url == url:
            break
        url = next_url

    def populate(template: Dict[str, Any]) -> Dict[str, Any]:
        name: str = template.get("name", "")
        logger.debug(f"Fetching content for {name}")
        logger.info(f"Downloading template for {name}")
        html = fetch_template_content(session, name)
        template["html"] = html
        template["variables"] = extract_template_variables(html)
        return template

    workers = max(1, min(workers, 8))
    logger.debug("Downloading HTML with %d parallel workers", workers)
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
        populated_items = list(executor.map(populate, items))
    logger.info("Fetched HTML content for all templates.")

    if output_format in ("json", "both"):
        with open(TEMPLATES_FILE_JSON, "w", encoding="utf-8") as jf:
            json.dump({"items": populated_items}, jf, indent=2)
        logger.info(f"Exporting {len(populated_items)} templates to JSON file")
        logger.info(f"Wrote populated JSON to {TEMPLATES_FILE_JSON}")

    if output_format in ("csv", "both"):
        fieldnames = ["name", "description", "html", "variables"]
        with open(TEMPLATES_FILE_CSV, "w", newline="", encoding="utf-8") as cf:
            writer = csv.DictWriter(cf, fieldnames=fieldnames, extrasaction="ignore")
            writer.writeheader()
            for template in populated_items:
                writer.writerow(
                    {
                        "name": template.get("name", ""),
                        "description": template.get("description", ""),
                        "html": template.get("html", ""),
                        "variables": "|".join(
                            extract_template_variables(template.get("html", ""))
                        ),
                    }
                )
        logger.info(f"Exporting {len(populated_items)} templates to CSV file")
        logger.info(f"Wrote enriched CSV to {TEMPLATES_FILE_CSV}")


def fetch_template_content(session: requests.Session, name: str) -> str:
    """Return the HTML body of the latest active version of a template."""
    url = f"https://api.mailgun.net/v3/{MAILGUN_SOURCE_DOMAIN}/templates/{quote(name, safe='')}?active=yes"
    resp = session.get(
        url,
        auth=("api", MAILGUN_SOURCE_API_KEY),
        params={"active": "yes"},
        timeout=(5, 30),
    )
    resp.raise_for_status()

    template = resp.json().get("template", {})
    version: Dict[str, Any] = template.get("version") or {}
    template_html = version.get("template", "")
    return template_html


def create_template(
    session: requests.Session,
    template_name: str,
    template_description: str,
    template_html: str,
):
    create_url = f"https://api.mailgun.net/v3/{MAILGUN_DESTINATION_DOMAIN}/templates"
    payload: dict[str, str] = {
        "name": template_name,
        "description": template_description,
        "template": template_html,
        "engine": "handlebars",
    }
    resp = session.post(
        create_url, auth=("api", MAILGUN_DESTINATION_API_KEY), data=payload
    )
    resp.raise_for_status()
    logger.info(f"Created template '{template_name}' on destination")


@click.group()
@click.option(
    "--dry-run/--no-dry-run",
    default=False,
    help="Show what would be done without making any changes.",
)
@click.pass_context
def cli(ctx: click.Context, dry_run: bool):
    """
    Mailgun Template Migration Tool

    Use `export` to download templates from the source account into JSON.
    Use `migrate` to copy templates from source and create them in the destination account.
    """
    ctx.ensure_object(dict)
    ctx.obj["DRY_RUN"] = dry_run

    for var in (
        "MAILGUN_SOURCE_API_KEY",
        "MAILGUN_SOURCE_DOMAIN",
    ):
        if not os.getenv(var):
            logger.error(f"Environment variable {var} is not set.")
            sys.exit(1)
    ctx.obj["SESSION"] = requests.Session()


@cli.command(help="Export templates to JSON or CSV with content included for JSON.")
@click.option(
    "--format",
    "-f",
    type=click.Choice(["json", "csv", "both"]),
    default="json",
    show_default=True,
    help="Output format for exported templates.",
)
@click.option(
    "--workers",
    "-w",
    type=click.IntRange(1, 8),
    default=2,
    show_default=True,
    help="Number of parallel downloads (1-8).",
)
@click.pass_context
def export(ctx: click.Context, format: str, workers: int):
    """
    Fetch templates metadata and content, write them out in the chosen format.
    """
    if ctx.obj["DRY_RUN"]:
        logger.info(f"[dry-run] Would fetch templates and write {format}")
        return
    else:
        download_templates(
            session=ctx.obj["SESSION"], output_format=format, workers=workers
        )


@cli.command(help="Migrate templates from exported JSON to destination account")
@click.option(
    "--input-file",
    default=TEMPLATES_FILE_JSON,
    show_default=True,
    help="Path to the exported templates JSON.",
)
@click.pass_context
def migrate(ctx: click.Context, input_file: str):
    dry: bool = ctx.obj["DRY_RUN"]
    session = ctx.obj["SESSION"]

    for var in (
        "MAILGUN_DESTINATION_API_KEY",
        "MAILGUN_DESTINATION_DOMAIN",
    ):
        if not os.getenv(var):
            logger.error(f"Environment variable {var} is not set.")
            sys.exit(1)

    if not os.path.exists(input_file):
        logger.warning(f"Input file {input_file} not found. Exporting templates")
        download_templates(session, output_format="json")

    with open(input_file, encoding="utf-8") as fh:
        data = json.load(fh)
        items = data.get("items", [])
        logger.info(f"Found {len(items)} templates to migrate.")

        for template in items:
            template_name: str = template.get("name")
            template_description: str = template.get("description", "")
            template_html: str = template.get("html", "")
            logger.info(f"Migrating template '{template_name}'...")
            if dry:
                logger.info(f"[dry-run] Would create '{template_name}'.")
            else:
                create_template(
                    session=session,
                    template_name=template_name,
                    template_description=template_description,
                    template_html=template_html,
                )

        logger.info(f"Migration {'dry-run ' if dry else ''} complete.")
        sys.exit(0)


if __name__ == "__main__":
    cli()

pyproject.toml

[project]
name = "mailgun-migrate"
version = "0.1.0"
description = "Sekai to Isekai"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
    "click>=8.2.1",
    "requests>=2.32.4",
]

requirements.txt

certifi==2025.6.15
charset-normalizer==3.4.2
click==8.2.1
idna==3.10
requests==2.32.4
urllib3==2.4.0

uv.lock

version = 1
revision = 2
requires-python = ">=3.13"

[[package]]
name = "certifi"
version = "2025.6.15"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/73/f7/f14b46d4bcd21092d7d3ccef689615220d8a08fb25e564b65d20738e672e/certifi-2025.6.15.tar.gz", hash = "sha256:d747aa5a8b9bbbb1bb8c22bb13e22bd1f18e9796defa16bab421f7f7a317323b", size = 158753, upload-time = "2025-06-15T02:45:51.329Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/84/ae/320161bd181fc06471eed047ecce67b693fd7515b16d495d8932db763426/certifi-2025.6.15-py3-none-any.whl", hash = "sha256:2e0c7ce7cb5d8f8634ca55d2ba7e6ec2689a2fd6537d8dec1296a477a4910057", size = 157650, upload-time = "2025-06-15T02:45:49.977Z" },
]

[[package]]
name = "charset-normalizer"
version = "3.4.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/e4/33/89c2ced2b67d1c2a61c19c6751aa8902d46ce3dacb23600a283619f5a12d/charset_normalizer-3.4.2.tar.gz", hash = "sha256:5baececa9ecba31eff645232d59845c07aa030f0c81ee70184a90d35099a0e63", size = 126367, upload-time = "2025-05-02T08:34:42.01Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/ea/12/a93df3366ed32db1d907d7593a94f1fe6293903e3e92967bebd6950ed12c/charset_normalizer-3.4.2-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:926ca93accd5d36ccdabd803392ddc3e03e6d4cd1cf17deff3b989ab8e9dbcf0", size = 199622, upload-time = "2025-05-02T08:32:56.363Z" },
    { url = "https://files.pythonhosted.org/packages/04/93/bf204e6f344c39d9937d3c13c8cd5bbfc266472e51fc8c07cb7f64fcd2de/charset_normalizer-3.4.2-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:eba9904b0f38a143592d9fc0e19e2df0fa2e41c3c3745554761c5f6447eedabf", size = 143435, upload-time = "2025-05-02T08:32:58.551Z" },
    { url = "https://files.pythonhosted.org/packages/22/2a/ea8a2095b0bafa6c5b5a55ffdc2f924455233ee7b91c69b7edfcc9e02284/charset_normalizer-3.4.2-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3fddb7e2c84ac87ac3a947cb4e66d143ca5863ef48e4a5ecb83bd48619e4634e", size = 153653, upload-time = "2025-05-02T08:33:00.342Z" },
    { url = "https://files.pythonhosted.org/packages/b6/57/1b090ff183d13cef485dfbe272e2fe57622a76694061353c59da52c9a659/charset_normalizer-3.4.2-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:98f862da73774290f251b9df8d11161b6cf25b599a66baf087c1ffe340e9bfd1", size = 146231, upload-time = "2025-05-02T08:33:02.081Z" },
    { url = "https://files.pythonhosted.org/packages/e2/28/ffc026b26f441fc67bd21ab7f03b313ab3fe46714a14b516f931abe1a2d8/charset_normalizer-3.4.2-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c9379d65defcab82d07b2a9dfbfc2e95bc8fe0ebb1b176a3190230a3ef0e07c", size = 148243, upload-time = "2025-05-02T08:33:04.063Z" },
    { url = "https://files.pythonhosted.org/packages/c0/0f/9abe9bd191629c33e69e47c6ef45ef99773320e9ad8e9cb08b8ab4a8d4cb/charset_normalizer-3.4.2-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e635b87f01ebc977342e2697d05b56632f5f879a4f15955dfe8cef2448b51691", size = 150442, upload-time = "2025-05-02T08:33:06.418Z" },
    { url = "https://files.pythonhosted.org/packages/67/7c/a123bbcedca91d5916c056407f89a7f5e8fdfce12ba825d7d6b9954a1a3c/charset_normalizer-3.4.2-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:1c95a1e2902a8b722868587c0e1184ad5c55631de5afc0eb96bc4b0d738092c0", size = 145147, upload-time = "2025-05-02T08:33:08.183Z" },
    { url = "https://files.pythonhosted.org/packages/ec/fe/1ac556fa4899d967b83e9893788e86b6af4d83e4726511eaaad035e36595/charset_normalizer-3.4.2-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:ef8de666d6179b009dce7bcb2ad4c4a779f113f12caf8dc77f0162c29d20490b", size = 153057, upload-time = "2025-05-02T08:33:09.986Z" },
    { url = "https://files.pythonhosted.org/packages/2b/ff/acfc0b0a70b19e3e54febdd5301a98b72fa07635e56f24f60502e954c461/charset_normalizer-3.4.2-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:32fc0341d72e0f73f80acb0a2c94216bd704f4f0bce10aedea38f30502b271ff", size = 156454, upload-time = "2025-05-02T08:33:11.814Z" },
    { url = "https://files.pythonhosted.org/packages/92/08/95b458ce9c740d0645feb0e96cea1f5ec946ea9c580a94adfe0b617f3573/charset_normalizer-3.4.2-cp313-cp313-musllinux_1_2_s390x.whl", hash = "sha256:289200a18fa698949d2b39c671c2cc7a24d44096784e76614899a7ccf2574b7b", size = 154174, upload-time = "2025-05-02T08:33:13.707Z" },
    { url = "https://files.pythonhosted.org/packages/78/be/8392efc43487ac051eee6c36d5fbd63032d78f7728cb37aebcc98191f1ff/charset_normalizer-3.4.2-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:4a476b06fbcf359ad25d34a057b7219281286ae2477cc5ff5e3f70a246971148", size = 149166, upload-time = "2025-05-02T08:33:15.458Z" },
    { url = "https://files.pythonhosted.org/packages/44/96/392abd49b094d30b91d9fbda6a69519e95802250b777841cf3bda8fe136c/charset_normalizer-3.4.2-cp313-cp313-win32.whl", hash = "sha256:aaeeb6a479c7667fbe1099af9617c83aaca22182d6cf8c53966491a0f1b7ffb7", size = 98064, upload-time = "2025-05-02T08:33:17.06Z" },
    { url = "https://files.pythonhosted.org/packages/e9/b0/0200da600134e001d91851ddc797809e2fe0ea72de90e09bec5a2fbdaccb/charset_normalizer-3.4.2-cp313-cp313-win_amd64.whl", hash = "sha256:aa6af9e7d59f9c12b33ae4e9450619cf2488e2bbe9b44030905877f0b2324980", size = 105641, upload-time = "2025-05-02T08:33:18.753Z" },
    { url = "https://files.pythonhosted.org/packages/20/94/c5790835a017658cbfabd07f3bfb549140c3ac458cfc196323996b10095a/charset_normalizer-3.4.2-py3-none-any.whl", hash = "sha256:7f56930ab0abd1c45cd15be65cc741c28b1c9a34876ce8c17a2fa107810c0af0", size = 52626, upload-time = "2025-05-02T08:34:40.053Z" },
]

[[package]]
name = "click"
version = "8.2.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
    { name = "colorama", marker = "sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/60/6c/8ca2efa64cf75a977a0d7fac081354553ebe483345c734fb6b6515d96bbc/click-8.2.1.tar.gz", hash = "sha256:27c491cc05d968d271d5a1db13e3b5a184636d9d930f148c50b038f0d0646202", size = 286342, upload-time = "2025-05-20T23:19:49.832Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/85/32/10bb5764d90a8eee674e9dc6f4db6a0ab47c8c4d0d83c27f7c39ac415a4d/click-8.2.1-py3-none-any.whl", hash = "sha256:61a3265b914e850b85317d0b3109c7f8cd35a670f963866005d6ef1d5175a12b", size = 102215, upload-time = "2025-05-20T23:19:47.796Z" },
]

[[package]]
name = "colorama"
version = "0.4.6"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
]

[[package]]
name = "gists"
version = "0.1.0"
source = { virtual = "." }
dependencies = [
    { name = "click" },
    { name = "requests" },
]

[package.metadata]
requires-dist = [
    { name = "click", specifier = ">=8.2.1" },
    { name = "requests", specifier = ">=2.32.4" },
]

[[package]]
name = "idna"
version = "3.10"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f1/70/7703c29685631f5a7590aa73f1f1d3fa9a380e654b86af429e0934a32f7d/idna-3.10.tar.gz", hash = "sha256:12f65c9b470abda6dc35cf8e63cc574b1c52b11df2c86030af0ac09b01b13ea9", size = 190490, upload-time = "2024-09-15T18:07:39.745Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/76/c6/c88e154df9c4e1a2a66ccf0005a88dfb2650c1dffb6f5ce603dfbd452ce3/idna-3.10-py3-none-any.whl", hash = "sha256:946d195a0d259cbba61165e88e65941f16e9b36ea6ddb97f00452bae8b1287d3", size = 70442, upload-time = "2024-09-15T18:07:37.964Z" },
]

[[package]]
name = "requests"
version = "2.32.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
    { name = "certifi" },
    { name = "charset-normalizer" },
    { name = "idna" },
    { name = "urllib3" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e1/0a/929373653770d8a0d7ea76c37de6e41f11eb07559b103b1c02cafb3f7cf8/requests-2.32.4.tar.gz", hash = "sha256:27d0316682c8a29834d3264820024b62a36942083d52caf2f14c0591336d3422", size = 135258, upload-time = "2025-06-09T16:43:07.34Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/7c/e4/56027c4a6b4ae70ca9de302488c5ca95ad4a39e190093d6c1a8ace08341b/requests-2.32.4-py3-none-any.whl", hash = "sha256:27babd3cda2a6d50b30443204ee89830707d396671944c998b5975b031ac2b2c", size = 64847, upload-time = "2025-06-09T16:43:05.728Z" },
]

[[package]]
name = "urllib3"
version = "2.4.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/8a/78/16493d9c386d8e60e442a35feac5e00f0913c0f4b7c217c11e8ec2ff53e0/urllib3-2.4.0.tar.gz", hash = "sha256:414bc6535b787febd7567804cc015fee39daab8ad86268f1310a9250697de466", size = 390672, upload-time = "2025-04-10T15:23:39.232Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/6b/11/cc635220681e93a0183390e26485430ca2c7b5f9d33b15c74c2861cb8091/urllib3-2.4.0-py3-none-any.whl", hash = "sha256:4e16665048960a0900c702d4a66415956a584919c03361cac9f1df5c5dd7e813", size = 128680, upload-time = "2025-04-10T15:23:37.377Z" },
]