Automatically Transferring DB Dumps, Backup, and Archive Files to Remote Storage

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")   # optional; falls back to the current workspace

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed
NOT_SUCCEEDED = {4, 5, 9, 99}


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(
        method, BASE_URL + path,
        headers=headers, json=body, params=params, timeout=30,
    )

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)

import base64
import time


def encode_path(device_id, raw_path):
    normalized = str(raw_path or "").replace("\\", "/")
    token = base64.b64encode(normalized.encode("utf-8")).decode("ascii")
    return f"{device_id}_ino_{token}"


def now_iso():
    return time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())

import glob
import os


def latest_backup(directory, pattern="*.dump"):
    files = glob.glob(os.path.join(directory, pattern))

    if not files:
        raise FileNotFoundError(directory)

    return max(files, key=os.path.getmtime)


def send_backup(source, target, backup_path, target_path):
    transfer = api("POST", "/api/transfers/manual", {
        "sourceDevice": source,
        "targetDevice": target,
        "targetPath": target_path,
        "sourceItem": [{
            "path": backup_path,
            "isDir": False,
            "fileSize": os.path.getsize(backup_path),
        }],
        "sendAllFolder": False,
        "checkIntegrity": True,
        "transferOptions": {"target-action": "numbering"},
    })

    return transfer["monitorId"]

ROUTES = {
    "dump":    ("device-dc-01", "/backup/dump"),
    "daily":   ("device-cloud-01", "/backup/daily"),
    "archive": ("device-archive-01", "/backup/archive"),
}


def classify(filename):
    name = os.path.basename(filename).lower()

    if name.endswith(".dump"):
        return "dump"

    if "archive" in name or name.endswith(".tar.gz"):
        return "archive"

    return "daily"


def dispatch(source, files):
    transfers = {}

    for path in files:
        target, target_path = ROUTES[classify(path)]
        transfers[path] = send_backup(source, target, path, target_path)

    return transfers

copies = {
    target: send_backup("device-db-01", target, backup_path, path)
    for target, path in [("device-dc-01", "/backup/dump"),
                         ("device-cloud-01", "/backup/mirror")]
}

def build_schedule_automation(name, source, source_path, target, target_path,
                              schedule):
    return {
        "name": name,
        "flowName": name,
        "transferType": "normal",
        "timezone": "Asia/Seoul",
        "step": 1,
        "isUpcoming": False,
        "details": [
            {
                "senderId": source,
                "receiverId": target,
                "sourceItem": [
                    {
                        "hash": encode_path(source, source_path),
                        "filePath": source_path,
                        "isDir": True,
                    }
                ],
                "targetPath": encode_path(target, target_path),
                "step": 1,
                "transferOptions": {
                    "noSchedule": False,
                    "target-action": "numbering",
                    "send-fileoption": {},
                },
            }
        ],
        "schedules": [schedule],
    }


DAILY_3AM = {
    "type": "day",
    "startDateType": "now",
    "hour": "03",
    "minute": "00",
    "ampm": "am",
    "startDate": now_iso(),
    "timezone": "Asia/Seoul",
}

MONTHLY = {
    "type": "month",
    "startDateType": "now",
    "day": "1",
    "hour": "04",
    "minute": "00",
    "ampm": "am",
    "startDate": now_iso(),
    "timezone": "Asia/Seoul",
}

api("POST", "/api/automations", build_schedule_automation(
    "daily backup", "device-db-01", "/backup",
    "device-dc-01", "/backup/daily", DAILY_3AM))

def find_automation(name):
    # the name we send is stored as flowName in the response
    # automationName is a server generated id like T4037-8500-1815, not the name we set.
    for page in range(1, 6):
        result = api("GET", "/api/automations",
                     params={"page": page, "size": 100, "search": name}) or {}

        items = [item
                 for flow in result.get("automations") or []
                 for item in flow.get("automations") or []]

        for item in items:
            if item.get("flowName") == name:
                return item

        if len(items) < 100:
            return None

    return None

def verify(monitor_id, timeout=1800, interval=10):
    api("POST", f"/api/transfers/{monitor_id}/verification", {})

    deadline = time.time() + timeout

    while time.time() < deadline:
        result = api("GET", f"/api/transfers/{monitor_id}/verification") or {}

        if result.get("verified"):
            return result

        time.sleep(interval)

    raise TimeoutError(monitor_id)

result = verify(monitor_id)

if not result["checksumMatched"]:
    alert(f"backup integrity failed: {result.get('mismatchedFiles')}")

def wait(monitor_id, timeout=3600, interval=3):
    deadline = time.time() + timeout

    while time.time() < deadline:
        detail = api("GET", f"/api/transfers/{monitor_id}")

        if is_terminal(detail):
            return detail

        time.sleep(interval)

    raise TimeoutError(monitor_id)


def failed_files(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    return [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]


def retry_failed(monitor_id):
    rows = failed_files(monitor_id)

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)

detail = wait(monitor_id, timeout=7200)

if detail["status"] != STATUS_COMPLETE:
    alert(f"backup transfer failed - retried {retry_failed(monitor_id)} files")

from datetime import datetime, timedelta, timezone


def paginate(path, params=None, limit=200, max_pages=50):
    query = dict(params or {})
    query["limit"] = limit
    cursor = None

    for _ in range(max_pages):
        if cursor:
            query["cursor"] = cursor

        result = api("GET", path, params=query) or {}

        for record in result.get("data") or []:
            yield record

        pagination = result.get("pagination") or {}

        if not pagination.get("hasMore"):
            return

        cursor = pagination.get("nextCursor")

        if not cursor:
            return


end = datetime.now(timezone.utc)
fmt = "%Y-%m-%dT%H:%M:%SZ"

rows = list(paginate("/api/transfer-history", params={
    "startDate": (end - timedelta(days=1)).strftime(fmt),
    "endDate": end.strftime(fmt),
}))

failures = [r for r in rows if r.get("status") in NOT_SUCCEEDED]

if failures:
    alert(f"{len(failures)} backup transfers failed yesterday")

params = {
    "periodDays": 30,
    "page": 1,
    "size": 10000,
    "filter": "[]",              # the server parses this as a JSON string, so send an empty array
    "sort": "startDate:desc",
}