35 - Google Cloud Storage Advanced Features

from google.cloud import storage  # Import GCS client
from typing import Dict, Any  # For type annotations
import logging  # For logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def set_lifecycle_rule(bucket_name: str, days: int) -> None:
    """Set lifecycle rule to delete files older than days."""
    client = storage.Client()  # Initialize GCS client
    bucket = client.get_bucket(bucket_name)  # Get bucket
    if days <= 0:
        logger.warning(f"Invalid lifecycle days {days}; skipping rule")
        return
    rule = {
        "action": {"type": "Delete"},  # Delete action
        "condition": {"age": days}  # Files older than days
    }
    bucket.lifecycle_rules = [rule]  # Set lifecycle rules
    bucket.patch()  # Update bucket
    logger.info(f"Set lifecycle rule on {bucket_name} to delete after {days} days")
    print(f"Lifecycle rules: {list(bucket.lifecycle_rules)}")  # Debug

# Example usage
set_lifecycle_rule("my-data-lake-bucket", 365)  # Delete files after 1 year

# Expected Output:
# INFO:__main__:Set lifecycle rule on my-data-lake-bucket to delete after 365 days
# Lifecycle rules: [{'action': {'type': 'Delete'}, 'condition': {'age': 365}}]

from google.cloud import storage  # Import GCS client
from typing import Dict, Any  # For type annotations
import logging  # For logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def set_iam_policy(bucket_name: str, member: str, role: str) -> None:
    """Set IAM policy to grant role to member."""
    client = storage.Client()  # Initialize client
    bucket = client.get_bucket(bucket_name)  # Get bucket
    valid_roles = ["roles/storage.objectViewer", "roles/storage.objectAdmin", "roles/storage.admin"]
    if role not in valid_roles:
        logger.warning(f"Invalid IAM role {role}; skipping policy update")
        return
    policy = bucket.get_iam_policy()  # Get current policy
    policy[role].add(member)  # Add member to role
    bucket.set_iam_policy(policy)  # Update policy
    logger.info(f"Granted {role} to {member} on {bucket_name}")
    print(f"Policy bindings: {dict(policy)}")  # Debug

# Example usage
set_iam_policy("my-data-lake-bucket", "user:[email protected]", "roles/storage.objectViewer")

# Expected Output:
# INFO:__main__:Granted roles/storage.objectViewer to user:[email protected] on my-data-lake-bucket
# Policy bindings: {'roles/storage.objectViewer': {'user:[email protected]'}}

from google.cloud import storage  # Import GCS client
from typing import Dict, Any  # For type annotations
import logging  # For logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def enable_versioning(bucket_name: str) -> None:
    """Enable versioning on bucket."""
    client = storage.Client()  # Initialize client
    bucket = client.get_bucket(bucket_name)  # Get bucket
    bucket.versioning_enabled = True  # Enable versioning
    bucket.patch()  # Update bucket
    logger.info(f"Enabled versioning on {bucket_name}")
    print(f"Versioning enabled: {bucket.versioning_enabled}")  # Debug

# Example usage
enable_versioning("my-data-lake-bucket")

# Expected Output:
# INFO:__main__:Enabled versioning on my-data-lake-bucket
# Versioning enabled: True

bucket_name: 'my-data-lake-bucket'
lifecycle_days: 365
iam_member: 'user:[email protected]'
iam_role: 'roles/storage.objectViewer'
versioning_enabled: true
max_quantity: 100
min_price: 10.0
required_fields:
  - transaction_id
  - product
  - price
  - quantity
  - date
product_prefix: 'Halal'
max_decimals: 2

def upload_and_verify_versioning(bucket: storage.Bucket, file_path: str, destination_path: str) -> None:
    """Upload file and verify versioning."""
    blob = bucket.blob(destination_path)
    blob.upload_from_filename(file_path)
    logger.info(f"Uploaded {file_path} to {destination_path}")
    versions = list(bucket.list_blobs(versions=True, prefix=destination_path))
    print(f"Versions for {destination_path}: {len(versions)}")
    if len(versions) >= 2:
        latest = bucket.get_blob(destination_path, generation=versions[0].generation)
        previous = bucket.get_blob(destination_path, generation=versions[1].generation)
        latest_content = latest.download_as_text()
        previous_content = previous.download_as_text()
        print(f"Latest version matches previous: {latest_content == previous_content}")

# File: de-onboarding/utils.py (extended from Chapter 34)
from typing import Dict, Any  # For type annotations
import logging  # For logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def clean_string(s: str) -> str:
    """Strip whitespace from string."""
    return s.strip()

def is_numeric(s: str, max_decimals: int = 2) -> bool:
    """Check if string is a decimal number with up to max_decimals."""
    parts = s.split(".")
    if len(parts) != 2 or not parts[0].replace("-", "").isdigit() or not parts[1].isdigit():
        return False
    return len(parts[1]) <= max_decimals

def is_numeric_value(x: Any) -> bool:
    """Check if value is numeric."""
    return isinstance(x, (int, float))

def has_valid_decimals(x: Any, max_decimals: int) -> bool:
    """Check if value has valid decimal places."""
    return is_numeric(str(x), max_decimals)

def apply_valid_decimals(x: Any, max_decimals: int) -> bool:
    """Apply has_valid_decimals to a value."""
    return has_valid_decimals(x, max_decimals)

def is_integer(x: Any) -> bool:
    """Check if value is an integer when converted to string."""
    return str(x).isdigit()

def validate_transaction(transaction: Dict[str, Any], config: Dict[str, Any]) -> bool:
    """Validate transaction based on config rules."""
    required_fields = config["required_fields"]
    min_price = config["min_price"]
    max_quantity = config["max_quantity"]
    prefix = config["product_prefix"]
    max_decimals = config["max_decimals"]

    logger.info(f"Validating transaction: {transaction}")
    for field in required_fields:
        if field not in transaction or not transaction[field] or clean_string(str(transaction[field])) == "":
            logger.warning(f"Invalid transaction: missing {field}: {transaction}")
            return False

    product = clean_string(str(transaction.get("product", "")))
    if product and not product.startswith(prefix):
        logger.warning(f"Invalid transaction: product lacks '{prefix}' prefix: {transaction}")
        return False

    price = str(transaction.get("price", ""))
    if not is_numeric(price, max_decimals) or float(price) < min_price or float(price) <= 0:
        logger.warning(f"Invalid transaction: invalid price: {transaction}")
        return False

    quantity = str(transaction.get("quantity", ""))
    if not quantity.isdigit() or int(quantity) > max_quantity:
        logger.warning(f"Invalid transaction: invalid quantity: {transaction}")
        return False

    return True

# File: de-onboarding/data_lake.py
from google.cloud import storage  # Import GCS client
import yaml  # For YAML parsing
import pandas as pd  # For CSV handling
from typing import Dict, Any, Tuple  # For type annotations
import logging  # For logging
import os  # For file existence check
import utils  # Import utils module

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def read_config(config_path: str) -> Dict[str, Any]:
    """Read YAML configuration."""
    logger.info(f"Opening config: {config_path}")
    with open(config_path, "r") as file:
        config = yaml.safe_load(file)
    logger.info(f"Loaded config: {config}")
    return config

def validate_config(config: Dict[str, Any]) -> bool:
    """Validate configuration."""
    required_keys = ["bucket_name", "lifecycle_days", "iam_member", "iam_role", "versioning_enabled"]
    for key in required_keys:
        if key not in config:
            logger.error(f"Missing config key: {key}")
            return False
    if config["product_prefix"] != "Halal":
        logger.error(f"Invalid product_prefix: {config['product_prefix']}; must be 'Halal' for Sharia compliance")
        return False
    return True

def setup_bucket(bucket_name: str, lifecycle_days: int, iam_member: str, iam_role: str, versioning_enabled: bool) -> storage.Bucket:
    """Setup GCS bucket with advanced features."""
    client = storage.Client()
    try:
        bucket = client.get_bucket(bucket_name)
        logger.info(f"Bucket {bucket_name} exists")
    except:
        bucket = client.create_bucket(bucket_name)
        logger.info(f"Created bucket {bucket_name}")

    # Enable versioning
    if versioning_enabled:
        bucket.versioning_enabled = True
        bucket.patch()
        logger.info(f"Enabled versioning on {bucket_name}")

    # Set lifecycle rule
    if lifecycle_days > 0:
        rule = {
            "action": {"type": "Delete"},
            "condition": {"age": lifecycle_days}
        }
        bucket.lifecycle_rules = [rule]
        bucket.patch()
        logger.info(f"Set lifecycle rule on {bucket_name} to delete after {lifecycle_days} days")
    else:
        logger.warning(f"Invalid lifecycle days {lifecycle_days}; skipping rule")

    # Set IAM policy
    valid_roles = ["roles/storage.objectViewer", "roles/storage.objectAdmin", "roles/storage.admin"]
    if iam_role in valid_roles:
        policy = bucket.get_iam_policy()
        policy[iam_role].add(iam_member)
        bucket.set_iam_policy(policy)
        logger.info(f"Granted {iam_role} to {iam_member} on {bucket_name}")
    else:
        logger.warning(f"Invalid IAM role {iam_role}; skipping policy update")

    return bucket

def upload_file(bucket: storage.Bucket, file_path: str, destination_path: str) -> None:
    """Upload file to GCS bucket."""
    blob = bucket.blob(destination_path)
    blob.upload_from_filename(file_path)
    logger.info(f"Uploaded {file_path} to {destination_path} in {bucket.name}")
    print(f"Uploaded file: {destination_path}")  # Debug

def process_data_lake(csv_path: str, config: Dict[str, Any]) -> Tuple[int, int]:
    """Process transaction data lake with advanced GCS features."""
    if not validate_config(config):
        logger.error("Invalid configuration")
        return 0, 0

    # Setup bucket
    bucket = setup_bucket(
        config["bucket_name"],
        config["lifecycle_days"],
        config["iam_member"],
        config["iam_role"],
        config["versioning_enabled"]
    )

    # Validate and upload CSV
    df = pd.read_csv(csv_path)
    valid_transactions = 0
    for _, row in df.iterrows():
        transaction = row.to_dict()
        if utils.validate_transaction(transaction, config):
            valid_transactions += 1

    destination_path = f"transactions/{os.path.basename(csv_path)}"
    upload_file(bucket, csv_path, destination_path)

    total_records = len(df)
    logger.info(f"Processed {valid_transactions} valid transactions out of {total_records}")
    return valid_transactions, total_records

def main() -> None:
    """Main function to enhance data lake."""
    csv_path = "data/transactions.csv"
    config_path = "data/config.yaml"

    config = read_config(config_path)
    valid_transactions, total_records = process_data_lake(csv_path, config)

    print("\nData Lake Report:")
    print(f"Total Records Processed: {total_records}")
    print(f"Valid Transactions: {valid_transactions}")
    print(f"Invalid Transactions: {total_records - valid_transactions}")
    print("Data lake processing completed")

if __name__ == "__main__":
    main()

# File: de-onboarding/tests/test_data_lake.py
import pytest
from google.cloud import storage
import yaml
import os
from data_lake import read_config, validate_config, setup_bucket, upload_file, process_data_lake
import utils

@pytest.fixture
def config():
    """Fixture for test configuration."""
    return {
        "bucket_name": "test-data-lake-bucket",
        "lifecycle_days": 365,
        "iam_member": "user:[email protected]",
        "iam_role": "roles/storage.objectViewer",
        "versioning_enabled": True,
        "max_quantity": 100,
        "min_price": 10.0,
        "required_fields": ["transaction_id", "product", "price", "quantity", "date"],
        "product_prefix": "Halal",
        "max_decimals": 2
    }

@pytest.fixture
def bucket():
    """Fixture for test bucket."""
    client = storage.Client()
    bucket_name = "test-data-lake-bucket"
    try:
        bucket = client.get_bucket(bucket_name)
    except:
        bucket = client.create_bucket(bucket_name)
    yield bucket
    # Cleanup: Delete bucket contents
    for blob in bucket.list_blobs():
        blob.delete()
    bucket.delete()

def test_read_config(tmp_path, config):
    """Test reading configuration."""
    config_path = tmp_path / "config.yaml"
    with open(config_path, "w") as f:
        yaml.dump(config, f)
    loaded_config = read_config(str(config_path))
    assert loaded_config == config

def test_validate_config(config):
    """Test configuration validation."""
    assert validate_config(config)
    invalid_config = config.copy()
    del invalid_config["bucket_name"]
    assert not validate_config(invalid_config)
    invalid_config = config.copy()
    invalid_config["product_prefix"] = "Non-Halal"
    assert not validate_config(invalid_config)

def test_setup_bucket(bucket, config):
    """Test bucket setup with advanced features."""
    bucket = setup_bucket(
        config["bucket_name"],
        config["lifecycle_days"],
        config["iam_member"],
        config["iam_role"],
        config["versioning_enabled"]
    )
    assert bucket.versioning_enabled
    assert any(rule["condition"]["age"] == config["lifecycle_days"] for rule in bucket.lifecycle_rules)
    policy = bucket.get_iam_policy()
    assert config["iam_member"] in policy[config["iam_role"]]

def test_invalid_lifecycle_days(bucket, config):
    """Test handling of invalid lifecycle days."""
    config["lifecycle_days"] = 0
    bucket = setup_bucket(
        config["bucket_name"],
        config["lifecycle_days"],
        config["iam_member"],
        config["iam_role"],
        config["versioning_enabled"]
    )
    assert not bucket.lifecycle_rules  # No rules set for invalid days

def test_invalid_iam_role(bucket, config):
    """Test handling of invalid IAM role."""
    config["iam_role"] = "roles/invalid.role"
    bucket = setup_bucket(
        config["bucket_name"],
        config["lifecycle_days"],
        config["iam_member"],
        config["iam_role"],
        config["versioning_enabled"]
    )
    policy = bucket.get_iam_policy()
    assert config["iam_member"] not in policy.get(config["iam_role"], set())  # Invalid role not applied

def test_upload_file(bucket, tmp_path):
    """Test file upload."""
    file_path = tmp_path / "test.csv"
    with open(file_path, "w") as f:
        f.write("test")
    upload_file(bucket, str(file_path), "test.csv")
    assert bucket.get_blob("test.csv") is not None

def test_validate_transaction_invalid_price(config):
    """Test transaction validation with invalid price."""
    transaction = {
        "transaction_id": "T006",
        "product": "Halal Laptop",
        "price": -999.99,
        "quantity": 2,
        "date": "2023-10-06"
    }
    assert not utils.validate_transaction(transaction, config)
    transaction["price"] = 999.99
    transaction["product"] = "Non-Halal Laptop"
    assert not utils.validate_transaction(transaction, config)

def test_process_data_lake(tmp_path, config):
    """Test data lake processing."""
    csv_path = tmp_path / "transactions.csv"
    with open(csv_path, "w") as f:
        f.write("transaction_id,product,price,quantity,date\n")
        f.write("T001,Halal Laptop,999.99,2,2023-10-01\n")
        f.write("T002,Mouse,24.99,10,2023-10-02\n")
    config_path = tmp_path / "config.yaml"
    with open(config_path, "w") as f:
        yaml.dump(config, f)
    valid_transactions, total_records = process_data_lake(str(csv_path), config)
    assert valid_transactions == 1
    assert total_records == 2

bucket_name = "test-bucket"
days = 180

bucket_name = "test-bucket"
member = "user:[email protected]"
role = "roles/storage.objectViewer"

bucket_name = "test-bucket"

from google.cloud import storage
from typing import Dict, Any

def setup_bucket(bucket_name: str, days: int) -> None:
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    rule = {"action": "Delete", "condition": days}  # Bug: Incorrect rule format
    bucket.lifecycle_rules = rule
    bucket.patch()

bucket_name = "test-bucket"
days = 180
member = "user:[email protected]"
role = "roles/storage.objectViewer"
config = {"product_prefix": "Halal"}

bucket_name = "test-bucket"

from google.cloud import storage
from typing import Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def set_lifecycle_rule(bucket_name: str, days: int) -> None:
    """Set lifecycle rule to delete files older than days."""
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    if days <= 0:
        logger.warning(f"Invalid lifecycle days {days}; skipping rule")
        return
    rule = {
        "action": {"type": "Delete"},
        "condition": {"age": days}
    }
    bucket.lifecycle_rules = [rule]
    bucket.patch()
    logger.info(f"Set lifecycle rule on {bucket_name} to delete after {days} days")

from google.cloud import storage
from typing import Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def set_iam_policy(bucket_name: str, member: str, role: str) -> None:
    """Set IAM policy to grant role to member."""
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    valid_roles = ["roles/storage.objectViewer", "roles/storage.objectAdmin", "roles/storage.admin"]
    if role not in valid_roles:
        logger.warning(f"Invalid IAM role {role}; skipping policy update")
        return
    policy = bucket.get_iam_policy()
    policy[role].add(member)
    bucket.set_iam_policy(policy)
    logger.info(f"Granted {role} to {member} on {bucket_name}")

from google.cloud import storage
from typing import Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def enable_versioning(bucket_name: str) -> None:
    """Enable versioning on bucket."""
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    bucket.versioning_enabled = True
    bucket.patch()
    logger.info(f"Enabled versioning on {bucket_name}")

from google.cloud import storage
from typing import Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def setup_bucket(bucket_name: str, days: int) -> None:
    """Set lifecycle rule for bucket."""
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    if days <= 0:
        logger.warning(f"Invalid lifecycle days {days}; skipping rule")
        return
    rule = {
        "action": {"type": "Delete"},  # Fix: Correct action format
        "condition": {"age": days}  # Fix: Correct condition format
    }
    bucket.lifecycle_rules = [rule]  # Fix: List of rules
    bucket.patch()
    logger.info(f"Set lifecycle rule on {bucket_name} to delete after {days} days")

from google.cloud import storage
from typing import Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def setup_lifecycle_and_iam(bucket_name: str, days: int, member: str, role: str, config: Dict[str, Any]) -> Dict[str, Any]:
    """Set lifecycle rule and IAM policy, return bucket config."""
    if config.get("product_prefix") != "Halal":
        logger.error("Invalid product_prefix; must be 'Halal' for Sharia compliance")
        return {}

    client = storage.Client()
    bucket = client.get_bucket(bucket_name)

    # Set lifecycle rule
    if days > 0:
        rule = {
            "action": {"type": "Delete"},
            "condition": {"age": days}
        }
        bucket.lifecycle_rules = [rule]
        bucket.patch()
        logger.info(f"Set lifecycle rule on {bucket_name} to delete after {days} days")
    else:
        logger.warning(f"Invalid lifecycle days {days}; skipping rule")

    # Set IAM policy
    valid_roles = ["roles/storage.objectViewer", "roles/storage.objectAdmin", "roles/storage.admin"]
    if role in valid_roles:
        policy = bucket.get_iam_policy()
        policy[role].add(member)
        bucket.set_iam_policy(policy)
        logger.info(f"Granted {role} to {member} on {bucket_name}")
    else:
        logger.warning(f"Invalid IAM role {role}; skipping policy update")

    # Return configuration
    config = {
        "versioning_enabled": bucket.versioning_enabled,
        "lifecycle_rules": list(bucket.lifecycle_rules)
    }
    return config

from google.cloud import storage
from typing import Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def analyze_versioning(bucket_name: str, output_path: str) -> None:
    """Check versioning status and write trade-offs explanation."""
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    versioning_status = bucket.versioning_enabled
    logger.info(f"Versioning status for {bucket_name}: {versioning_status}")

    explanation = (
        "Versioning in GCS allows recovery from accidental overwrites, critical for "
        "Hijra Group’s transaction data integrity. However, it increases storage costs, "
        "potentially doubling usage for frequent updates. Disable versioning for "
        "non-critical data to save costs.\n\n"
        "For Hijra Group’s transaction data, enable versioning for critical financial "
        "records to ensure data recovery, but disable it for temporary or archival "
        "data to optimize storage costs."
    )

    with open(output_path, "w") as f:
        f.write(explanation)
    logger.info(f"Explanation written to {output_path}")
    print(f"Versioning status: {versioning_status}")
    print(f"Explanation written to {output_path}")