feat: adds new ValidateInLineCondition component

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -4355,6 +4355,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.
@@ -4383,12 +4384,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 evaluates 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

@@ -1559,6 +1559,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(
@@ -2030,7 +2034,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",
@@ -2046,11 +2050,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

@@ -427,6 +427,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,
@@ -572,6 +575,7 @@
     DpathValidator,
     PredicateValidator,
     ValidateAdheresToSchema,
+    ValidateInLineCondition,
 )
 from airbyte_cdk.sources.http_logger import format_http_message
 from airbyte_cdk.sources.message import (
@@ -738,6 +742,7 @@ def _init_mappings(self) -> None:
             SpecModel: self.create_spec,
             SubstreamPartitionRouterModel: self.create_substream_partition_router,
             ValidateAdheresToSchemaModel: self.create_validate_adheres_to_schema,
+            ValidateInLineConditionModel: self.create_validate_in_line_condition,
             WaitTimeFromHeaderModel: self.create_wait_time_from_header,
             WaitUntilTimeFromHeaderModel: self.create_wait_until_time_from_header,
             AsyncRetrieverModel: self.create_async_retriever,
@@ -900,6 +905,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/predicate_validator.py

@@ -6,18 +6,19 @@
 from typing import Any
 
 from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+from airbyte_cdk.sources.declarative.validators.validator import Validator
 
 
 @dataclass
-class PredicateValidator:
+class PredicateValidator(Validator):
     """
     Validator that applies a validation strategy to a value.
     """
 
     value: Any
     strategy: ValidationStrategy
 
-    def validate(self) -> None:
+    def validate(self, input_data: Any) -> None:
         """
         Applies the validation strategy to the value.
 

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.")