storage.py

import os
import subprocess
from google.cloud.storage import Client, Bucket

def create_tar(paths: list[str], tar_name: str) -> None:
    """Create an uncompressed tarball without including the full path of the files inside."""
    abs_paths = [os.path.abspath(path) for path in paths]
    common_dir = os.path.commonpath(abs_paths)
    cmd = ['tar', '--create', '--file', tar_name]
    for path in abs_paths:
        cmd.append('--directory')
        cmd.append(common_dir)
        cmd.append(os.path.relpath(path, common_dir))

    subprocess.run(cmd, check=True)
    print(f"Created uncompressed tarball: {tar_name}")


def extract_tar(tar_path: str, extract_to_path: str) -> None:
    """Extract the tarball to the recovery path."""
    print(f"Extracting {tar_path} to {extract_to_path}...")

    # Command to extract tarball
    cmd = ['tar', '--extract', '--file', tar_path, '--directory', extract_to_path]
    subprocess.run(cmd, check=True)
    print(f"Extraction complete: {extract_to_path}")


def download_from_bucket(storage_client: Client, bucket_name: str, bucket_path: str, local_path: str) -> None:
    """Download the snapshot tar file from Google Cloud Storage."""
    bucket: Bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(bucket_path)
    print(f"Downloading {bucket_path} from gs://{bucket_name}...")
    
    # Download the tar file to the local path
    blob.download_to_filename(local_path)
    print(f"Downloaded {bucket_path} to {local_path}")


def upload_file_to_bucket(storage_client: Client, bucket_name: str, source_file: str, destination_blob_name: str) -> None:
    """Upload a file to the specified Google Cloud Storage bucket."""
    try:
        print(f"Start uploading to {destination_blob_name}.")
        bucket: Bucket = storage_client.get_bucket(bucket_name)
        blob = bucket.blob(destination_blob_name)
        blob.upload_from_filename(source_file)
        print(f"File {source_file} uploaded to {destination_blob_name}.")
    except Exception as e:
        print(f"Error uploading file: {e}")