Language Ecosystem

# pip - Python Package Index
pip install pandas numpy scikit-learn
pip install -r requirements.txt
pip freeze > requirements.txt
 
# Virtual environments
python -m venv data_env
source data_env/bin/activate  # Unix
data_env\Scripts\activate     # Windows
 
# conda - Comprehensive package manager
conda create -n data_science python=3.9
conda activate data_science
conda install pandas numpy matplotlib -c conda-forge

# npm - Node Package Manager
npm init -y
npm install express lodash moment
npm install --save-dev jest typescript @types/node
 
# Package scripts
npm run build
npm run test
npm start
 
# yarn - Alternative package manager
yarn add express lodash moment
yarn add --dev jest typescript
yarn build

# Cargo - Rust package manager
cargo new data_processor
cargo add tokio serde sqlx
cargo add --dev criterion  # Development dependency
 
# Building and running
cargo build
cargo run
cargo test
cargo bench
 
# Publishing
cargo publish

# Go modules
go mod init github.com/username/project
go get github.com/gin-gonic/gin
go get -u github.com/lib/pq  # Update dependency
 
# Building and running
go build
go run main.go
go test ./...
 
# Vendoring
go mod vendor

# Installing packages
install.packages(c("dplyr", "ggplot2", "tidyr"))
install.packages("devtools")
 
# Bioconductor packages
BiocManager::install("genomics_package")
 
# GitHub packages
devtools::install_github("username/package")
 
# Loading packages
library(dplyr)
library(ggplot2)
 
# Package management
packrat::init()  # Project-specific libraries
renv::init()     # Modern dependency management

# Popular Python IDEs and editors:
# - PyCharm: Full-featured IDE with debugging, profiling
# - VS Code: Lightweight with Python extensions
# - Jupyter: Interactive notebooks for data science
# - Spyder: Scientific Python IDE
 
# Jupyter notebook example
import pandas as pd
import matplotlib.pyplot as plt
 
# Inline plotting
%matplotlib inline
 
# Load and visualize data
df = pd.read_csv('data.csv')
df.plot(x='date', y='value')
plt.show()
 
# Interactive widgets
from ipywidgets import interact
 
@interact(multiplier=(0.1, 3.0, 0.1))
def plot_data(multiplier=1.0):
    plt.figure(figsize=(10, 6))
    plt.plot(df['date'], df['value'] * multiplier)
    plt.show()

// VS Code settings.json for JavaScript/TypeScript
{
    "editor.formatOnSave": true,
    "editor.codeActionsOnSave": {
        "source.fixAll.eslint": true
    },
    "typescript.preferences.importModuleSpecifier": "relative",
    "extensions.recommendations": [
        "ms-vscode.vscode-typescript-next",
        "esbenp.prettier-vscode",
        "ms-vscode.vscode-eslint"
    ]
}
 
// .eslintrc.js configuration
module.exports = {
    extends: [
        '@typescript-eslint/recommended',
        'prettier/@typescript-eslint'
    ],
    parser: '@typescript-eslint/parser',
    plugins: ['@typescript-eslint'],
    rules: {
        '@typescript-eslint/explicit-function-return-type': 'warn',
        '@typescript-eslint/no-unused-vars': 'error'
    }
};

# Multi-language Dockerfile example
FROM python:3.9-slim as python-base
 
# Python dependencies
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# Node.js for frontend
FROM node:16-alpine as frontend-builder
WORKDIR /frontend
COPY frontend/package*.json ./
RUN npm ci --only=production
COPY frontend/ .
RUN npm run build
 
# Final image
FROM python-base as final
COPY --from=frontend-builder /frontend/dist ./static/
COPY src/ ./src/
EXPOSE 8000
CMD ["python", "-m", "uvicorn", "src.main:app", "--host", "0.0.0.0"]

# GitHub Actions workflow
name: Data Pipeline CI/CD
 
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
 
jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        language: [python, node, go]
        
    steps:
    - uses: actions/checkout@v3
    
    # Python testing
    - name: Set up Python
      if: matrix.language == 'python'
      uses: actions/setup-python@v4
      with:
        python-version: '3.9'
    
    - name: Install Python dependencies
      if: matrix.language == 'python'
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov
    
    - name: Run Python tests
      if: matrix.language == 'python'
      run: pytest --cov=src --cov-report=xml
    
    # Node.js testing
    - name: Set up Node.js
      if: matrix.language == 'node'
      uses: actions/setup-node@v3
      with:
        node-version: '16'
        cache: 'npm'
    
    - name: Install Node dependencies
      if: matrix.language == 'node'
      run: npm ci
    
    - name: Run Node tests
      if: matrix.language == 'node'
      run: npm test
    
    # Go testing
    - name: Set up Go
      if: matrix.language == 'go'
      uses: actions/setup-go@v4
      with:
        go-version: '1.19'
    
    - name: Run Go tests
      if: matrix.language == 'go'
      run: go test -v ./...
 
  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
    - name: Deploy to staging
      run: |
        # Deployment script
        echo "Deploying to staging environment"

"""
Data Processing Module
 
This module provides utilities for processing and transforming data
from various sources including CSV files, databases, and APIs.
 
Example:
    Basic usage of the data processor:
    
    >>> from data_processor import DataProcessor
    >>> processor = DataProcessor()
    >>> result = processor.process_file('data.csv')
    >>> print(f"Processed {result.record_count} records")
 
Attributes:
    DEFAULT_BATCH_SIZE (int): Default number of records to process at once
    SUPPORTED_FORMATS (list): List of supported file formats
"""
 
from typing import List, Dict, Optional, Union
from dataclasses import dataclass
import logging
 
logger = logging.getLogger(__name__)
 
@dataclass
class ProcessingResult:
    """Result of data processing operation.
    
    Attributes:
        record_count: Number of records processed
        success_count: Number of successfully processed records
        error_count: Number of records that failed processing
        errors: List of error messages
    """
    record_count: int
    success_count: int
    error_count: int
    errors: List[str]
 
class DataProcessor:
    """Main data processing class.
    
    This class handles the processing of data from various sources,
    applying transformations and validations as needed.
    
    Args:
        batch_size: Number of records to process at once
        validate_input: Whether to validate input data
        
    Raises:
        ValueError: If batch_size is less than 1
        
    Example:
        >>> processor = DataProcessor(batch_size=1000)
        >>> result = processor.process_file('large_dataset.csv')
    """
    
    def __init__(self, batch_size: int = 100, validate_input: bool = True):
        if batch_size < 1:
            raise ValueError("Batch size must be at least 1")
        
        self.batch_size = batch_size
        self.validate_input = validate_input
        logger.info(f"DataProcessor initialized with batch_size={batch_size}")
    
    def process_file(self, file_path: str, **kwargs) -> ProcessingResult:
        """Process data from a file.
        
        Args:
            file_path: Path to the file to process
            **kwargs: Additional processing options
                - encoding: File encoding (default: 'utf-8')
                - delimiter: CSV delimiter (default: ',')
                - skip_header: Skip first row (default: False)
        
        Returns:
            ProcessingResult containing processing statistics
            
        Raises:
            FileNotFoundError: If the file doesn't exist
            PermissionError: If the file cannot be read
            ValueError: If the file format is unsupported
            
        Example:
            >>> result = processor.process_file(
            ...     'data.csv',
            ...     encoding='utf-8',
            ...     delimiter=','
            ... )
            >>> print(f"Success rate: {result.success_count / result.record_count:.2%}")
        """
        # Implementation here
        pass

/**
 * Data Analytics API
 * 
 * Provides endpoints for data analysis and visualization
 * 
 * @swagger
 * components:
 *   schemas:
 *     DataPoint:
 *       type: object
 *       required:
 *         - id
 *         - value
 *         - timestamp
 *       properties:
 *         id:
 *           type: string
 *           description: Unique identifier
 *         value:
 *           type: number
 *           description: Numeric value
 *         timestamp:
 *           type: string
 *           format: date-time
 *           description: ISO timestamp
 */
 
import express from 'express';
import swaggerJsdoc from 'swagger-jsdoc';
import swaggerUi from 'swagger-ui-express';
 
const app = express();
 
/**
 * @swagger
 * /api/data:
 *   post:
 *     summary: Submit data points for analysis
 *     requestBody:
 *       required: true
 *       content:
 *         application/json:
 *           schema:
 *             type: array
 *             items:
 *               ref: '#/components/schemas/DataPoint'
 *     responses:
 *       200:
 *         description: Data processed successfully
 *         content:
 *           application/json:
 *             schema:
 *               type: object
 *               properties:
 *                 processed_count:
 *                   type: number
 *                 analysis_id:
 *                   type: string
 */
app.post('/api/data', async (req: express.Request, res: express.Response) => {
    // Implementation
});
 
// Swagger setup
const swaggerOptions = {
    definition: {
        openapi: '3.0.0',
        info: {
            title: 'Data Analytics API',
            version: '1.0.0',
            description: 'API for data analysis and visualization'
        }
    },
    apis: ['./src/*.ts']
};
 
const specs = swaggerJsdoc(swaggerOptions);
app.use('/api-docs', swaggerUi.serve, swaggerUi.setup(specs));

# pytest configuration (pytest.ini)
[tool:pytest]
addopts = 
    --strict-markers
    --strict-config
    --cov=src
    --cov-branch
    --cov-report=term-missing
    --cov-report=html
    --cov-fail-under=80
    
markers =
    unit: Unit tests
    integration: Integration tests
    slow: Slow running tests
    external: Tests requiring external services
 
# Test example with multiple techniques
import pytest
from unittest.mock import Mock, patch, MagicMock
from hypothesis import given, strategies as st
from freezegun import freeze_time
from datetime import datetime
 
class TestDataProcessor:
    
    @pytest.fixture
    def processor(self):
        return DataProcessor(batch_size=10)
    
    @pytest.fixture
    def sample_data(self):
        return [
            {'id': '1', 'value': 100, 'timestamp': '2024-01-01T00:00:00Z'},
            {'id': '2', 'value': 200, 'timestamp': '2024-01-01T01:00:00Z'},
        ]
    
    @pytest.mark.unit
    def test_processor_initialization(self):
        processor = DataProcessor(batch_size=5)
        assert processor.batch_size == 5
        assert processor.validate_input is True
    
    @pytest.mark.unit
    def test_invalid_batch_size(self):
        with pytest.raises(ValueError, match="Batch size must be at least 1"):
            DataProcessor(batch_size=0)
    
    @pytest.mark.integration
    @patch('your_module.database_connection')
    def test_process_with_database(self, mock_db, processor, sample_data):
        mock_db.save_batch.return_value = True
        
        result = processor.process_data(sample_data)
        
        assert result.success_count == 2
        assert result.error_count == 0
        mock_db.save_batch.assert_called_once()
    
    @pytest.mark.slow
    @given(st.lists(st.dictionaries(
        keys=st.sampled_from(['id', 'value', 'timestamp']),
        values=st.one_of(st.text(), st.integers(), st.datetimes())
    ), min_size=1, max_size=100))
    def test_process_random_data(self, processor, random_data):
        # Property-based testing with random data
        result = processor.process_data(random_data)
        assert result.record_count == len(random_data)
        assert result.success_count + result.error_count == result.record_count
    
    @pytest.mark.external
    @freeze_time("2024-01-01 12:00:00")
    def test_timestamp_processing(self, processor):
        # Test with frozen time
        data = [{'id': '1', 'value': 100}]  # No timestamp provided
        result = processor.process_data(data)
        
        # Verify default timestamp is used
        assert result.success_count == 1
    
    @pytest.mark.parametrize("batch_size,expected_batches", [
        (1, 5),
        (2, 3),
        (5, 1),
        (10, 1),
    ])
    def test_batching_logic(self, batch_size, expected_batches):
        processor = DataProcessor(batch_size=batch_size)
        data = [{'id': str(i), 'value': i} for i in range(5)]
        
        with patch.object(processor, '_process_batch') as mock_batch:
            processor.process_data(data)
            assert mock_batch.call_count == expected_batches

# Example script to assess package quality
import requests
import json
from datetime import datetime, timedelta
 
def assess_package_quality(package_name: str, language: str) -> dict:
    """Assess the quality of a package based on various metrics."""
    
    if language == 'python':
        return assess_pypi_package(package_name)
    elif language == 'javascript':
        return assess_npm_package(package_name)
    elif language == 'rust':
        return assess_crates_package(package_name)
    
def assess_pypi_package(package_name: str) -> dict:
    """Assess Python package quality from PyPI."""
    
    # Get package info from PyPI API
    response = requests.get(f"https://pypi.org/pypi/{package_name}/json")
    if response.status_code != 200:
        return {"error": "Package not found"}
    
    data = response.json()
    info = data['info']
    releases = data['releases']
    
    # Calculate metrics
    latest_version = info['version']
    description_length = len(info['description']) if info['description'] else 0
    has_documentation = bool(info.get('home_page') or info.get('project_urls', {}).get('Documentation'))
    has_repository = bool(info.get('project_urls', {}).get('Repository'))
    
    # Release frequency
    recent_releases = [
        v for v, releases_info in releases.items()
        if releases_info and any(
            datetime.fromisoformat(r['upload_time'].replace('Z', '+00:00')) > 
            datetime.now().replace(tzinfo=None) - timedelta(days=365)
            for r in releases_info
        )
    ]
    
    return {
        'package_name': package_name,
        'latest_version': latest_version,
        'description_quality': 'good' if description_length > 100 else 'poor',
        'has_documentation': has_documentation,
        'has_repository': has_repository,
        'release_frequency': len(recent_releases),
        'maintainer': info.get('author', 'Unknown'),
        'license': info.get('license', 'Not specified'),
        'quality_score': calculate_quality_score({
            'description': description_length > 100,
            'documentation': has_documentation,
            'repository': has_repository,
            'recent_activity': len(recent_releases) > 0
        })
    }
 
def calculate_quality_score(indicators: dict) -> float:
    """Calculate overall quality score from various indicators."""
    weights = {
        'description': 0.2,
        'documentation': 0.3,
        'repository': 0.2,
        'recent_activity': 0.3
    }
    
    score = sum(
        weights[indicator] * (1.0 if value else 0.0)
        for indicator, value in indicators.items()
        if indicator in weights
    )
    
    return score
 
# Usage example
packages_to_assess = [
    ('pandas', 'python'),
    ('numpy', 'python'),
    ('express', 'javascript'),
    ('tokio', 'rust')
]
 
for package, language in packages_to_assess:
    quality_info = assess_package_quality(package, language)
    print(f"{package} ({language}): Quality Score = {quality_info.get('quality_score', 'N/A')}")

# Python calling Rust code via PyO3
# Rust side (lib.rs)
"""
use pyo3::prelude::*;
 
#[pyfunction]
fn fast_data_processing(data: Vec<f64>) -> PyResult<Vec<f64>> {
    // High-performance processing in Rust
    let processed: Vec<f64> = data
        .iter()
        .map(|&x| x * 2.0 + 1.0)  // Example transformation
        .collect();
    
    Ok(processed)
}
 
#[pymodule]
fn rust_extensions(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(fast_data_processing, m)?)?;
    Ok(())
}
"""
 
# Python side
import rust_extensions
import numpy as np
 
def hybrid_processing_pipeline(data: np.ndarray) -> np.ndarray:
    """Combine Python and Rust for optimal performance."""
    
    # Data preparation in Python
    cleaned_data = data[~np.isnan(data)]
    
    # Heavy computation in Rust
    processed_data = rust_extensions.fast_data_processing(cleaned_data.tolist())
    
    # Post-processing in Python
    result = np.array(processed_data)
    return result / np.sum(result)  # Normalize

// JavaScript calling Rust compiled to WebAssembly
class WasmDataProcessor {
    constructor() {
        this.wasmModule = null;
    }
    
    async initialize() {
        // Load WebAssembly module
        const wasmModule = await import('./pkg/data_processor.js');
        await wasmModule.default();
        this.wasmModule = wasmModule;
    }
    
    processLargeDataset(data) {
        if (!this.wasmModule) {
            throw new Error('WASM module not initialized');
        }
        
        // Convert JavaScript array to WASM-compatible format
        const wasmArray = new Float64Array(data);
        
        // Call WASM function for heavy computation
        const result = this.wasmModule.process_data_fast(wasmArray);
        
        // Convert back to JavaScript array
        return Array.from(result);
    }
}
 
// Usage
const processor = new WasmDataProcessor();
await processor.initialize();
 
const largeDataset = new Array(1000000).fill().map(() => Math.random());
const processed = processor.processLargeDataset(largeDataset);
console.log(`Processed {processed.length} data points`);

# docker-compose.yml for polyglot microservices
version: '3.8'
 
services:
  # Python ML service
  ml-service:
    build: ./ml-service
    ports:
      - "8001:8000"
    environment:
      - MODEL_PATH=/models
    volumes:
      - ./models:/models
    depends_on:
      - redis
      - postgres
  
  # Go API gateway
  api-gateway:
    build: ./api-gateway
    ports:
      - "8080:8080"
    environment:
      - ML_SERVICE_URL=http://ml-service:8000
      - FRONTEND_SERVICE_URL=http://frontend:3000
    depends_on:
      - ml-service
      - frontend
  
  # JavaScript frontend
  frontend:
    build: ./frontend
    ports:
      - "3000:3000"
    environment:
      - REACT_APP_API_URL=http://localhost:8080
  
  # Rust data processing service
  data-processor:
    build: ./data-processor
    ports:
      - "8002:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@postgres:5432/datadb
    depends_on:
      - postgres
  
  # Shared services
  postgres:
    image: postgres:13
    environment:
      POSTGRES_DB: datadb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
 
volumes:
  postgres_data: