# Usually installed with pip:
# pip install black

# Let's create a simple script with inconsistent formatting
from typing import List, Dict, Any
import os
import random

def process_data(data_files: List[str],output_dir:str="./output") -> Dict[str, Any]:
    """Process multiple data files and return statistics."""
    if not os.path.exists(output_dir):
          os.makedirs(output_dir)  # Extra indentation here

    results = {"processed_files": 0,"total_records":0, "errors": 0}

    for file in     data_files:
        try:
            with open(file, 'r') as f:
                lines = f.readlines()

                # Process each line
                for line in lines:
                    results["total_records"] += 1
                    # Imagine more processing here

            results["processed_files"] += 1
        except Exception as e:
            results[ "errors" ] += 1
            print(f"Error processing {file}: {e}")

    # Print results
    print(f"Processed {results['processed_files']} files with {results['errors']} errors")
    print(f"Total records: {results['total_records']}")

    return     results

# Example usage
sample_files = ["data1.txt", "data2.txt"]
stats = process_data(sample_files)
print(f"Stats: {stats}")

# Would print (assuming files don't exist):
# Error processing data1.txt: [Errno 2] No such file or directory: 'data1.txt'
# Error processing data2.txt: [Errno 2] No such file or directory: 'data2.txt'
# Processed 0 files with 2 errors
# Total records: 0
# Stats: {'processed_files': 0, 'total_records': 0, 'errors': 2}

from typing import List, Dict, TypedDict, Union, Optional
import os
import random


class ResultStats(TypedDict):
    """Type definition for the result statistics."""
    processed_files: int
    total_records: int
    errors: int


def process_data(data_files: List[str], output_dir: str = "./output") -> ResultStats:
    """Process multiple data files and return statistics."""
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)  # Indentation fixed

    results: ResultStats = {"processed_files": 0, "total_records": 0, "errors": 0}

    for file in data_files:
        try:
            with open(file, "r") as f:
                lines = f.readlines()

                # Process each line
                for line in lines:
                    results["total_records"] += 1
                    # Imagine more processing here

            results["processed_files"] += 1
        except Exception as e:
            results["errors"] += 1
            print(f"Error processing {file}: {e}")

    # Print results
    print(f"Processed {results['processed_files']} files with {results['errors']} errors")
    print(f"Total records: {results['total_records']}")

    return results


# Example usage
sample_files = ["data1.txt", "data2.txt"]
stats = process_data(sample_files)
print(f"Stats: {stats}")

# Would print (assuming files don't exist):
# Error processing data1.txt: [Errno 2] No such file or directory: 'data1.txt'
# Error processing data2.txt: [Errno 2] No such file or directory: 'data2.txt'
# Processed 0 files with 2 errors
# Total records: 0
# Stats: {'processed_files': 0, 'total_records': 0, 'errors': 2}

from typing import List, Dict, Any
import os
import sys
import json  # Unused import
import random

def read_data_file(filename: str) -> List[Dict[str, Any]]:
    """Read data from a JSON file."""
    with open(filename, 'r') as f:
        data = json.load(f)

    # Return without proper error handling
    return data

def analyze_data(data: List[Dict[str, Any]], min_value: int = 0) -> Dict[str, Any]:
    """Analyze the data and return statistics."""
    total = 0
    count = 0

    # Iterate through the data
    for item in data:
        value = item.get('value', 0)

        # Dead code - value is already treated as 0 if missing
        if 'value' not in item:
            value = 0

        if value > min_value:
            total += value
            count += 1

    # Potential division by zero
    average = total / count

    results = {
        'total': total,
        'count': count,
        'average': average
    }

    print(f"Analysis results: {results}")
    return results

# Create test data
test_data = [
    {"id": 1, "value": 10},
    {"id": 2, "value": 5},
    {"id": 3, "value": 15},
    {"id": 4}  # Missing value
]

# Write test data to a file
with open("test_data.json", "w") as f:
    json.dump(test_data, f)

# Read and analyze the data
data = read_data_file("test_data.json")
stats = analyze_data(data, min_value=6)
print(f"Final stats: {stats}")

# Would print:
# Analysis results: {'total': 25, 'count': 2, 'average': 12.5}
# Final stats: {'total': 25, 'count': 2, 'average': 12.5}

from typing import List, Dict, TypedDict, Optional, Union, Literal
import os
import json

class DataItem(TypedDict, total=False):
    """Type definition for an item in our dataset."""
    id: int
    value: Union[int, float]


def read_data_file(filename: str) -> Optional[List[DataItem]]:
    """Read data from a JSON file.

    Args:
        filename: Path to the JSON file

    Returns:
        List of data items or None if an error occurs
    """
    try:
        with open(filename, 'r') as f:
            data = json.load(f)
        return data
    except (IOError, json.JSONDecodeError) as e:
        print(f"Error reading file {filename}: {e}")
        return None


class AnalysisResult(TypedDict):
    """Type definition for analysis results."""
    total: int
    count: int
    average: float


def analyze_data(data: List[DataItem], min_value: int = 0) -> AnalysisResult:
    """Analyze the data and return statistics.

    Args:
        data: List of data items to analyze
        min_value: Minimum value threshold for inclusion

    Returns:
        Dictionary with analysis results
    """
    total = 0
    count = 0

    # Iterate through the data
    for item in data:
        value = item.get('value', 0)

        if value > min_value:
            total += value
            count += 1

    # Handle the case where no items meet the criteria
    results: AnalysisResult = {
        'total': total,
        'count': count,
        'average': total / count if count > 0 else 0
    }

    print(f"Analysis results: {results}")
    return results

# Create test data
test_data = [
    {"id": 1, "value": 10},
    {"id": 2, "value": 5},
    {"id": 3, "value": 15},
    {"id": 4}  # Missing value
]

# Write test data to a file
try:
    with open("test_data.json", "w") as f:
        json.dump(test_data, f)
    print("Test data written to test_data.json")
except IOError as e:
    print(f"Error writing test data: {e}")
    exit(1)

# Read and analyze the data
data = read_data_file("test_data.json")
if data is not None:
    stats = analyze_data(data, min_value=6)
    print(f"Final stats: {stats}")
else:
    print("Analysis failed: could not read data")

# Would print:
# Test data written to test_data.json
# Analysis results: {'total': 25, 'count': 2, 'average': 12.5}
# Final stats: {'total': 25, 'count': 2, 'average': 12.5}

# Install Ruff
# pip install ruff

# Run Ruff on a file
ruff check your_file.py

# Run Ruff on a directory
ruff check your_project_directory/

ruff check --fix your_file.py

from typing import Dict, Any
import requests

# BAD - hardcoded API key in your code
def get_data_from_api_insecure() -> Dict[str, Any]:
    """Retrieve data from an external API."""
    api_key = "sk_test_abcdef1234567890"  # SECURITY RISK!
    url = "https://api.example.com/data"

    response = requests.get(
        url,
        headers={"Authorization": f"Bearer {api_key}"}
    )

    print(f"API response status: {response.status_code}")
    return response.json()

from typing import Dict, Union, Optional, TypedDict, Literal
import os
import requests
from typing import List

class WeatherResponse(TypedDict):
    """Type definition for weather API response."""
    temperature: float
    humidity: float
    conditions: str
    forecast: List[Dict[str, Union[str, float]]]  # Using Any here would be valid as external API response format may change


def get_data_from_api_secure() -> Optional[WeatherResponse]:
    """Retrieve data from an external API using securely stored credentials."""
    # Get API key from environment variable
    api_key = os.environ.get("API_KEY")

    if not api_key:
        print("Error: API_KEY environment variable not set")
        return None

    url = "https://api.example.com/data"

    try:
        response = requests.get(
            url,
            headers={"Authorization": f"Bearer {api_key}"}
        )
        response.raise_for_status()  # Raise exception for 4XX/5XX responses

        print(f"API response status: {response.status_code}")
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"API request failed: {e}")
        return None

# Usage (after setting environment variable):
# export API_KEY="your_api_key"
# data = get_data_from_api_secure()
# print(f"Retrieved data: {data}")

from typing import List, Dict, Any
import sqlite3

# BAD - vulnerable to SQL injection
def get_user_data_insecure(username: str) -> List[Dict[str, Any]]:
    """Retrieve user data based on username - VULNERABLE CODE."""
    conn = sqlite3.connect("users.db")
    cursor = conn.cursor()

    # DANGEROUS: Direct string interpolation in SQL
    query = f"SELECT * FROM users WHERE username = '{username}'"

    cursor.execute(query)
    rows = cursor.fetchall()

    # Convert rows to dictionaries
    columns = [desc[0] for desc in cursor.description]
    result = [dict(zip(columns, row)) for row in rows]

    conn.close()

    print(f"Retrieved {len(result)} user records")
    return result

# If called with username = "admin' OR '1'='1" this would return ALL users!

from typing import List, Dict, Optional, TypedDict, Union
import sqlite3


class UserRecord(TypedDict):
    """Type definition for user records from the database."""
    id: int
    username: str
    email: str
    created_at: str
    # Add other known fields here


def get_user_data_secure(username: str) -> Optional[List[UserRecord]]:
    """Retrieve user data based on username - SECURE CODE."""
    try:
        conn = sqlite3.connect("users.db")
        cursor = conn.cursor()

        # SECURE: Using parameterized query
        query = "SELECT * FROM users WHERE username = ?"

        cursor.execute(query, (username,))
        rows = cursor.fetchall()

        # Convert rows to dictionaries
        columns = [desc[0] for desc in cursor.description]
        result = []
        for row in rows:
            # Create a dictionary with proper typing
            user_dict = {}
            for i, col in enumerate(columns):
                user_dict[col] = row[i]
            result.append(user_dict)  # Type checker will validate this against UserRecord

        conn.close()

        print(f"Retrieved {len(result)} user records")
        return result
    except sqlite3.Error as e:
        print(f"Database error: {e}")
        return None

# Now safe from SQL injection

from typing import Optional
import os

def read_config_file_insecure(filename: str) -> Optional[str]:
    """Read a configuration file - INSECURE VERSION."""
    config_dir = "/app/configs/"

    # DANGEROUS: Path could be "../../../etc/passwd"
    file_path = os.path.join(config_dir, filename)

    try:
        with open(file_path, 'r') as f:
            content = f.read()

        print(f"Read configuration from {file_path}")
        return content
    except IOError as e:
        print(f"Error reading config: {e}")
        return None

from typing import Optional
import os

def read_config_file_secure(filename: str) -> Optional[str]:
    """Read a configuration file - SECURE VERSION."""
    config_dir = "/app/configs/"

    # Validate the filename to prevent directory traversal
    if not filename.isalnum() or not filename.endswith(".conf"):
        print(f"Invalid config filename: {filename}")
        return None

    file_path = os.path.join(config_dir, filename)

    # Further security: ensure the resolved path is within the config directory
    real_path = os.path.realpath(file_path)
    if not real_path.startswith(os.path.realpath(config_dir)):
        print(f"Security violation: attempted path traversal with {filename}")
        return None

    try:
        with open(file_path, 'r') as f:
            content = f.read()

        print(f"Read configuration from {file_path}")
        return content
    except IOError as e:
        print(f"Error reading config: {e}")
        return None

# Install pre-commit
# pip install pre-commit

# Create a .pre-commit-config.yaml file

repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.4.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml
      - id: check-added-large-files

  - repo: https://github.com/psf/black
    rev: 23.3.0
    hooks:
      - id: black

  - repo: https://github.com/charliermarsh/ruff-pre-commit
    rev: v0.0.262
    hooks:
      - id: ruff
        args: [--fix]

  - repo: https://github.com/RobertCraigie/pyright-python
    rev: v1.1.354
    hooks:
      - id: pyright

# Install the hooks
pre-commit install

from typing import Dict, List, Any
import json

def load_config(config_file: str) -> Dict[str, Any]:
    """Load configuration from a JSON file."""
    with open(config_file, 'r') as f:
        return json.load(f)  # Pre-commit would ensure proper error handling here

# Example usage
config = load_config('config.json')
print(f"Loaded config: {config}")

- repo: https://github.com/Yelp/detect-secrets
  rev: v1.4.0
  hooks:
    - id: detect-secrets

from typing import Dict, Any
import requests

def get_weather_data(city: str) -> Dict[str, Any]:
    """Get weather data for a city."""
    # This will be detected as a potential secret!
    api_key = "11a2b3c4d5e6f7g8h9i0j"

    url = f"https://api.weather.com/data?city={city}&key={api_key}"
    response = requests.get(url)

    print(f"Weather API response status: {response.status_code}")
    return response.json()

from typing import Dict, List, TypedDict, Optional, Union, Literal
import json

# Define specific types for our transformation data
class RenameTransformation(TypedDict):
    type: Literal['rename_fields']
    mapping: Dict[str, str]

class FilterTransformation(TypedDict):
    type: Literal['filter_values']
    field: str
    value: Union[str, int, float, bool]

# Union type for all possible transformation types
TransformationRules = Union[RenameTransformation, FilterTransformation]

# Input and output data types
class DataItem(TypedDict, total=False):
    """Type definition for data items with flexible fields."""
    id: int
    first_name: str
    last_name: str
    given_name: str  # After renaming
    family_name: str  # After renaming
    age: int

def transform_data(
    input_data: List[DataItem],
    transformation_rules: TransformationRules
) -> Optional[List[DataItem]]:
    """Transform a dataset according to specified rules.

    This function applies a set of transformation rules to each item in the
    input dataset. Transformations can include field renaming, value conversion,
    filtering, and aggregation.

    Args:
        input_data: A list of dictionaries representing the dataset to transform
        transformation_rules: A dictionary specifying the transformation rules
            to apply to the dataset. Must contain at least a 'type' field.

    Returns:
        A list of transformed data items, or None if the transformation fails

    Raises:
        ValueError: If transformation_rules is invalid
        KeyError: If a required field is missing from an input item

    Example:
        >>> rules = {'type': 'rename_fields', 'mapping': {'old_name': 'new_name'}}
        >>> data = [{'old_name': 'value', 'id': 1}]
        >>> transform_data(data, rules)
        [{'new_name': 'value', 'id': 1}]
    """
    if 'type' not in transformation_rules:
        print("Error: Invalid transformation rules")
        return None

    result: List[DataItem] = []

    try:
        # Apply transformations based on the rule type
        if transformation_rules['type'] == 'rename_fields':
            # We know this is a RenameTransformation due to the type check
            rename_rules = transformation_rules  # Type narrowing
            field_mapping = rename_rules.get('mapping', {})

            for item in input_data:
                new_item = item.copy()

                for old_field, new_field in field_mapping.items():
                    if old_field in new_item:
                        # Copy the value and remove the old field
                        value = new_item[old_field]
                        del new_item[old_field]
                        new_item[new_field] = value

                result.append(new_item)

        elif transformation_rules['type'] == 'filter_values':
            # We know this is a FilterTransformation due to the type check
            filter_rules = transformation_rules  # Type narrowing
            field = filter_rules.get('field', '')
            value = filter_rules.get('value')

            if not field or value is None:
                raise ValueError("Filter requires field and value")

            result = [item for item in input_data if item.get(field) == value]

        else:
            # This shouldn't happen due to our type definitions
            print(f"Unknown transformation type: {transformation_rules['type']}")
            return None

    except Exception as e:
        print(f"Error during transformation: {e}")
        return None

    print(f"Transformed {len(input_data)} items into {len(result)} items")
    return result

# Example usage
data = [
    {"id": 1, "first_name": "Alice", "last_name": "Smith", "age": 30},
    {"id": 2, "first_name": "Bob", "last_name": "Jones", "age": 25},
    {"id": 3, "first_name": "Charlie", "last_name": "Brown", "age": 35}
]

# Rename fields
rename_rules = {
    "type": "rename_fields",
    "mapping": {
        "first_name": "given_name",
        "last_name": "family_name"
    }
}

transformed = transform_data(data, rename_rules)
print(f"After renaming: {json.dumps(transformed, indent=2)}")

# Filter values
filter_rules = {
    "type": "filter_values",
    "field": "age",
    "value": 30
}

filtered = transform_data(data, filter_rules)
print(f"After filtering: {json.dumps(filtered, indent=2)}")

# Would print:
# Transformed 3 items into 3 items
# After renaming: [
#   {
#     "id": 1,
#     "given_name": "Alice",
#     "family_name": "Smith",
#     "age": 30
#   },
#   {
#     "id": 2,
#     "given_name": "Bob",
#     "family_name": "Jones",
#     "age": 25
#   },
#   {
#     "id": 3,
#     "given_name": "Charlie",
#     "family_name": "Brown",
#     "age": 35
#   }
# ]
# Transformed 3 items into 1 items
# After filtering: [
#   {
#     "id": 1,
#     "first_name": "Alice",
#     "last_name": "Smith",
#     "age": 30
#   }
# ]

# Sales Data Pipeline

A data pipeline that processes daily sales data and loads it into our data warehouse.

## Overview

This pipeline extracts sales data from our POS system, transforms it to conform to our data warehouse schema, and loads it into BigQuery for analysis.

## Features

- Automated daily extraction from POS API
- Data validation and error handling
- Incremental loading to optimize performance
- Logging and monitoring integration

## Prerequisites

- Python 3.8+
- Google Cloud SDK
- Access to POS API

## Installation

1. Clone this repository
2. Install dependencies: `pip install -r requirements.txt`
3. Configure environment variables (see Configuration section)

## Configuration

Create a `.env` file with the following variables:

from typing import Dict, Any, Optional
import os
import json

def get_database_config() -> Dict[str, Any]:
    """Get database configuration from environment variables."""
    return {
        "host": os.environ.get("DB_HOST", "localhost"),
        "port": int(os.environ.get("DB_PORT", "5432")),
        "database": os.environ.get("DB_NAME", ""),
        "user": os.environ.get("DB_USER", ""),
        "password": os.environ.get("DB_PASSWORD", "")
    }

# Example usage (after setting environment variables)
config = get_database_config()
print(f"Database config (without password): {config['host']}:{config['port']}/{config['database']}")

from typing import Dict, Any, Optional
import json
import os

# Load config from a file
def load_config(config_path: str = "config.json") -> Optional[Dict[str, Any]]:
    """Load configuration from a JSON file."""
    try:
        with open(config_path, 'r') as f:
            config = json.load(f)
        print(f"Loaded configuration from {config_path}")
        return config
    except Exception as e:
        print(f"Error loading configuration: {e}")
        return None

# Create template config file
def create_config_template() -> None:
    """Create a template configuration file."""
    template = {
        "database": {
            "host": "localhost",
            "port": 5432,
            "name": "your_database_name",
            "user": "your_username",
            "password": "your_password"
        },
        "api": {
            "url": "https://api.example.com",
            "key": "your_api_key"
        }
    }

    with open("config.template.json", 'w') as f:
        json.dump(template, f, indent=2)

    print("Created configuration template at config.template.json")
    print("Copy to config.json and fill in your actual values.")

# Example usage
create_config_template()

# Would create a file named config.template.json:
# {
#   "database": {
#     "host": "localhost",
#     "port": 5432,
#     "name": "your_database_name",
#     "user": "your_username",
#     "password": "your_password"
#   },
#   "api": {
#     "url": "https://api.example.com",
#     "key": "your_api_key"
#   }
# }

# Create a new directory for the project
mkdir data_quality_demo
cd data_quality_demo

# Initialize git repository
git init

# Create initial README
echo "# Data Quality Demo" > README.md
git add README.md
git commit -m "Initial commit"

# Create and switch to development branch
git checkout -b development

echo "black==23.3.0
ruff==0.0.262
pyright==1.1.354
pre-commit==3.3.1
detect-secrets==1.4.0" > requirements-dev.txt

repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.4.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml
      - id: check-added-large-files

  - repo: https://github.com/psf/black
    rev: 23.3.0
    hooks:
      - id: black

  - repo: https://github.com/charliermarsh/ruff-pre-commit
    rev: v0.0.262
    hooks:
      - id: ruff
        args: [--fix]

  - repo: https://github.com/RobertCraigie/pyright-python
    rev: v1.1.354
    hooks:
      - id: pyright

  - repo: https://github.com/Yelp/detect-secrets
    rev: v1.4.0
    hooks:
      - id: detect-secrets

[tool.black]
line-length = 88
target-version = ["py38"]
include = '\.pyi?

[tool.ruff]
line-length = 88
target-version = "py38"
select = [
    "E",  # pycodestyle errors
    "F",  # pyflakes
    "I",  # isort
    "N",  # pep8-naming
    "W",  # pycodestyle warnings
    "B",  # flake8-bugbear
    "S",  # flake8-bandit
]
ignore = []

[tool.pyright]
pythonVersion = "3.8"
typeCheckingMode = "strict"
reportMissingImports = true
reportMissingTypeStubs = false

mkdir -p docs/templates

# Pipeline: [Pipeline Name]

## Overview

Brief description of the pipeline's purpose and functionality.

## Data Sources

- **Source 1**

  - Type: [Database, API, File, etc.]
  - Location: [URL, server, path, etc.]
  - Schema: [Link to schema documentation or brief description]
  - Update Frequency: [Real-time, daily, weekly, etc.]

- **Source 2**
  - ...

## Transformations

Describe the key data transformations that occur in this pipeline:

1. **Transformation 1**

   - Purpose: [Why this transformation occurs]
   - Logic: [Description or pseudocode of the transformation]
   - Dependencies: [Any dependencies or preconditions]

2. **Transformation 2**
   - ...

## Output

- **Destination**
  - Type: [Database, File, API, etc.]
  - Location: [URL, server, path, etc.]
  - Schema: [Link to schema documentation or brief description]

## Scheduling

- Frequency: [How often the pipeline runs]
- Trigger: [Event or time-based]
- Dependencies: [Any pipelines that must complete first]

## Monitoring

- Key metrics to monitor
- Alert thresholds
- Links to dashboards

## Failure Recovery

Steps to take in case of pipeline failure.

# Schema: [Schema Name]

## Overview

Brief description of what this data represents and its purpose.

## Table/Collection Structure

### [Table/Collection Name]

| Field Name | Type      | Description                                       | Example                | Required? |
| ---------- | --------- | ------------------------------------------------- | ---------------------- | --------- |
| id         | int       | Unique identifier                                 | 12345                  | Yes       |
| name       | string    | User's full name                                  | "John Smith"           | Yes       |
| email      | string    | User's email address                              | "john@example.com"     | Yes       |
| created_at | timestamp | When the record was created                       | "2023-05-01T12:34:56Z" | Yes       |
| status     | enum      | Account status: "active", "inactive", "suspended" | "active"               | Yes       |

### [Another Table/Collection Name]

...

## Relationships

Describe relationships between tables/collections:

- Table A has a one-to-many relationship with Table B through the `table_a_id` field
- ...

## Constraints

List any constraints or business rules:

- Field X must be unique
- Field Y must be in range [0, 100]
- ...

## Versioning

Information about schema versioning and change management.

mkdir -p src/data_quality_demo/pipeline
mkdir -p src/data_quality_demo/util
mkdir -p tests
mkdir -p config

from typing import Dict, TypedDict, Optional, Union, List
import os
import json

# Define a type for database configuration
class DatabaseConfig(TypedDict):
    host: str
    port: int
    database: str
    user: str
    password: str


class ConfigManager:
    """Manage configuration for data pipelines.

    This class handles loading and validating configuration from different sources,
    with environment variable overrides and secure credential handling.
    """

    def __init__(self, config_path: Optional[str] = None) -> None:
        """Initialize the configuration manager.

        Args:
            config_path: Optional path to a configuration file
        """
        self.config: Dict[str, Union[str, int, bool, Dict[str, Union[str, int, bool]]]] = {}
        # Using a somewhat complex type above because configuration could have varying
        # structures. In a real-world project, you might define more specific TypedDict
        # classes for each section of your configuration.

        if config_path:
            self._load_from_file(config_path)

        # Override with environment variables
        self._load_from_env()

        print(f"Configuration initialized with {len(self.config)} settings")

    def _load_from_file(self, config_path: str) -> None:
        """Load configuration from a JSON file.

        Args:
            config_path: Path to the configuration file
        """
        try:
            if not os.path.exists(config_path):
                print(f"Warning: Config file not found: {config_path}")
                return

            with open(config_path, 'r') as f:
                file_config = json.load(f)
                # Here we could use TypedDict validations if we had
                # more specific schema definitions
                self.config.update(file_config)

            print(f"Loaded configuration from {config_path}")
        except Exception as e:
            print(f"Error loading configuration from {config_path}: {e}")

    def _load_from_env(self) -> None:
        """Override configuration with environment variables.

        Environment variables that start with 'APP_' will be
        added to the configuration, with the 'APP_' prefix removed.
        """
        for key, value in os.environ.items():
            if key.startswith("APP_"):
                config_key = key[4:].lower()
                self.config[config_key] = value

    def get(self, key: str, default: Optional[Union[str, int, bool, Dict[str, Union[str, int, bool]]]] = None) -> Optional[Union[str, int, bool, Dict[str, Union[str, int, bool]]]]:
        """Get a configuration value.

        Args:
            key: The configuration key to look up
            default: Value to return if the key is not found

        Returns:
            The configuration value or the default
        """
        return self.config.get(key, default)

    def get_required(self, key: str) -> Union[str, int, bool, Dict[str, Union[str, int, bool]]]:
        """Get a required configuration value.

        Args:
            key: The configuration key to look up

        Returns:
            The configuration value

        Raises:
            KeyError: If the key is not found in the configuration
        """
        if key not in self.config:
            raise KeyError(f"Required configuration key not found: {key}")

        return self.config[key]

    def get_database_config(self) -> DatabaseConfig:
        """Get database configuration as a strongly-typed object.

        Returns:
            Database configuration

        Raises:
            KeyError: If database configuration is not complete
        """
        db_section = self.get('database', {})
        if not isinstance(db_section, dict):
            raise TypeError("Database configuration is not a dictionary")

        # Create a properly typed database config, with defaults for non-critical fields
        return {
            'host': db_section.get('host', 'localhost'),
            'port': int(db_section.get('port', 5432)),
            'database': db_section.get('name', ''),
            'user': db_section.get('user', ''),
            'password': db_section.get('password', '')
        }

# Example usage
if __name__ == "__main__":
    # Create a template config file for demonstration
    template_config = {
        "database": {
            "host": "localhost",
            "port": 5432,
            "name": "demo_db"
        },
        "api": {
            "url": "https://api.example.com",
            "timeout": 30
        },
        "logging": {
            "level": "INFO",
            "file": "logs/app.log"
        }
    }

    with open("config/config.template.json", 'w') as f:
        json.dump(template_config, f, indent=2)

    print("Created template configuration file")

    # Test the config manager
    config = ConfigManager("config/config.template.json")

    # Get configuration values
    db_host = config.get("database", {}).get("host", "")
    api_url = config.get("api", {}).get("url", "")

    print(f"Database host: {db_host}")
    print(f"API URL: {api_url}")

# Data Quality Demo

A demonstration project showing best practices for data engineering code quality.

## Overview

This project showcases tools and practices for maintaining high-quality, secure, and maintainable data engineering code. It includes:

- Code formatting with Black
- Linting with Ruff
- Static type checking with Pyright
- Security scanning with detect-secrets
- Pre-commit hooks for automatic quality checks
- Documentation templates for data pipelines

## Setup

### Prerequisites

- Python 3.8+
- Git

### Installation

1. Clone this repository
2. Create a virtual environment: `python -m venv venv`
3. Activate the virtual environment:
   - Windows: `venv\Scripts\activate`
   - macOS/Linux: `source venv/bin/activate`
4. Install development dependencies: `pip install -r requirements-dev.txt`
5. Install pre-commit hooks: `pre-commit install`

## Development

### Code Quality Standards

This project enforces the following standards:

- Code formatting using Black's default style
- Linting using Ruff with selected rule sets
- Static type checking with Pyright in strict mode
- Secret detection to prevent credential leakage

### Documentation

- All code should include docstrings following the Google style
- Data pipelines should be documented using the template in `docs/templates/pipeline_template.md`
- Data schemas should be documented using the template in `docs/templates/schema_template.md`

### Git Workflow

1. Create a feature branch from `development`
2. Make your changes, ensuring they pass all pre-commit checks
3. Submit a pull request to merge into `development`
4. After review and testing, changes will be merged to `main` for release

## Contributing

Please see [CONTRIBUTING.md](CONTRIBUTING.md) for details on our code of conduct and the process for submitting pull requests.

# Contributing Guidelines

Thank you for your interest in contributing to this project! This document outlines the process for contributing and the standards we follow.

## Code Quality

All code must pass our automated quality checks:

1. **Formatting**: We use Black for code formatting
2. **Linting**: We use Ruff to check for common issues
3. **Type Checking**: All code must be properly typed and pass Pyright checks
4. **Security**: Code must not contain hardcoded secrets or credentials

These checks are enforced through pre-commit hooks that run automatically when you commit code.

## Development Process

1. **Create an Issue**: For new features or bug fixes, start by creating an issue
2. **Fork and Branch**: Fork the repository and create a branch for your work
3. **Write Code**: Implement your changes, following our code quality standards
4. **Write Tests**: Add tests for your code to ensure it works as expected
5. **Document**: Update documentation to reflect your changes
6. **Submit PR**: Create a pull request with a clear description of your changes

## Documentation Standards

- All functions and classes must have docstrings following Google style
- Complex logic should include inline comments explaining the reasoning
- Update README and other documentation as needed for significant changes
- New data pipelines should include documentation using our templates

## Git Practices

- Keep commits focused on single logical changes
- Write clear commit messages that explain the "why" not just the "what"
- Rebase your branch on the latest development branch before submitting PR

Thank you for contributing!

from typing import List,Dict,Any
import os
import  json

def load_data(  filename:str,
    required_fields:List[str]=[],
default_values:Dict[str,Any]={}  )->List[Dict[str,Any]]:
  """Load data from a JSON file."""
  if not os.path.exists(filename):
    return []
  with open(filename,'r') as f:
   data=json.load(f)

  # Ensure required fields and apply defaults
  result=[]
  for item in data:
    # Check required fields
    valid=True
    for field in required_fields:
        if field not in item:
            valid=False
            break
    if not valid:
      continue

    # Apply defaults
    for key,value in default_values.items():
      if key not in item:
          item[key]=value

    result.append(item)

  return     result

from typing import List, Dict, Any
import os
import json
import datetime
import requests

def fetch_and_save_data(api_url: str, output_file: str) -> None:
    # Fetch data from API
    response = requests.get(api_url)
    data = response.json()

    # Add timestamp
    current_time = datetime.datetime.now()
    data['timestamp'] = str(current_time)

    # Save to file
    with open(output_file, 'w') as f:
        f.write(json.dumps(data))

    print('Data saved to ' + output_file)

def process_items(items):
    results = []
    for i in range(len(items)):
        item = items[i]
        if item['status'] == 'active':
            value = item.get('value', 0) * 2
            results.append({'id': item['id'], 'processed_value': value})
        else:
            pass

    print(f"Processed {len(results)} items out of {len(items)}")
    return results

def main():
    # Hardcoded API key
    api_key = "ak_12345abcdef"

    # Construct URL
    url = "https://api.example.com/data?key=" + api_key

    fetch_and_save_data(url, 'data/output.json')

    # Open the file we just saved
    f = open('data/output.json', 'r')
    data = json.load(f)
    f.close()

    # Process items
    items = data.get('items', [])
    results = process_items(items)

    # Save results
    with open('data/results.json', 'w') as f:
        json.dump(results, f)

if __name__ == "__main__":
    main()

from typing import Dict, Any
import requests
import json

def get_user_data(user_id: str) -> Dict[str, Any]:
    """Get user data from the API."""
    # API credentials (insecure)
    username = "admin"
    password = "s3cret_p@ssw0rd"

    # Construct auth header
    auth_header = f"Basic {username}:{password}"

    # Make API request
    response = requests.get(
        f"https://api.example.com/users/{user_id}",
        headers={"Authorization": auth_header}
    )

    return response.json()

def save_config(config: Dict[str, Any], filename: str) -> None:
    """Save configuration to a file."""
    # Configuration includes database credentials
    config["database"] = {
        "host": "db.example.com",
        "username": "db_user",
        "password": "db_p@ssw0rd"
    }

    # Save to file
    with open(filename, 'w') as f:
        json.dump(config, f, indent=2)

    print(f"Configuration saved to {filename}")

# Example usage
user_data = get_user_data("12345")
print(f"User data: {user_data}")

config = {"app_name": "MyApp", "version": "1.0.0"}
save_config(config, "config.json")