adds new ValidateInLineCondition component

Author: pnilan

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -4275,6 +4275,7 @@ definitions:
         description: The condition that the specified config value will be evaluated against
         anyOf:
           - "$ref": "#/definitions/ValidateAdheresToSchema"
+          - "$ref": "#/definitions/ValidateInLineCondition"
   PredicateValidator:
     title: Predicate Validator
     description: Validator that applies a validation strategy to a specified value.
@@ -4303,12 +4304,24 @@ definitions:
           - "test-value"
           - "{{ config['api_version'] }}"
           - "{{ config['tenant_id'] }}"
+          - "{{ config['start_date'] < now_utc() }}"
           - 123
       validation_strategy:
         title: Validation Strategy
         description: The validation strategy to apply to the value.
         anyOf:
           - "$ref": "#/definitions/ValidateAdheresToSchema"
+          - "$ref": "#/definitions/ValidateInLineCondition"
+  ValidateInLineCondition:
+    title: Validate In Line Condition
+    description: Validation strategy that evalutaes the value as an InterpolatedBoolean.
+    type: object
+    required:
+      - type
+    properties:
+      type:
+        type: string
+        enum: [ValidateInLineCondition]
   ValidateAdheresToSchema:
     title: Validate Adheres To Schema
     description: Validates that a user-provided schema adheres to a specified JSON schema.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -1532,6 +1534,10 @@ class Config:
     query: Dict[str, Any] = Field(..., description="The GraphQL query to be executed")
 
 
+class ValidateInLineCondition(BaseModel):
+    type: Literal["ValidateInLineCondition"]
+
+
 class ValidateAdheresToSchema(BaseModel):
     type: Literal["ValidateAdheresToSchema"]
     base_schema: Union[str, Dict[str, Any]] = Field(
@@ -2003,7 +2009,7 @@ class DpathValidator(BaseModel):
         ],
         title="Field Path",
     )
-    validation_strategy: ValidateAdheresToSchema = Field(
+    validation_strategy: Union[ValidateAdheresToSchema, ValidateInLineCondition] = Field(
         ...,
         description="The condition that the specified config value will be evaluated against",
         title="Validation Strategy",
@@ -2019,11 +2025,12 @@ class PredicateValidator(BaseModel):
             "test-value",
             "{{ config['api_version'] }}",
             "{{ config['tenant_id'] }}",
+            "{{ config['start_date'] < now_utc() }}",
             123,
         ],
         title="Value",
     )
-    validation_strategy: ValidateAdheresToSchema = Field(
+    validation_strategy: Union[ValidateAdheresToSchema, ValidateInLineCondition] = Field(
         ...,
         description="The validation strategy to apply to the value.",
         title="Validation Strategy",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -424,6 +424,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ValidateAdheresToSchema as ValidateAdheresToSchemaModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ValidateInLineCondition as ValidateInLineConditionModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     WaitTimeFromHeader as WaitTimeFromHeaderModel,
@@ -567,6 +570,7 @@
     DpathValidator,
     PredicateValidator,
     ValidateAdheresToSchema,
+    ValidateInLineCondition,
 )
 from airbyte_cdk.sources.http_logger import format_http_message
 from airbyte_cdk.sources.message import (
@@ -894,6 +898,12 @@ def create_validate_adheres_to_schema(
             schema=base_schema,
         )
 
+    @staticmethod
+    def create_validate_in_line_condition(
+        model: ValidateInLineConditionModel, config: Config, **kwargs: Any
+    ) -> ValidateInLineCondition:
+        return ValidateInLineCondition(config=config)
+
     @staticmethod
     def create_added_field_definition(
         model: AddedFieldDefinitionModel, config: Config, **kwargs: Any

airbyte_cdk/sources/declarative/validators/__init__.py

@@ -7,6 +7,9 @@
 from airbyte_cdk.sources.declarative.validators.validate_adheres_to_schema import (
     ValidateAdheresToSchema,
 )
+from airbyte_cdk.sources.declarative.validators.validate_in_line_condition import (
+    ValidateInLineCondition,
+)
 from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
 from airbyte_cdk.sources.declarative.validators.validator import Validator
 
@@ -16,4 +19,5 @@
     "ValidationStrategy",
     "ValidateAdheresToSchema",
     "PredicateValidator",
+    "ValidateInLineCondition",
 ]

airbyte_cdk/sources/declarative/validators/validate_in_line_condition.py

@@ -0,0 +1,43 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any
+
+from jinja2.exceptions import TemplateError
+
+from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
+from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+from airbyte_cdk.sources.types import Config
+
+
+@dataclass
+class ValidateInLineCondition(ValidationStrategy):
+    """
+    Validation strategy that evaluates the argument as an InterpolatedBoolean.
+    """
+
+    config: Config
+
+    def validate(self, value: Any) -> None:
+        """
+        Validates the argument as an InterpolatedBoolean.
+
+        :param value: The value to validate
+        :raises ValueError: If the condition is not a string or evaluates to False
+        """
+
+        if isinstance(value, str):
+            interpolated_condition = InterpolatedBoolean(value, parameters={})
+            try:
+                result = interpolated_condition.eval(self.config)
+            except TemplateError as e:
+                raise ValueError(f"Invalid jinja expression: {value}.") from e
+            except Exception as e:
+                raise ValueError(f"Unexpected error evaluating condition: {value}.") from e
+
+            if not result:
+                raise ValueError(f"Condition evaluated to False: {value}.")
+        else:
+            raise ValueError(f"Invalid condition argument: {value}. Should be a string.")