31 - Data Lakes with Google Cloud Storage

# File: de-onboarding/gcs_basics.py
from google.cloud import storage  # Import GCS client
from typing import Optional  # For type annotations

def create_bucket(project_id: str, bucket_name: str) -> Optional[storage.Bucket]:
    """Create a GCS bucket with type safety."""
    client = storage.Client(project=project_id)  # Initialize client
    bucket = client.bucket(bucket_name)  # Get bucket reference
    if not bucket.exists():  # Check if bucket exists
        bucket = client.create_bucket(bucket_name, location="us-central1")  # Create bucket
        print(f"Created bucket: {bucket_name}")  # Debug
    else:
        print(f"Bucket already exists: {bucket_name}")  # Debug
    return bucket  # Return bucket object

# Example usage (do not run without authentication)
project_id = "your-project-id"  # Replace with your project ID
bucket_name = "hijra-data-lake-unique"  # Unique bucket name
bucket = create_bucket(project_id, bucket_name)  # Create bucket
print(f"Bucket name: {bucket.name}")  # Debug

# Expected Output (if bucket is created):
# Created bucket: hijra-data-lake-unique
# Bucket name: hijra-data-lake-unique

# File: de-onboarding/gcs_upload.py
from google.cloud import storage  # Import GCS client
from typing import Optional  # For type annotations

def upload_file(bucket: storage.Bucket, source_path: str, destination_path: str) -> Optional[str]:
    """Upload a file to GCS with type safety."""
    blob = bucket.blob(destination_path)  # Create blob reference
    if not blob.exists():  # Check if blob exists
        blob.upload_from_filename(source_path)  # Upload file
        print(f"Uploaded {source_path} to {destination_path}")  # Debug
        return destination_path  # Return uploaded path
    else:
        print(f"File already exists: {destination_path}")  # Debug
    return None  # Return None if exists

# Example usage (do not run without authentication)
project_id = "your-project-id"  # Replace with your project ID
bucket_name = "hijra-data-lake-unique"  # Unique bucket name
client = storage.Client(project=project_id)  # Initialize client
bucket = client.bucket(bucket_name)  # Get bucket
source_path = "data/transactions.csv"  # Local file
destination_path = "transactions/transactions.csv"  # GCS path
uploaded = upload_file(bucket, source_path, destination_path)  # Upload
print(f"Uploaded path: {uploaded}")  # Debug

# Expected Output (if file is uploaded):
# Uploaded data/transactions.csv to transactions/transactions.csv
# Uploaded path: transactions/transactions.csv

# File: de-onboarding/gcs_pydantic.py
from google.cloud import storage  # Import GCS client
from pydantic import BaseModel, validator  # For validation
from typing import Optional  # For type annotations
import pandas as pd  # For CSV handling

class Transaction(BaseModel):
    """Pydantic model for transaction data."""
    transaction_id: str
    product: str
    price: float
    quantity: int
    date: str

    @validator("price")
    def price_positive(cls, v: float) -> float:
        """Ensure price is positive."""
        if v <= 0:
            raise ValueError("Price must be positive")
        return v

    @validator("quantity")
    def quantity_positive(cls, v: int) -> int:
        """Ensure quantity is positive."""
        if v <= 0:
            raise ValueError("Quantity must be positive")
        return v

    @validator("product")
    def product_not_empty(cls, v: str) -> str:
        """Ensure product is not empty."""
        if not v.strip():
            raise ValueError("Product cannot be empty")
        return v

def validate_and_upload(bucket: storage.Bucket, csv_path: str, destination_path: str) -> Optional[str]:
    """Validate CSV with Pydantic and upload to GCS."""
    df = pd.read_csv(csv_path)  # Load CSV
    print(f"Loaded CSV: {csv_path}")  # Debug
    print(df.head())  # Debug

    # Validate each row
    for _, row in df.iterrows():
        try:
            Transaction(**row.to_dict())  # Validate with Pydantic
        except ValueError as e:
            print(f"Invalid row: {row.to_dict()}, Error: {e}")  # Log error
            return None

    # Upload if valid
    blob = bucket.blob(destination_path)  # Create blob
    if not blob.exists():  # Check if exists
        blob.upload_from_filename(csv_path)  # Upload
        print(f"Uploaded {csv_path} to {destination_path}")  # Debug
        return destination_path
    print(f"File already exists: {destination_path}")  # Debug
    return None

# Example usage (do not run without authentication)
project_id = "your-project-id"  # Replace
bucket_name = "hijra-data-lake-unique"  # Replace
client = storage.Client(project=project_id)
bucket = client.bucket(bucket_name)
csv_path = "data/transactions.csv"
destination_path = "transactions/transactions.csv"
uploaded = validate_and_upload(bucket, csv_path, destination_path)
print(f"Uploaded path: {uploaded}")

# Expected Output (if valid):
# Loaded CSV: data/transactions.csv
#    transaction_id       product   price  quantity        date
# 0          T001  Halal Laptop  999.99         2  2023-10-01
# ...
# Invalid row: {'transaction_id': 'T004', 'product': '', ...}, Error: Product cannot be empty
# Uploaded path: None

gcs:
  project_id: 'your-project-id'
  bucket_name: 'hijra-data-lake-unique'
  destination_path: 'transactions/transactions.csv'
validation:
  min_price: 10.0
  max_quantity: 100
  required_fields:
    - transaction_id
    - product
    - price
    - quantity
    - date
  product_prefix: 'Halal'
  max_decimals: 2

# File: de-onboarding/verify_config.py
import yaml  # For YAML parsing
from typing import Dict, Any  # For type annotations

def verify_config(config_path: str) -> bool:
    """Verify config.yaml structure."""
    print(f"Verifying config: {config_path}")  # Debug
    with open(config_path, "r") as file:
        config = yaml.safe_load(file)
    required_keys = ["gcs", "validation"]
    gcs_keys = ["project_id", "bucket_name", "destination_path"]
    validation_keys = ["min_price", "max_quantity", "required_fields", "product_prefix", "max_decimals"]

    if not all(key in config for key in required_keys):
        print(f"Missing keys: {required_keys}")  # Debug
        return False
    if not all(key in config["gcs"] for key in gcs_keys):
        print(f"Missing GCS keys: {gcs_keys}")  # Debug
        return False
    if not all(key in config["validation"] for key in validation_keys):
        print(f"Missing validation keys: {validation_keys}")  # Debug
        return False

    print(f"Config verified: {config}")  # Debug
    return True

if __name__ == "__main__":
    config_path = "data/config.yaml"
    verify_config(config_path)

# File: de-onboarding/data_lake.py
from google.cloud import storage  # GCS client
from pydantic import BaseModel, validator  # For validation
from typing import Optional, Dict, Any  # For type annotations
import pandas as pd  # For CSV handling
import yaml  # For YAML parsing
import os  # For file checks

class Transaction(BaseModel):
    """Pydantic model for transaction data."""
    transaction_id: str
    product: str
    price: float
    quantity: int
    date: str

    @validator("price")
    def price_positive(cls, v: float) -> float:
        """Ensure price is positive."""
        if v <= 0:
            raise ValueError("Price must be positive")
        return v

    @validator("quantity")
    def quantity_positive(cls, v: int) -> int:
        """Ensure quantity is positive."""
        if v <= 0:
            raise ValueError("Quantity must be positive")
        return v

    @validator("product")
    def product_not_empty(cls, v: str) -> str:
        """Ensure product is not empty and has Halal prefix."""
        if not v.strip():
            raise ValueError("Product cannot be empty")
        if not v.startswith("Halal"):
            raise ValueError("Product must start with 'Halal'")
        return v

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

def create_bucket(project_id: str, bucket_name: str) -> Optional[storage.Bucket]:
    """Create a GCS bucket."""
    client = storage.Client(project=project_id)
    bucket = client.bucket(bucket_name)
    if not bucket.exists():
        bucket = client.create_bucket(bucket_name, location="us-central1")
        print(f"Created bucket: {bucket_name}")  # Debug
    else:
        print(f"Bucket already exists: {bucket_name}")  # Debug
    return bucket

def validate_and_upload(bucket: storage.Bucket, csv_path: str, destination_path: str) -> Optional[str]:
    """Validate CSV and upload to GCS."""
    if not os.path.exists(csv_path):
        print(f"File not found: {csv_path}")  # Debug
        return None
    df = pd.read_csv(csv_path)
    print(f"Loaded CSV: {csv_path}")  # Debug
    print(df.head())  # Debug

    for _, row in df.iterrows():
        try:
            Transaction(**row.to_dict())  # Validate
        except ValueError as e:
            print(f"Invalid row: {row.to_dict()}, Error: {e}")  # Log error
            return None

    blob = bucket.blob(destination_path)
    if not blob.exists():
        blob.upload_from_filename(csv_path)
        print(f"Uploaded {csv_path} to {destination_path}")  # Debug
        return destination_path
    print(f"File already exists: {destination_path}")  # Debug
    return None

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

    config = read_config(config_path)
    gcs_config = config["gcs"]
    project_id = gcs_config["project_id"]
    bucket_name = gcs_config["bucket_name"]
    destination_path = gcs_config["destination_path"]

    bucket = create_bucket(project_id, bucket_name)
    uploaded = validate_and_upload(bucket, csv_path, destination_path)

    print("\nData Lake Report:")  # Debug
    print(f"Bucket: {bucket_name}")
    print(f"Uploaded Path: {uploaded or 'None'}")
    print("Processing completed")  # Debug

if __name__ == "__main__":
    main()

# File: de-onboarding/test_data_lake.py
import pytest  # For testing
from google.cloud import storage  # GCS client
from data_lake import create_bucket, validate_and_upload, Transaction  # Import functions
import os  # For file checks
import pandas as pd  # For CSV creation

@pytest.fixture
def gcs_client():
    """Fixture for GCS client."""
    project_id = os.getenv("GOOGLE_CLOUD_PROJECT", "your-project-id")
    return storage.Client(project=project_id)

@pytest.fixture
def test_bucket(gcs_client):
    """Fixture for test bucket."""
    bucket_name = "hijra-test-lake-unique"  # Unique name
    bucket = create_bucket(gcs_client.project, bucket_name)
    yield bucket
    # Cleanup: Delete bucket if empty
    if bucket.exists():
        blobs = list(bucket.list_blobs())
        if not blobs:
            bucket.delete()
        else:
            print(f"Cleanup skipped: Bucket {bucket.name} contains {len(blobs)} blobs. Delete manually in GCS Console if needed.")

def test_create_bucket(gcs_client):
    """Test bucket creation."""
    bucket_name = "hijra-test-lake-unique2"
    bucket = create_bucket(gcs_client.project, bucket_name)
    assert bucket.exists()
    # Cleanup
    if bucket.exists():
        blobs = list(bucket.list_blobs())
        if not blobs:
            bucket.delete()
        else:
            print(f"Cleanup skipped: Bucket {bucket.name} contains {len(blobs)} blobs. Delete manually in GCS Console if needed.")

def test_validate_and_upload(test_bucket, tmp_path):
    """Test validation and upload."""
    # Note: Skipped cleanups are normal if test data persists. Delete buckets manually in GCS Console if needed.
    # Create valid CSV
    valid_csv = tmp_path / "valid.csv"
    df = pd.DataFrame([
        {"transaction_id": "T001", "product": "Halal Laptop", "price": 999.99, "quantity": 2, "date": "2023-10-01"},
        {"transaction_id": "T002", "product": "Halal Mouse", "price": 24.99, "quantity": 10, "date": "2023-10-02"}
    ])
    df.to_csv(valid_csv, index=False)

    destination = "test/valid.csv"
    uploaded = validate_and_upload(test_bucket, str(valid_csv), destination)
    assert uploaded == destination
    assert test_bucket.blob(destination).exists()

    # Test invalid CSV
    invalid_csv = tmp_path / "invalid.csv"
    df_invalid = pd.DataFrame([
        {"transaction_id": "T004", "product": "", "price": 29.99, "quantity": 3, "date": "2023-10-04"}
    ])
    df_invalid.to_csv(invalid_csv, index=False)

    uploaded = validate_and_upload(test_bucket, str(invalid_csv), "test/invalid.csv")
    assert uploaded is None

from pydantic import BaseModel
class Transaction(BaseModel):
    transaction_id: str
    product: str
    price: float
    quantity: int
    date: str

import pytest
from google.cloud import storage
from data_lake import create_bucket, validate_and_upload
import pandas as pd

def test_gcs_operations(tmp_path):
    project_id = "your-project-id"  # Replace
    bucket_name = "hijra-test-lake-unique3"
    # Add test logic for bucket creation and upload

import time
def measure_access_times(project_id: str, bucket_name: str, object_path: str, local_path: str) -> None:
    start_time = time.time()
    # Add GCS access code here
    gcs_time = time.time() - start_time
    # Add local access code here

from google.cloud import storage
from typing import Optional

def create_bucket(project_id: str, bucket_name: str) -> Optional[storage.Bucket]:
    """Create a GCS bucket."""
    client = storage.Client(project=project_id)
    bucket = client.bucket(bucket_name)
    if not bucket.exists():
        bucket = client.create_bucket(bucket_name, location="us-central1")
        print(f"Created bucket: {bucket_name}")
    else:
        print(f"Bucket already exists: {bucket_name}")
    return bucket

project_id = "your-project-id"
bucket_name = "hijra-test-lake"
bucket = create_bucket(project_id, bucket_name)
print(f"Bucket name: {bucket.name}")

from google.cloud import storage
from typing import Optional

def upload_file(bucket: storage.Bucket, source_path: str, destination_path: str) -> Optional[str]:
    """Upload a file to GCS."""
    blob = bucket.blob(destination_path)
    if not blob.exists():
        blob.upload_from_filename(source_path)
        print(f"Uploaded {source_path} to {destination_path}")
        return destination_path
    print(f"File already exists: {destination_path}")
    return None

project_id = "your-project-id"
bucket_name = "hijra-test-lake"
client = storage.Client(project=project_id)
bucket = client.bucket(bucket_name)
source_path = "data/transactions.csv"
destination_path = "transactions/test.csv"
uploaded = upload_file(bucket, source_path, destination_path)
print(f"Uploaded path: {uploaded}")

from pydantic import BaseModel, validator
import pandas as pd

class Transaction(BaseModel):
    transaction_id: str
    product: str
    price: float
    quantity: int
    date: str

    @validator("product")
    def product_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Product cannot be empty")
        return v

    @validator("price")
    def price_positive(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("Price must be positive")
        return v

def validate_csv(csv_path: str) -> int:
    """Validate CSV with Pydantic."""
    df = pd.read_csv(csv_path)
    # Add a negative price row for testing
    negative_row = pd.DataFrame([{
        "transaction_id": "T006",
        "product": "Halal Monitor",
        "price": -199.99,
        "quantity": 2,
        "date": "2023-10-06"
    }])
    df = pd.concat([df, negative_row], ignore_index=True)

    valid_count = 0
    for _, row in df.iterrows():
        try:
            Transaction(**row.to_dict())
            valid_count += 1
        except ValueError as e:
            print(f"Invalid row: {row.to_dict()}, Error: {e}")
    print(f"Valid rows: {valid_count}")
    return valid_count

validate_csv("data/transactions.csv")

import pytest
from google.cloud import storage
from data_lake import create_bucket, validate_and_upload
import pandas as pd

def test_gcs_operations(tmp_path):
    project_id = "your-project-id"  # Replace
    bucket_name = "hijra-test-lake-unique3"

    # Test bucket creation
    bucket = create_bucket(project_id, bucket_name)
    assert bucket.exists()

    # Test file upload
    csv_path = tmp_path / "test.csv"
    df = pd.DataFrame([
        {"transaction_id": "T001", "product": "Halal Laptop", "price": 999.99, "quantity": 2, "date": "2023-10-01"}
    ])
    df.to_csv(csv_path, index=False)

    destination = "test/test.csv"
    uploaded = validate_and_upload(bucket, str(csv_path), destination)
    assert uploaded == destination
    assert bucket.blob(destination).exists()

    # Cleanup
    blobs = list(bucket.list_blobs())
    if not blobs:
        bucket.delete()
    else:
        print(f"Cleanup skipped: Bucket {bucket.name} contains {len(blobs)} blobs. Delete manually in GCS Console if needed.")

# File: de-onboarding/ex5_concepts.py
from google.cloud import storage
import time
import os

def measure_access_times(project_id: str, bucket_name: str, object_path: str, local_path: str) -> None:
    """Measure GCS vs. local file access time."""
    client = storage.Client(project=project_id)
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(object_path)

    # Measure GCS access time
    start_time = time.time()
    blob.download_to_filename("temp.csv")
    gcs_time = time.time() - start_time
    os.remove("temp.csv")

    # Measure local access time
    start_time = time.time()
    with open(local_path, "r") as file:
        file.read()
    local_time = time.time() - start_time

    # Save metrics
    with open("ex5_metrics.txt", "w") as file:
        file.write(f"GCS Access Time: {gcs_time:.3f} seconds\n")
        file.write(f"Local Access Time: {local_time:.3f} seconds\n")
        file.write("Note: GCS times may vary (0.1–1 seconds) due to network latency.\n")

    # Save conceptual analysis
    explanation = (
        "GCS’s distributed key-value store enables O(1) access by mapping object names to storage locations "
        "across global data centers, using replication for durability. Local file systems rely on hierarchical "
        "directories, with O(log n) access for large directories due to tree traversal. GCS is ideal for scalable "
        "data lakes, while local systems suit small, structured data."
    )
    with open("ex5_concepts.txt", "w") as file:
        file.write(explanation)

    print("Saved to ex5_concepts.txt and ex5_metrics.txt")
    print(f"Note: GCS access time ({gcs_time:.3f}s) may vary due to network latency.")

project_id = "your-project-id"
bucket_name = "hijra-test-lake"
object_path = "transactions/test.csv"
local_path = "data/transactions.csv"
measure_access_times(project_id, bucket_name, object_path, local_path)