Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

SQL Operator Context (via DbApiHook integration):


Code Block
languagepy
collapsetrue
# Automatically injected context for PostgreSQL:


{

    "database_type": "postgresql",

    "version": "15.2",

    "available_tables": ["customers", "orders", "products"],

    "schema_info": {

        "customers": {

            "customer_id": {"type": "integer", "nullable": False, "primary_key": True},

            "email_address": {"type": "varchar(255)", "nullable": False, "unique": True},

            "total_revenue": {"type": "decimal(10,2)", "nullable": True}

        }

    },

    "sample_data": {

        "customers": [




            {"customer_id": 1, "email_address": "john@example.com", "total_revenue": 1250.00},

            {"customer_id": 2, "email_address": "jane@example.com", "total_revenue": 890.50}

        ]

    },

    "dialect_features": {

        "supports_window_functions": True,

        "supports_cte": True,

        "date_functions": ["DATE_TRUNC", "EXTRACT", "AGE"]

    }

}

File Operator Context (via S3Hook/GCSHook integration):

# Automatically injected context for S3 Parquet files:

{




    "storage_type": "s3",

    "file_format": "parquet", 

    "file_size_mb": 245,

    "estimated_rows": 1000000,

    "schema_info": {

        "id": "int64",

        "name": "string", 

        "email": "string",

        "signup_date": "timestamp[ns]"

    },

    "sample_data": [

        {"id": 1, "name": "John Doe", "email": "john@example.com"},

        {"id": 2, "name": "Jane Smith", "email": "jane@example.com"}

    ],

    "partitioning": ["year", "month"],

    "compression": "snappy"

}

...

SQL Operator Built-in Protection:


Code Block
languagepy
collapsetrue
class LLMSQLQueryOperator(BaseOperator):

    # Built-in dangerous operation blocking


    BLOCKED_KEYWORDS = ["DROP", "TRUNCATE", "DELETE FROM", "ALTER TABLE", "GRANT", "REVOKE"]


    DEFAULT_SYSTEM_PROMPT = """You are a SQL expert integrated with {database_type}.

...

File Operator Built-in Protection:


Code Block
languagepy
collapsetrue
class LLMFileAnalysisOperator(BaseOperator):

    ALLOWED_OPERATIONS = ["read", "analyze", "summarize", "validate"]


    DEFAULT_SYSTEM_PROMPT = """You are a file analysis expert for {storage_type} {file_format} files.

...

Option A: Embedded HITL


Code Block
languagepy
collapsetrue
quality_check = LLMDataQualityOperator(

    task_id="customer_quality_analysis",

    data_sources=[customer_s3],

    prompt="Generate data quality validation queries",

    require_approval=True,  # Built-in HITL

    approval_timeout=timedelta(hours=2)

)

...

Option B: Separate HITL Steps


Code Block
languagepy
collapsetrue
# Generate queries

generate_queries = LLMDataQualityOperator(

    task_id="generate_quality_queries",

    data_sources=[customer_s3],

    prompt="Generate data quality validation queries",

    dry_run=True  # Don't execute, just generate

)

# Human approval step

approve_queries = ApprovalOperator(

    task_id="approve_queries",

    body="{{ ti.xcom_pull(task_ids='generate_quality_queries') }}",

    allow_modifications=True  # Users can edit generated queries

)


# Execute approved queries

execute_analysis = AnalyticsOperator(

    task_id="execute_quality_checks",

    query="{{ ti.xcom_pull(task_ids='approve_queries') }}",

    data_sources=[customer_s3]

)


generate_queries >> approve_queries >> execute_analysis

...