Comprehensive comparison for Database technology in Software Development applications

from google.cloud import bigquery
from google.cloud.exceptions import NotFound
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Optional

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

class DeploymentAnalytics:
    """
    Production-ready BigQuery service for analyzing software deployment metrics.
    Tracks deployment frequency, success rates, and rollback patterns.
    """
    
    def __init__(self, project_id: str, dataset_id: str):
        self.client = bigquery.Client(project=project_id)
        self.project_id = project_id
        self.dataset_id = dataset_id
        self.table_id = f"{project_id}.{dataset_id}.deployments"
        self._ensure_table_exists()
    
    def _ensure_table_exists(self) -> None:
        """Create deployments table if it doesn't exist."""
        try:
            self.client.get_table(self.table_id)
            logger.info(f"Table {self.table_id} already exists")
        except NotFound:
            schema = [
                bigquery.SchemaField("deployment_id", "STRING", mode="REQUIRED"),
                bigquery.SchemaField("service_name", "STRING", mode="REQUIRED"),
                bigquery.SchemaField("version", "STRING", mode="REQUIRED"),
                bigquery.SchemaField("environment", "STRING", mode="REQUIRED"),
                bigquery.SchemaField("deployed_by", "STRING", mode="REQUIRED"),
                bigquery.SchemaField("deployed_at", "TIMESTAMP", mode="REQUIRED"),
                bigquery.SchemaField("status", "STRING", mode="REQUIRED"),
                bigquery.SchemaField("duration_seconds", "INTEGER"),
                bigquery.SchemaField("rollback", "BOOLEAN", mode="REQUIRED"),
            ]
            table = bigquery.Table(self.table_id, schema=schema)
            table.time_partitioning = bigquery.TimePartitioning(
                type_=bigquery.TimePartitioningType.DAY,
                field="deployed_at"
            )
            self.client.create_table(table)
            logger.info(f"Created table {self.table_id}")
    
    def record_deployment(self, deployment_data: Dict) -> bool:
        """Insert a deployment record with error handling."""
        try:
            rows_to_insert = [deployment_data]
            errors = self.client.insert_rows_json(self.table_id, rows_to_insert)
            
            if errors:
                logger.error(f"Errors inserting rows: {errors}")
                return False
            
            logger.info(f"Recorded deployment {deployment_data['deployment_id']}")
            return True
        except Exception as e:
            logger.error(f"Failed to record deployment: {str(e)}")
            return False
    
    def get_deployment_metrics(self, days: int = 30) -> Optional[List[Dict]]:
        """Calculate key deployment metrics using parameterized query."""
        query = f"""
        WITH deployment_stats AS (
            SELECT
                service_name,
                environment,
                COUNT(*) as total_deployments,
                COUNTIF(status = 'success') as successful_deployments,
                COUNTIF(rollback = true) as rollback_count,
                AVG(duration_seconds) as avg_duration,
                MAX(deployed_at) as last_deployment
            FROM `{self.table_id}`
            WHERE deployed_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL @days DAY)
            GROUP BY service_name, environment
        )
        SELECT
            service_name,
            environment,
            total_deployments,
            successful_deployments,
            ROUND(successful_deployments / total_deployments * 100, 2) as success_rate,
            rollback_count,
            ROUND(avg_duration / 60, 2) as avg_duration_minutes,
            last_deployment
        FROM deployment_stats
        ORDER BY total_deployments DESC
        """
        
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("days", "INT64", days)
            ]
        )
        
        try:
            query_job = self.client.query(query, job_config=job_config)
            results = query_job.result()
            
            metrics = [dict(row) for row in results]
            logger.info(f"Retrieved metrics for {len(metrics)} services")
            return metrics
        except Exception as e:
            logger.error(f"Failed to retrieve metrics: {str(e)}")
            return None

if __name__ == "__main__":
    analytics = DeploymentAnalytics(
        project_id="my-project",
        dataset_id="software_metrics"
    )
    
    sample_deployment = {
        "deployment_id": "dep-12345",
        "service_name": "api-gateway",
        "version": "v2.3.1",
        "environment": "production",
        "deployed_by": "ci-cd-pipeline",
        "deployed_at": datetime.utcnow().isoformat(),
        "status": "success",
        "duration_seconds": 180,
        "rollback": False
    }
    
    analytics.record_deployment(sample_deployment)
    metrics = analytics.get_deployment_metrics(days=30)
    
    if metrics:
        for metric in metrics:
            print(f"Service: {metric['service_name']} - Success Rate: {metric['success_rate']}%")