Version Management

def get_version_information() -> str:
    """
    Get comprehensive version information for dbt-core and installed plugins.
    
    Returns formatted string containing:
    - dbt-core version and update status
    - All installed dbt adapter plugin versions
    - Update recommendations
    
    Returns:
        str: Multi-line formatted version information
    """

def get_installed_version():
    """
    Get the currently installed dbt-core version.
    
    Returns:
        str: Version string (e.g., "1.10.10")
    """

def get_latest_version():
    """
    Get the latest available dbt-core version from PyPI.
    
    Makes HTTP request to PyPI API to check for newer releases.
    
    Returns:
        str: Latest version string, or None if check fails
    """

from dbt.version import get_installed_version, get_latest_version

# Get current version
current = get_installed_version()
print(f"Installed dbt-core: {current}")

# Check for updates
latest = get_latest_version()
if latest and latest != current:
    print(f"Update available: {latest}")
else:
    print("dbt-core is up to date")

from dbt.version import get_version_information

# Get full version report
version_info = get_version_information()
print(version_info)

# Example output:
# Core:
#   - dbt-core: 1.10.10 (latest)
#
# Plugins:
#   - dbt-postgres: 1.10.10 (latest)
#   - dbt-snowflake: 1.10.10 (latest)
#   - dbt-bigquery: 1.10.10 (latest)

import dbt_common.semver as semver

def check_version_compatibility(required_version, current_version):
    """Check if current version meets requirements."""
    try:
        required = semver.VersionSpecifier(required_version)
        current = semver.Version(current_version)
        return required.is_satisfied_by(current)
    except Exception:
        return False

# Usage
if check_version_compatibility(">=1.10.0", get_installed_version()):
    print("Version meets requirements")
else:
    print("Version upgrade required")

def check_adapter_versions():
    """Check versions of installed adapter plugins."""
    version_info = get_version_information()
    lines = version_info.split('\n')
    
    plugins = {}
    in_plugins = False
    
    for line in lines:
        if line.startswith('Plugins:'):
            in_plugins = True
            continue
        elif in_plugins and line.strip().startswith('- '):
            # Parse plugin line: "  - dbt-postgres: 1.10.10 (latest)"
            plugin_info = line.strip()[2:]  # Remove "- "
            if ':' in plugin_info:
                name, version_part = plugin_info.split(':', 1)
                version = version_part.strip().split()[0]
                plugins[name.strip()] = version
    
    return plugins

# Get adapter versions
adapters = check_adapter_versions()
for adapter, version in adapters.items():
    print(f"{adapter}: {version}")

from dbt.version import get_installed_version, get_latest_version
import requests
from packaging import version

def check_for_updates(notify=True):
    """
    Check for dbt-core updates with custom notification.
    
    Args:
        notify: Whether to print update notifications
    
    Returns:
        dict: Update information
    """
    current = get_installed_version()
    latest = get_latest_version()
    
    update_info = {
        'current': current,
        'latest': latest,
        'update_available': False,
        'is_prerelease': False
    }
    
    if latest and current:
        try:
            current_ver = version.parse(current)
            latest_ver = version.parse(latest)
            
            update_info['update_available'] = latest_ver > current_ver
            update_info['is_prerelease'] = latest_ver.is_prerelease
            
            if notify and update_info['update_available']:
                print(f"dbt-core update available: {current} → {latest}")
                if not update_info['is_prerelease']:
                    print("Run: pip install --upgrade dbt-core")
                
        except Exception as e:
            if notify:
                print(f"Could not check for updates: {e}")
    
    return update_info

import os
import time
from pathlib import Path

def should_check_updates():
    """Check if it's time for an update check (daily)."""
    check_file = Path.home() / '.dbt' / 'last_update_check'
    
    if not check_file.exists():
        return True
    
    try:
        last_check = float(check_file.read_text())
        return time.time() - last_check > 86400  # 24 hours
    except:
        return True

def record_update_check():
    """Record that we performed an update check."""
    check_file = Path.home() / '.dbt' / 'last_update_check'
    check_file.parent.mkdir(exist_ok=True)
    check_file.write_text(str(time.time()))

def periodic_update_check():
    """Perform periodic update check."""
    if should_check_updates():
        update_info = check_for_updates(notify=True)
        record_update_check()
        return update_info
    return None

from dbt.version import get_installed_version
from packaging import version

def supports_feature(feature_name, min_version):
    """Check if current version supports a feature."""
    current = get_installed_version()
    if not current:
        return False
    
    try:
        return version.parse(current) >= version.parse(min_version)
    except:
        return False

# Feature checking
if supports_feature("contracts", "1.5.0"):
    print("Model contracts supported")

if supports_feature("saved_queries", "1.8.0"):  
    print("Saved queries supported")

if supports_feature("unit_tests", "1.8.0"):
    print("Unit tests supported")

def get_version_config():
    """Get configuration based on dbt version."""
    current = get_installed_version()
    if not current:
        return {}
    
    config = {}
    current_ver = version.parse(current)
    
    # Version-specific features
    if current_ver >= version.parse("1.8.0"):
        config['supports_unit_tests'] = True
        config['supports_saved_queries'] = True
    
    if current_ver >= version.parse("1.5.0"):
        config['supports_contracts'] = True
        config['supports_model_versions'] = True
    
    if current_ver >= version.parse("1.0.0"):
        config['supports_metrics'] = True
        config['supports_exposures'] = True
    
    return config

# Usage
config = get_version_config()
if config.get('supports_contracts'):
    print("Can use model contracts")

import click
from dbt.version import get_version_information

@click.command()
@click.option('--check-updates', is_flag=True, help='Check for updates')
def version_command(check_updates):
    """Display version information."""
    version_info = get_version_information()
    click.echo(version_info)
    
    if check_updates:
        update_info = check_for_updates()
        if update_info['update_available']:
            click.echo(f"\nUpdate available: {update_info['latest']}")
        else:
            click.echo("\ndbt-core is up to date")

def health_check():
    """Perform application health check including version validation."""
    health = {
        'dbt_version': get_installed_version(),
        'update_available': False,
        'plugins': {},
        'status': 'healthy'
    }
    
    try:
        # Check for updates
        update_info = check_for_updates(notify=False)
        health['update_available'] = update_info['update_available']
        health['latest_version'] = update_info['latest']
        
        # Check plugin versions
        health['plugins'] = check_adapter_versions()
        
        # Validate minimum version requirements
        min_required = "1.8.0"
        if not supports_feature("minimum", min_required):
            health['status'] = 'warning'
            health['message'] = f"dbt-core {min_required}+ recommended"
            
    except Exception as e:
        health['status'] = 'error'
        health['error'] = str(e)
    
    return health

# Usage in monitoring
health = health_check()
if health['status'] != 'healthy':
    print(f"Health check warning: {health.get('message', 'Unknown issue')}")

def validate_environment():
    """Validate that the dbt environment meets requirements."""
    issues = []
    
    # Check core version
    current = get_installed_version()
    if not current:
        issues.append("dbt-core not installed")
        return issues
    
    # Check for major version compatibility
    current_ver = version.parse(current)
    if current_ver.major < 1:
        issues.append(f"dbt-core {current} is outdated (1.0+ required)")
    
    # Check for adapter compatibility
    adapters = check_adapter_versions()
    for adapter, adapter_version in adapters.items():
        try:
            adapter_ver = version.parse(adapter_version)
            if adapter_ver.major != current_ver.major:
                issues.append(f"{adapter} version mismatch: {adapter_version} vs core {current}")
        except:
            issues.append(f"Could not parse {adapter} version: {adapter_version}")
    
    return issues

# Environment validation
issues = validate_environment()
if issues:
    print("Environment issues found:")
    for issue in issues:
        print(f"  - {issue}")
else:
    print("Environment validation passed")