Data Ingestion Strategies and Tools Questions

import boto3, sqlite3, os, time, random
from botocore.exceptions import ClientError

CHUNK_SIZE = 8 * 1024 * 1024

def backoff_retry(fn, max_attempts=6):
    for attempt in range(1, max_attempts+1):
        try:
            return fn()
        except ClientError as e:
            if attempt == max_attempts:
                raise
            sleep = (2 ** (attempt - 1)) + random.uniform(0, 1)
            time.sleep(sleep)

class MultipartUploader:
    def __init__(self, bucket, key, db_path='uploads.db'):
        self.s3 = boto3.client('s3')
        self.bucket = bucket; self.key = key
        self.db = sqlite3.connect(db_path, timeout=10)
        self._ensure_table()

    def _ensure_table(self):
        c = self.db.cursor()
        c.execute('''CREATE TABLE IF NOT EXISTS uploads
                     (key TEXT PRIMARY KEY, upload_id TEXT, parts TEXT)''')
        self.db.commit()

    def _save_state(self, upload_id, parts):
        c = self.db.cursor()
        c.execute('REPLACE INTO uploads (key, upload_id, parts) VALUES (?, ?, ?)',
                  (self.key, upload_id, repr(parts)))
        self.db.commit()

    def _load_state(self):
        c = self.db.cursor()
        r = c.execute('SELECT upload_id, parts FROM uploads WHERE key=?', (self.key,)).fetchone()
        if not r: return None, {}
        return r[0], eval(r[1])

    def upload_file(self, path):
        upload_id, parts = self._load_state()
        if not upload_id:
            resp = backoff_retry(lambda: self.s3.create_multipart_upload(Bucket=self.bucket, Key=self.key))
            upload_id = resp['UploadId']
            parts = {}
            self._save_state(upload_id, parts)

        # reconcile with S3 in case some parts exist
        remote_parts = {}
        resp = self.s3.list_parts(Bucket=self.bucket, Key=self.key, UploadId=upload_id)
        for p in resp.get('Parts', []):
            remote_parts[p['PartNumber']] = p['ETag']
        parts.update(remote_parts)
        self._save_state(upload_id, parts)

        with open(path, 'rb') as f:
            part_num = 1
            while True:
                chunk = f.read(CHUNK_SIZE)
                if not chunk:
                    break
                if part_num in parts:
                    part_num += 1
                    continue
                def upload_part():
                    return self.s3.upload_part(Bucket=self.bucket, Key=self.key,
                                               PartNumber=part_num, UploadId=upload_id,
                                               Body=chunk)
                resp = backoff_retry(upload_part)
                parts[part_num] = resp['ETag']
                self._save_state(upload_id, parts)
                part_num += 1

        # verify and complete exactly once
        # double-check with list_parts
        final_parts = [{'ETag': parts[p], 'PartNumber': p} for p in sorted(parts)]
        # Ensure no duplicate final object by checking for existing completed object with same metadata optional
        backoff_retry(lambda: self.s3.complete_multipart_upload(
            Bucket=self.bucket, Key=self.key, UploadId=upload_id,
            MultipartUpload={'Parts': final_parts}))
        # cleanup state
        c = self.db.cursor(); c.execute('DELETE FROM uploads WHERE key=?', (self.key,)); self.db.commit()