Page History

# Schema comparison with custom logic

@task.llm_schema_compare(data_sources=[s3_asset, postgres_asset])

def intelligent_schema_validation():

    # Pre-processing: check business calendar

    is_migration_window = check_migration_window()

    if is_migration_window:

        prompt = "Compare schemas and generate migration plan for scheduled maintenance window"

    else:

        prompt = "Compare schemas and flag breaking changes - no migrations allowed"

    return {

        "prompt": prompt,

        "migration_allowed": is_migration_window,

        "additional_context": {"maintenance_window": is_migration_window}

    }

# Data quality with dynamic rules

@task.llm_data_quality(data_sources=[customer_asset])

def adaptive_quality_checks():

    # Pre-processing: get current business rules

    current_rules = fetch_business_rules()

    seasonal_adjustments = get_seasonal_data_patterns()

    prompt = f"""

    Validate customer data against current business rules:

    {current_rules}

    Apply seasonal adjustments for data volume expectations:

    {seasonal_adjustments}

    Generate appropriate validation queries.

    """

    return {

        "prompt": prompt,

        "business_rules": current_rules,

        "seasonal_context": seasonal_adjustments

    }

# File analysis with preprocessing

@task.llm_file_analysis(data_sources=[log_files_asset])

def analyze_logs_with_context():

    # Pre-processing: get system context

    recent_deployments = get_recent_deployments()

    system_alerts = get_active_alerts()

    prompt = f"""

    Analyze log files for anomalies, considering:

    - Recent deployments: {recent_deployments}

    - Active system alerts: {system_alerts}

    Focus on correlation between deployment events and error patterns.

    """

    return {

        "prompt": prompt,

        "deployment_context": recent_deployments,

        "alert_context": system_alerts

    }