33 - BigQuery and Google Sheets Integration

from typing import List  # For type annotations
import gspread  # For Google Sheets API
from google.oauth2.credentials import Credentials  # For OAuth2

# Authenticate with Google Sheets
creds = Credentials.from_authorized_user_file("credentials.json", scopes=[
    "https://www.googleapis.com/auth/spreadsheets",
    "https://www.googleapis.com/auth/drive"
])
client = gspread.authorize(creds)  # Create gspread client

# Open a Google Sheet by ID
sheet = client.open_by_key("YOUR_SHEET_ID")  # Replace with your Sheet ID
worksheet = sheet.get_worksheet(0)  # Access first worksheet

# Write data
data: List[List[str]] = [
    ["Product", "Total Sales"],
    ["Halal Laptop", "1999.98"],
    ["Halal Mouse", "249.90"]
]
worksheet.update("A1", data)  # Write data starting at A1

# Read data
values: List[List[str]] = worksheet.get_all_values()  # Read all values
print("Sheet Contents:")  # Debug
print(values)  # Print values

# Expected Output:
# Sheet Contents:
# [['Product', 'Total Sales'], ['Halal Laptop', '1999.98'], ['Halal Mouse', '249.90']]

from typing import List, Dict, Any  # For type annotations
from google.cloud import bigquery  # For BigQuery
import pandas as pd  # For DataFrame conversion

# Initialize BigQuery client
client = bigquery.Client.from_service_account_json("credentials.json")

# Query data mart
query = """
SELECT product, SUM(price * quantity) AS total_sales
FROM `your-project.sales_data_mart.sales`
GROUP BY product
"""
job = client.query(query)  # Run query
results: List[Dict[str, Any]] = [dict(row) for row in job.result()]  # Convert to list of dicts

# Convert to DataFrame
df = pd.DataFrame(results)
print("Query Results:")  # Debug
print(df)  # Print DataFrame

# Expected Output:
# Query Results:
#           product  total_sales
# 0   Halal Laptop      1999.98
# 1    Halal Mouse       249.90
# 2  Halal Keyboard       249.95

from typing import List, Dict, Any  # For type annotations
import gspread  # For Google Sheets
from google.cloud import bigquery  # For BigQuery
from google.oauth2.credentials import Credentials  # For OAuth2
import pandas as pd  # For DataFrame

# Authenticate and open sheet
creds = Credentials.from_authorized_user_file("credentials.json", scopes=[
    "https://www.googleapis.com/auth/spreadsheets",
    "https://www.googleapis.com/auth/drive"
])
client = gspread.authorize(creds)
sheet = client.open_by_key("YOUR_SHEET_ID")
worksheet = sheet.get_worksheet(0)

# Query BigQuery
bq_client = bigquery.Client.from_service_account_json("credentials.json")
query = """
SELECT product, SUM(price * quantity) AS total_sales
FROM `your-project.sales_data_mart.sales`
GROUP BY product
"""
job = bq_client.query(query)
results: List[Dict[str, Any]] = [dict(row) for row in job.result()]
df = pd.DataFrame(results)
print("Query Results:")  # Debug
print(df)

# Prepare data for Sheets
header: List[str] = df.columns.tolist()
data: List[List[Any]] = df.values.tolist()
sheet_data: List[List[Any]] = [header] + data
worksheet.update("A1", sheet_data)  # Write to sheet

# Read back for verification
values: List[List[str]] = worksheet.get_all_values()
print("Sheet Contents:")  # Debug
print(values)

# Expected Output:
# Query Results:
#           product  total_sales
# 0   Halal Laptop      1999.98
# 1    Halal Mouse       249.90
# 2  Halal Keyboard       249.95
# Sheet Contents:
# [['product', 'total_sales'], ['Halal Laptop', '1999.98'], ['Halal Mouse', '249.9'], ['Halal Keyboard', '249.95']]

# File: de-onboarding/test_sheets_export.py
from typing import List, Dict, Any
import gspread
from google.cloud import bigquery
from google.oauth2.credentials import Credentials
import pandas as pd
import pytest
import sheets_exporter
import tempfile
import os
import yaml

@pytest.fixture
def setup_sheets():
    creds_path = "credentials.json"
    sheet_id = "YOUR_SHEET_ID"
    dataset = "your-project.sales_data_mart"
    table = "sales"
    config_path = "data/config.yaml"
    return creds_path, sheet_id, dataset, table, config_path

def test_sheets_export(setup_sheets):
    creds_path, sheet_id, dataset, table, config_path = setup_sheets
    config = sheets_exporter.read_config(config_path)
    df = sheets_exporter.query_data_mart(creds_path, dataset, table)
    df = sheets_exporter.validate_data(df, config, dataset)

    creds = Credentials.from_authorized_user_file(creds_path, scopes=[
        "https://www.googleapis.com/auth/spreadsheets",
        "https://www.googleapis.com/auth/drive"
    ])
    client = gspread.authorize(creds)
    sheet = client.open_by_key(sheet_id)
    worksheet = sheet.get_worksheet(0)

    initial_sheets = len(sheet.worksheets())
    sheets_exporter.export_to_sheets(df, sheet_id, creds_path)
    values: List[List[str]] = worksheet.get_all_values()

    assert len(values) == len(df) + 1, "Incorrect number of rows"
    assert values[0] == df.columns.tolist(), "Header mismatch"
    assert len(sheet.worksheets()) == initial_sheets + 1, "Chart sheet not created"
    print("Test passed: Data and chart exported correctly")

def test_empty_query_result(setup_sheets):
    creds_path, sheet_id, dataset, table, config_path = setup_sheets
    config = sheets_exporter.read_config(config_path)

    # Simulate empty query result
    empty_df = pd.DataFrame(columns=["product", "total_sales"])
    validated_df = sheets_exporter.validate_data(empty_df, config, dataset)

    assert validated_df.empty, "Empty DataFrame not handled correctly"
    print("Test passed: Empty query result handled")

def test_invalid_config(setup_sheets):
    creds_path, sheet_id, dataset, table, _ = setup_sheets

    # Create temporary invalid config
    invalid_config = {
        "min_price": 10.0,
        "max_quantity": 100,
        "required_fields": ["product", "price", "quantity"],
        "max_decimals": 2
        # Missing product_prefix
    }
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as temp_file:
        yaml.safe_dump(invalid_config, temp_file)
        temp_config_path = temp_file.name

    try:
        config = sheets_exporter.read_config(temp_config_path)
        df = sheets_exporter.query_data_mart(creds_path, dataset, table)
        validated_df = sheets_exporter.validate_data(df, config, dataset)
        assert validated_df.empty, "Invalid config should result in empty DataFrame"
        print("Test passed: Invalid config handled")
    finally:
        os.unlink(temp_config_path)  # Clean up temporary file

# Simplified chart configuration for a bar chart
chart_config = {
    "requests": [{
        "addChart": {
            "chart": {
                "spec": {
                    "title": "Simple Sales Chart",
                    "basicChart": {
                        "chartType": "BAR",
                        "domains": [{"domain": {"sourceRange": {"sources": [{"startColumnIndex": 0}]}}}]  # X-axis: product
                        "series": [{"series": {"sourceRange": {"sources": [{"startColumnIndex": 1}]}}}]  # Y-axis: total_sales
                    }
                },
                "position": {"newSheet": True}
            }
        }
    }]
}
# Key fields:
# - domains: Defines the X-axis data (e.g., product names).
# - series: Defines the Y-axis data (e.g., sales amounts).
# - position: Places the chart on a new sheet.

from typing import List, Dict, Any
import gspread
from google.cloud import bigquery
from google.oauth2.credentials import Credentials
import pandas as pd

# Authenticate and open sheet
creds = Credentials.from_authorized_user_file("credentials.json", scopes=[
    "https://www.googleapis.com/auth/spreadsheets",
    "https://www.googleapis.com/auth/drive"
])
client = gspread.authorize(creds)
sheet = client.open_by_key("YOUR_SHEET_ID")
worksheet = sheet.get_worksheet(0)

# Query BigQuery
bq_client = bigquery.Client.from_service_account_json("credentials.json")
query = """
SELECT product, SUM(price * quantity) AS total_sales
FROM `your-project.sales_data_mart.sales`
GROUP BY product
"""
job = bq_client.query(query)
results: List[Dict[str, Any]] = [dict(row) for row in job.result()]
df = pd.DataFrame(results)

# Export to sheet
header: List[str] = df.columns.tolist()
data: List[List[Any]] = df.values.tolist()
sheet_data: List[List[Any]] = [header] + data
worksheet.update("A1", sheet_data)

# Add chart (bar chart)
chart_config = {
    "requests": [{
        "addChart": {
            "chart": {
                "spec": {
                    "title": "Sales by Product",
                    "basicChart": {
                        "chartType": "BAR",
                        "legendPosition": "RIGHT_LEGEND",
                        "domains": [{"domain": {"sourceRange": {"sources": [{"startRowIndex": 0, "endRowIndex": len(df) + 1, "startColumnIndex": 0, "endColumnIndex": 1}]}}}}],
                        "series": [{"series": {"sourceRange": {"sources": [{"startRowIndex": 0, "endRowIndex": len(df) + 1, "startColumnIndex": 1, "endColumnIndex": 2}]}}}}]
                    }
                },
                "position": {"newSheet": True}
            }
        }
    }]
}
sheet.batch_update(chart_config)
print("Chart added to new sheet")

# File: de-onboarding/setup_bigquery.py
from typing import List, Dict, Any
from google.cloud import bigquery
import pandas as pd

def setup_bigquery_table(creds_path: str, dataset_id: str, table_id: str, csv_path: str) -> None:
    """Create and seed BigQuery table with sales data."""
    client = bigquery.Client.from_service_account_json(creds_path)

    # Create dataset if not exists
    dataset_ref = client.dataset(dataset_id)
    dataset = bigquery.Dataset(dataset_ref)
    client.create_dataset(dataset, exists_ok=True)

    # Define schema
    schema = [
        bigquery.SchemaField("product", "STRING"),
        bigquery.SchemaField("price", "FLOAT"),
        bigquery.SchemaField("quantity", "INTEGER")
    ]

    # Create table
    table_ref = dataset_ref.table(table_id)
    table = bigquery.Table(table_ref, schema=schema)
    client.create_table(table, exists_ok=True)

    # Load data from CSV
    df = pd.read_csv(csv_path)
    job_config = bigquery.LoadJobConfig(
        schema=schema,
        skip_leading_rows=1,
        write_disposition="WRITE_TRUNCATE"
    )
    job = client.load_table_from_dataframe(df, table_ref, job_config=job_config)
    job.result()
    print(f"Loaded {job.output_rows} rows into {dataset_id}.{table_id}")

# Run setup
setup_bigquery_table(
    creds_path="credentials.json",
    dataset_id="your-project.sales_data_mart",
    table_id="sales",
    csv_path="data/sales.csv"
)

query = """
SELECT title AS product, COUNT(*) AS total_sales
FROM `bigquery-public-data.samples.wikipedia`
GROUP BY title
LIMIT 10
"""

min_price: 10.0
max_quantity: 100
required_fields:
  - product
  - price
  - quantity
product_prefix: 'Halal'
max_decimals: 2

# File: de-onboarding/utils.py (updated from Chapter 32)
from typing import Dict, Any, Union  # For type annotations

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(".")  # Split on decimal point
    if len(parts) != 2 or not parts[0].replace("-", "").isdigit() or not parts[1].isdigit():
        return False  # Invalid format
    return len(parts[1]) <= max_decimals  # Check decimal places

def clean_string(s: Union[str, Any]) -> str:
    """Strip whitespace from string."""
    return str(s).strip()

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

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."""
    try:
        return float(str(x)).is_integer()
    except (ValueError, TypeError):
        return False

def validate_sale(sale: Dict[str, Any], config: Dict[str, Any]) -> bool:
    """Validate sale 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"]

    print(f"Validating sale: {sale}")  # Debug
    for field in required_fields:
        if field not in sale or not sale[field] or clean_string(sale[field]) == "":
            print(f"Invalid sale: missing {field}: {sale}")
            return False

    product = clean_string(sale["product"])
    if not product.startswith(prefix):
        print(f"Invalid sale: product lacks '{prefix}' prefix: {sale}")
        return False

    price = sale["price"]
    if not is_numeric_value(price) or float(price) < min_price or float(price) <= 0:
        print(f"Invalid sale: invalid price: {sale}")
        return False

    if not has_valid_decimals(price, max_decimals):
        print(f"Invalid sale: too many decimals in price: {sale}")
        return False

    quantity = sale["quantity"]
    if not is_integer(quantity) or int(quantity) > max_quantity:
        print(f"Invalid sale: invalid quantity: {sale}")
        return False

    return True

# File: de-onboarding/sheets_exporter.py
from typing import List, Dict, Any  # For type annotations
import gspread
from google.cloud import bigquery
from google.oauth2.credentials import Credentials
import pandas as pd
import yaml
import os
from utils import is_numeric_value, apply_valid_decimals  # Import utils

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 query_data_mart(creds_path: str, dataset: str, table: str) -> pd.DataFrame:
    """Query BigQuery data mart."""
    client = bigquery.Client.from_service_account_json(creds_path)
    query = f"""
    SELECT product, SUM(price * quantity) AS total_sales
    FROM `{dataset}.{table}`
    GROUP BY product
    """
    print(f"Running query: {query}")  # Debug
    job = client.query(query)
    results: List[Dict[str, Any]] = [dict(row) for row in job.result()]
    df = pd.DataFrame(results)
    print("Query Results:")  # Debug
    print(df)
    return df

def validate_data(df: pd.DataFrame, config: Dict[str, Any], dataset: str = "") -> pd.DataFrame:
    """Validate DataFrame using config rules."""
    if df.empty:
        print("No data to validate")
        return df

    required_fields = config["required_fields"]
    missing_fields = [f for f in required_fields if f not in df.columns]
    if missing_fields:
        print(f"Missing columns: {missing_fields}")
        return pd.DataFrame()

    df = df.dropna(subset=["product"])
    if dataset != "bigquery-public-data.samples.wikipedia":
        if "product_prefix" not in config:
            print("Missing product_prefix in config")
            return pd.DataFrame()
        df = df[df["product"].str.startswith(config["product_prefix"])]
    df = df[df["total_sales"].apply(is_numeric_value)]
    df = df[df["total_sales"] >= config["min_price"]]
    df = df[df["total_sales"].apply(lambda x: apply_valid_decimals(x, config["max_decimals"]))]

    print("Validated DataFrame:")  # Debug
    print(df)
    return df

def export_to_sheets(df: pd.DataFrame, sheet_id: str, creds_path: str) -> None:
    """Export DataFrame to Google Sheet and add chart."""
    if df.empty:
        print("No data to export")
        return

    creds = Credentials.from_authorized_user_file(creds_path, scopes=[
        "https://www.googleapis.com/auth/spreadsheets",
        "https://www.googleapis.com/auth/drive"
    ])
    client = gspread.authorize(creds)
    sheet = client.open_by_key(sheet_id)
    worksheet = sheet.get_worksheet(0)

    header: List[str] = df.columns.tolist()
    data: List[List[Any]] = df.values.tolist()
    sheet_data: List[List[Any]] = [header] + data
    worksheet.update("A1", sheet_data)
    print("Data exported to sheet")

    chart_config = {
        "requests": [{
            "addChart": {
                "chart": {
                    "spec": {
                        "title": "Sales by Product",
                        "basicChart": {
                            "chartType": "BAR",
                            "legendPosition": "RIGHT_LEGEND",
                            "domains": [{"domain": {"sourceRange": {"sources": [{"startRowIndex": 0, "endRowIndex": len(df) + 1, "startColumnIndex": 0, "endColumnIndex": 1}]}}}}],
                            "series": [{"series": {"sourceRange": {"sources": [{"startRowIndex": 0, "endRowIndex": len(df) + 1, "startColumnIndex": 1, "endColumnIndex": 2}]}}}}]
                        }
                    },
                    "position": {"newSheet": True}
                }
            }
        }]
    }
    sheet.batch_update(chart_config)
    print("Chart added to new sheet")

def main() -> None:
    """Main function to export sales data to Google Sheets."""
    config_path = "data/config.yaml"
    creds_path = "credentials.json"
    sheet_id = "YOUR_SHEET_ID"
    dataset = "your-project.sales_data_mart"
    table = "sales"

    config = read_config(config_path)
    df = query_data_mart(creds_path, dataset, table)
    df = validate_data(df, config, dataset)
    export_to_sheets(df, sheet_id, creds_path)

    print("\nExport Report:")
    print(f"Records Processed: {len(df)}")
    print("Export completed")

if __name__ == "__main__":
    main()

# File: de-onboarding/test_sheets_exporter.py
from typing import List, Dict, Any
import gspread
from google.cloud import bigquery
from google.oauth2.credentials import Credentials
import pandas as pd
import pytest
import sheets_exporter
import tempfile
import os
import yaml

@pytest.fixture
def setup_sheets():
    creds_path = "credentials.json"
    sheet_id = "YOUR_SHEET_ID"
    dataset = "your-project.sales_data_mart"
    table = "sales"
    config_path = "data/config.yaml"
    return creds_path, sheet_id, dataset, table, config_path

def test_sheets_export(setup_sheets):
    creds_path, sheet_id, dataset, table, config_path = setup_sheets
    config = sheets_exporter.read_config(config_path)
    df = sheets_exporter.query_data_mart(creds_path, dataset, table)
    df = sheets_exporter.validate_data(df, config, dataset)

    creds = Credentials.from_authorized_user_file(creds_path, scopes=[
        "https://www.googleapis.com/auth/spreadsheets",
        "https://www.googleapis.com/auth/drive"
    ])
    client = gspread.authorize(creds)
    sheet = client.open_by_key(sheet_id)
    worksheet = sheet.get_worksheet(0)

    initial_sheets = len(sheet.worksheets())
    sheets_exporter.export_to_sheets(df, sheet_id, creds_path)
    values: List[List[str]] = worksheet.get_all_values()

    assert len(values) == len(df) + 1, "Incorrect number of rows"
    assert values[0] == df.columns.tolist(), "Header mismatch"
    assert len(sheet.worksheets()) == initial_sheets + 1, "Chart sheet not created"
    print("Test passed: Data and chart exported correctly")

def test_empty_query_result(setup_sheets):
    creds_path, sheet_id, dataset, table, config_path = setup_sheets
    config = sheets_exporter.read_config(config_path)

    # Simulate empty query result
    empty_df = pd.DataFrame(columns=["product", "total_sales"])
    validated_df = sheets_exporter.validate_data(empty_df, config, dataset)

    assert validated_df.empty, "Empty DataFrame not handled correctly"
    print("Test passed: Empty query result handled")

def test_invalid_config(setup_sheets):
    creds_path, sheet_id, dataset, table, _ = setup_sheets

    # Create temporary invalid config
    invalid_config = {
        "min_price": 10.0,
        "max_quantity": 100,
        "required_fields": ["product", "price", "quantity"],
        "max_decimals": 2
        # Missing product_prefix
    }
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as temp_file:
        yaml.safe_dump(invalid_config, temp_file)
        temp_config_path = temp_file.name

    try:
        config = sheets_exporter.read_config(temp_config_path)
        df = sheets_exporter.query_data_mart(creds_path, dataset, table)
        validated_df = sheets_exporter.validate_data(df, config, dataset)
        assert validated_df.empty, "Invalid config should result in empty DataFrame"
        print("Test passed: Invalid config handled")
    finally:
        os.unlink(temp_config_path)  # Clean up temporary file

creds_path = "credentials.json"
dataset = "your-project.sales_data_mart"
table = "sales"

df = pd.DataFrame({"product": ["Halal Laptop", "Halal Mouse"], "total_sales": [1999.98, 249.90]})
sheet_id = "YOUR_SHEET_ID"
creds_path = "credentials.json"

df = pd.DataFrame({"product": ["Halal Laptop", "Halal Mouse"], "total_sales": [1999.98, 249.90]})
sheet_id = "YOUR_SHEET_ID"
creds_path = "credentials.json"

from typing import List
import gspread
from google.oauth2.credentials import Credentials
import pandas as pd

def export_to_sheets(df: pd.DataFrame, sheet_id: str, creds_path: str) -> None:
    creds = Credentials.from_authorized_user_file(creds_path, scopes=[
        "https://www.googleapis.com/auth/spreadsheets"
    ])
    client = gspread.authorize(creds)
    sheet = client.open_by_key(sheet_id)
    worksheet = sheet.get_worksheet(0)
    header: List[str] = ["Product", "Sales"]  # Bug: Hardcoded headers
    data: List[List[Any]] = df.values.tolist()
    sheet_data: List[List[Any]] = [header] + data
    worksheet.update("A1", sheet_data)

df = pd.DataFrame({"product": ["Halal Laptop"], "total_sales": [1999.98]})
export_to_sheets(df, "YOUR_SHEET_ID", "credentials.json")

df = pd.DataFrame(columns=["product", "total_sales"])
sheet_id = "YOUR_SHEET_ID"
creds_path = "credentials.json"

from typing import List, Dict, Any
from google.cloud import bigquery
import pandas as pd

def query_data_mart(creds_path: str, dataset: str, table: str) -> pd.DataFrame:
    """Query BigQuery data mart."""
    client = bigquery.Client.from_service_account_json(creds_path)
    query = f"""
    SELECT product, SUM(price * quantity) AS total_sales
    FROM `{dataset}.{table}`
    GROUP BY product
    """
    print(f"Running query: {query}")
    job = client.query(query)
    results: List[Dict[str, Any]] = [dict(row) for row in job.result()]
    df = pd.DataFrame(results)
    print("Query Results:")
    print(df)
    return df

# Test
print(query_data_mart("credentials.json", "your-project.sales_data_mart", "sales"))

from typing import List, Any
import gspread
from google.oauth2.credentials import Credentials
import pandas as pd

def export_to_sheets(df: pd.DataFrame, sheet_id: str, creds_path: str) -> None:
    """Export DataFrame to Google Sheet."""
    if df.empty:
        print("No data to export")
        return
    creds = Credentials.from_authorized_user_file(creds_path, scopes=[
        "https://www.googleapis.com/auth/spreadsheets"
    ])
    client = gspread.authorize(creds)
    sheet = client.open_by_key(sheet_id)
    worksheet = sheet.get_worksheet(0)
    header: List[str] = df.columns.tolist()
    data: List[List[Any]] = df.values.tolist()
    sheet_data: List[List[Any]] = [header] + data
    worksheet.update("A1", sheet_data)
    print("Data exported to sheet")

# Test
df = pd.DataFrame({"product": ["Halal Laptop"], "total_sales": [1999.98]})
export_to_sheets(df, "YOUR_SHEET_ID", "credentials.json")

from typing import List, Any
import gspread
from google.oauth2.credentials import Credentials
import pandas as pd

def add_chart(df: pd.DataFrame, sheet_id: str, creds_path: str) -> None:
    """Add bar chart to Google Sheet."""
    if df.empty:
        print("No data for chart")
        return
    creds = Credentials.from_authorized_user_file(creds_path, scopes=[
        "https://www.googleapis.com/auth/spreadsheets"
    ])
    client = gspread.authorize(creds)
    sheet = client.open_by_key(sheet_id)
    chart_config = {
        "requests": [{
            "addChart": {
                "chart": {
                    "spec": {
                        "title": "Sales by Product",
                        "basicChart": {
                            "chartType": "BAR",
                            "legendPosition": "RIGHT_LEGEND",
                            "domains": [{"domain": {"sourceRange": {"sources": [{"startRowIndex": 0, "endRowIndex": len(df) + 1, "startColumnIndex": 0, "endColumnIndex": 1}]}}}}],
                            "series": [{"series": {"sourceRange": {"sources": [{"startRowIndex": 0, "endRowIndex": len(df) + 1, "startColumnIndex": 1, "endColumnIndex": 2}]}}}}]
                        }
                    },
                    "position": {"newSheet": True}
                }
            }
        }]
    }
    sheet.batch_update(chart_config)
    print("Chart added to new sheet")

# Test
df = pd.DataFrame({"product": ["Halal Laptop"], "total_sales": [1999.98]})
add_chart(df, "YOUR_SHEET_ID", "credentials.json")

from typing import List, Any
import gspread
from google.oauth2.credentials import Credentials
import pandas as pd

def export_to_sheets(df: pd.DataFrame, sheet_id: str, creds_path: str) -> None:
    """Export DataFrame to Google Sheet."""
    creds = Credentials.from_authorized_user_file(creds_path, scopes=[
        "https://www.googleapis.com/auth/spreadsheets"
    ])
    client = gspread.authorize(creds)
    sheet = client.open_by_key(sheet_id)
    worksheet = sheet.get_worksheet(0)
    header: List[str] = df.columns.tolist()  # Fix: Use DataFrame columns
    data: List[List[Any]] = df.values.tolist()
    sheet_data: List[List[Any]] = [header] + data
    worksheet.update("A1", sheet_data)
    print("Data exported to sheet")

# Test
df = pd.DataFrame({"product": ["Halal Laptop"], "total_sales": [1999.98]})
export_to_sheets(df, "YOUR_SHEET_ID", "credentials.json")

def analyze_performance() -> None:
    """Analyze performance trade-offs of gspread vs. CSV exports."""
    explanation = """
For Hijra Group’s stakeholder reporting, gspread uses the Google Sheets API, incurring network latency (O(n) for n cells due to HTTP requests, ~100ms per call) but enables real-time collaboration and programmatic chart creation. For example, exporting daily sales summaries to a shared Sheet allows stakeholders to view live updates and create dashboards. CSV exports involve file I/O (O(n) for n rows, ~10ms for small files) and are faster for local storage but lack collaboration features. gspread is preferred for cloud-based, interactive analytics like sales dashboards, while CSV is better for batch processing large datasets.
"""
    with open("ex5_concepts.txt", "w") as file:
        file.write(explanation.strip())
    print("Analysis saved to ex5_concepts.txt")

# Test
analyze_performance()

from typing import List, Any
import gspread
from google.oauth2.credentials import Credentials
import pandas as pd

def export_to_sheets(df: pd.DataFrame, sheet_id: str, creds_path: str) -> None:
    """Export DataFrame to Google Sheet, handling empty DataFrame."""
    creds = Credentials.from_authorized_user_file(creds_path, scopes=[
        "https://www.googleapis.com/auth/spreadsheets"
    ])
    client = gspread.authorize(creds)
    sheet = client.open_by_key(sheet_id)
    worksheet = sheet.get_worksheet(0)

    if df.empty:
        worksheet.update("A1", [["No sales data available"]])
        print("No data to export, writing message to sheet")
        return

    header: List[str] = df.columns.tolist()
    data: List[List[Any]] = df.values.tolist()
    sheet_data: List[List[Any]] = [header] + data
    worksheet.update("A1", sheet_data)
    print("Data exported to sheet")

# Test
df = pd.DataFrame(columns=["product", "total_sales"])
export_to_sheets(df, "YOUR_SHEET_ID", "credentials.json")