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 List, Dict, Optional

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

class DeploymentMetricsService:
    """
    Service for tracking and analyzing software deployment metrics in BigQuery.
    Supports DORA metrics calculation for development teams.
    """
    
    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."""
        schema = [
            bigquery.SchemaField("deployment_id", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("service_name", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("environment", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("version", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("deployed_at", "TIMESTAMP", mode="REQUIRED"),
            bigquery.SchemaField("deployed_by", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("status", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("rollback", "BOOLEAN", mode="REQUIRED"),
            bigquery.SchemaField("lead_time_minutes", "INTEGER"),
        ]
        
        try:
            self.client.get_table(self.table_id)
            logger.info(f"Table {self.table_id} already exists")
        except NotFound:
            table = bigquery.Table(self.table_id, schema=schema)
            table.time_partitioning = bigquery.TimePartitioning(
                type_=bigquery.TimePartitioningType.DAY,
                field="deployed_at"
            )
            table.clustering_fields = ["service_name", "environment"]
            self.client.create_table(table)
            logger.info(f"Created table {self.table_id}")
    
    def record_deployment(self, deployment_data: Dict) -> str:
        """Insert a new deployment record with validation."""
        required_fields = ["deployment_id", "service_name", "environment", 
                          "version", "deployed_by", "status"]
        
        for field in required_fields:
            if field not in deployment_data:
                raise ValueError(f"Missing required field: {field}")
        
        deployment_data["deployed_at"] = datetime.utcnow().isoformat()
        deployment_data["rollback"] = deployment_data.get("rollback", False)
        
        errors = self.client.insert_rows_json(self.table_id, [deployment_data])
        
        if errors:
            logger.error(f"Failed to insert deployment: {errors}")
            raise Exception(f"BigQuery insert failed: {errors}")
        
        logger.info(f"Recorded deployment {deployment_data['deployment_id']}")
        return deployment_data["deployment_id"]
    
    def get_deployment_frequency(self, service_name: str, 
                                days: int = 30) -> Dict:
        """Calculate deployment frequency (DORA metric)."""
        query = f"""
        SELECT 
            service_name,
            environment,
            COUNT(*) as total_deployments,
            COUNT(*) / {days} as deployments_per_day,
            COUNTIF(status = 'success') as successful_deployments,
            COUNTIF(rollback = true) as rollbacks,
            SAFE_DIVIDE(COUNTIF(rollback = true), COUNT(*)) * 100 as rollback_rate
        FROM `{self.table_id}`
        WHERE service_name = @service_name
            AND deployed_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL @days DAY)
            AND environment = 'production'
        GROUP BY service_name, environment
        """
        
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("service_name", "STRING", service_name),
                bigquery.ScalarQueryParameter("days", "INT64", days)
            ]
        )
        
        try:
            results = self.client.query(query, job_config=job_config).result()
            metrics = [dict(row) for row in results]
            return metrics[0] if metrics else {}
        except Exception as e:
            logger.error(f"Query failed: {e}")
            raise
    
    def get_lead_time_metrics(self, service_name: str, days: int = 30) -> Dict:
        """Calculate lead time for changes (DORA metric)."""
        query = f"""
        SELECT 
            service_name,
            AVG(lead_time_minutes) as avg_lead_time_minutes,
            APPROX_QUANTILES(lead_time_minutes, 100)[OFFSET(50)] as median_lead_time,
            APPROX_QUANTILES(lead_time_minutes, 100)[OFFSET(95)] as p95_lead_time
        FROM `{self.table_id}`
        WHERE service_name = @service_name
            AND deployed_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL @days DAY)
            AND lead_time_minutes IS NOT NULL
            AND status = 'success'
        GROUP BY service_name
        """
        
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("service_name", "STRING", service_name),
                bigquery.ScalarQueryParameter("days", "INT64", days)
            ]
        )
        
        results = self.client.query(query, job_config=job_config).result()
        metrics = [dict(row) for row in results]
        return metrics[0] if metrics else {}

if __name__ == "__main__":
    service = DeploymentMetricsService(
        project_id="my-project",
        dataset_id="engineering_metrics"
    )
    
    deployment = {
        "deployment_id": "deploy-12345",
        "service_name": "payment-api",
        "environment": "production",
        "version": "v2.3.1",
        "deployed_by": "ci-cd-pipeline",
        "status": "success",
        "rollback": False,
        "lead_time_minutes": 45
    }
    
    service.record_deployment(deployment)
    frequency = service.get_deployment_frequency("payment-api", days=30)
    lead_time = service.get_lead_time_metrics("payment-api", days=30)
    
    logger.info(f"Deployment frequency: {frequency}")
    logger.info(f"Lead time metrics: {lead_time}")