diff --git a/docs/docs/resources/pipeline-components/consumer-app.yaml b/docs/docs/resources/pipeline-components/consumer-app.yaml index 73abb162b..c11c09388 100644 --- a/docs/docs/resources/pipeline-components/consumer-app.yaml +++ b/docs/docs/resources/pipeline-components/consumer-app.yaml @@ -27,8 +27,6 @@ component-as-extra-pattern: type: pattern # Implied to be an extra pattern if `label` is defined label: some-role - # to: # While the consumer-app does inherit from kafka-app, it does not need a - # `to` section, hence it does not support it. namespace: namespace # required # Allowed configs: # https://github.com/bakdata/streams-bootstrap/tree/master/charts/consumer-app diff --git a/docs/docs/resources/pipeline-components/dependencies/defaults_pipeline_component_dependencies.yaml b/docs/docs/resources/pipeline-components/dependencies/defaults_pipeline_component_dependencies.yaml index 435a30cba..a612d4559 100644 --- a/docs/docs/resources/pipeline-components/dependencies/defaults_pipeline_component_dependencies.yaml +++ b/docs/docs/resources/pipeline-components/dependencies/defaults_pipeline_component_dependencies.yaml @@ -1,5 +1,4 @@ consumer-app.yaml: -- to-consumer-app.yaml - values-consumer-app.yaml consumer-producer-app.yaml: - values-consumer-producer-app.yaml @@ -8,24 +7,15 @@ helm-app.yaml: - repo_config-helm-app.yaml kafka-app.yaml: - prefix.yaml -- from_.yaml -- to.yaml kafka-connector.yaml: -- prefix.yaml -- from_.yaml -- to.yaml - config-kafka-connector.yaml kafka-sink-connector.yaml: [] -kafka-source-connector.yaml: -- from_-kafka-source-connector.yaml +kafka-source-connector.yaml: [] kubernetes-app.yaml: - prefix.yaml -- from_.yaml -- to.yaml - namespace.yaml - values-kubernetes-app.yaml producer-app.yaml: -- from_-producer-app.yaml - values-producer-app.yaml streams-app.yaml: - values-streams-app.yaml diff --git a/docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml b/docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml index c435fbd8b..ddcb1b36e 100644 --- a/docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml +++ b/docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml @@ -1,10 +1,9 @@ kpops_components_fields: consumer-app: - - name - enabled - prefix - from_ - - to + - name - namespace - values - repo_config @@ -12,11 +11,11 @@ kpops_components_fields: - version - timeout consumer-producer-app: - - name - enabled - prefix - from_ - to + - name - namespace - values - repo_config @@ -24,11 +23,9 @@ kpops_components_fields: - version - timeout helm-app: - - name - enabled - prefix - - from_ - - to + - name - namespace - values - repo_config @@ -36,55 +33,46 @@ kpops_components_fields: - version - timeout kafka-app: - - name - enabled - prefix - - from_ - - to - kafka-connector: - name + kafka-connector: - enabled - prefix - - from_ - to + - name - config - state kafka-sink-connector: - - name - enabled - prefix - from_ - to + - name - config - state kafka-source-connector: - - name - enabled - prefix - - from_ - to + - name - config - state kubernetes-app: - - name - enabled - prefix - - from_ - - to + - name - namespace - values pipeline-component: - - name - enabled - prefix - - from_ - - to - producer-app: - name + producer-app: - enabled - prefix - - from_ - to + - name - namespace - values - repo_config @@ -92,11 +80,11 @@ kpops_components_fields: - version - timeout streams-app: - - name - enabled - prefix - from_ - to + - name - namespace - values - repo_config @@ -104,11 +92,9 @@ kpops_components_fields: - version - timeout streams-bootstrap: - - name - enabled - prefix - - from_ - - to + - name - namespace - values - repo_config @@ -118,6 +104,7 @@ kpops_components_fields: kpops_components_inheritance_ref: consumer-app: bases: + - consumes-topics - streams-bootstrap parents: - streams-bootstrap @@ -128,6 +115,8 @@ kpops_components_inheritance_ref: - base-defaults-component consumer-producer-app: bases: + - consumes-topics + - produces-topics - streams-bootstrap parents: - streams-bootstrap @@ -151,15 +140,18 @@ kpops_components_inheritance_ref: - base-defaults-component kafka-connector: bases: - - pipeline-component + - produces-topics parents: + - kafka-app - pipeline-component - base-defaults-component kafka-sink-connector: bases: + - consumes-topics - kafka-connector parents: - kafka-connector + - kafka-app - pipeline-component - base-defaults-component kafka-source-connector: @@ -167,6 +159,7 @@ kpops_components_inheritance_ref: - kafka-connector parents: - kafka-connector + - kafka-app - pipeline-component - base-defaults-component kubernetes-app: @@ -181,6 +174,7 @@ kpops_components_inheritance_ref: - base-defaults-component producer-app: bases: + - produces-topics - streams-bootstrap parents: - streams-bootstrap @@ -191,6 +185,8 @@ kpops_components_inheritance_ref: - base-defaults-component streams-app: bases: + - consumes-topics + - produces-topics - streams-bootstrap parents: - streams-bootstrap diff --git a/docs/docs/resources/pipeline-components/dependencies/pipeline_component_dependencies.yaml b/docs/docs/resources/pipeline-components/dependencies/pipeline_component_dependencies.yaml index a880ff10e..93e7cc7a9 100644 --- a/docs/docs/resources/pipeline-components/dependencies/pipeline_component_dependencies.yaml +++ b/docs/docs/resources/pipeline-components/dependencies/pipeline_component_dependencies.yaml @@ -1,7 +1,6 @@ consumer-app.yaml: - prefix.yaml - from_.yaml -- to-consumer-app.yaml - namespace.yaml - values-consumer-app.yaml - repo_config-helm-app.yaml @@ -16,19 +15,14 @@ consumer-producer-app.yaml: - version-kafka-app.yaml helm-app.yaml: - prefix.yaml -- from_.yaml -- to.yaml - namespace.yaml - values-helm-app.yaml - repo_config-helm-app.yaml - version.yaml kafka-app.yaml: - prefix.yaml -- from_.yaml -- to.yaml kafka-connector.yaml: - prefix.yaml -- from_.yaml - to.yaml - config-kafka-connector.yaml kafka-sink-connector.yaml: @@ -38,18 +32,14 @@ kafka-sink-connector.yaml: - config-kafka-connector.yaml kafka-source-connector.yaml: - prefix.yaml -- from_-kafka-source-connector.yaml - to.yaml - config-kafka-connector.yaml kubernetes-app.yaml: - prefix.yaml -- from_.yaml -- to.yaml - namespace.yaml - values-kubernetes-app.yaml producer-app.yaml: - prefix.yaml -- from_-producer-app.yaml - to.yaml - namespace.yaml - values-producer-app.yaml diff --git a/docs/docs/resources/pipeline-components/helm-app.yaml b/docs/docs/resources/pipeline-components/helm-app.yaml index 9d9d84b44..2397c06ae 100644 --- a/docs/docs/resources/pipeline-components/helm-app.yaml +++ b/docs/docs/resources/pipeline-components/helm-app.yaml @@ -4,45 +4,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model namespace: namespace # required values: # required image: exampleImage # Example diff --git a/docs/docs/resources/pipeline-components/kafka-app.yaml b/docs/docs/resources/pipeline-components/kafka-app.yaml index 7677fd384..4184e28ed 100644 --- a/docs/docs/resources/pipeline-components/kafka-app.yaml +++ b/docs/docs/resources/pipeline-components/kafka-app.yaml @@ -5,42 +5,3 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model diff --git a/docs/docs/resources/pipeline-components/kafka-connector.yaml b/docs/docs/resources/pipeline-components/kafka-connector.yaml index 5f51d625b..91cde2bd0 100644 --- a/docs/docs/resources/pipeline-components/kafka-connector.yaml +++ b/docs/docs/resources/pipeline-components/kafka-connector.yaml @@ -3,27 +3,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role # Topic(s) into which the component will write output to: topics: diff --git a/docs/docs/resources/pipeline-components/kafka-source-connector.yaml b/docs/docs/resources/pipeline-components/kafka-source-connector.yaml index a698fb331..e72049798 100644 --- a/docs/docs/resources/pipeline-components/kafka-source-connector.yaml +++ b/docs/docs/resources/pipeline-components/kafka-source-connector.yaml @@ -4,8 +4,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - # The source connector has no `from` section - # from: # Topic(s) into which the component will write output to: topics: diff --git a/docs/docs/resources/pipeline-components/kubernetes-app.yaml b/docs/docs/resources/pipeline-components/kubernetes-app.yaml index 59aa52a0d..24ffce542 100644 --- a/docs/docs/resources/pipeline-components/kubernetes-app.yaml +++ b/docs/docs/resources/pipeline-components/kubernetes-app.yaml @@ -4,45 +4,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model namespace: namespace # required values: # required image: exampleImage # Example diff --git a/docs/docs/resources/pipeline-components/pipeline.yaml b/docs/docs/resources/pipeline-components/pipeline.yaml index e33628608..d73744aa1 100644 --- a/docs/docs/resources/pipeline-components/pipeline.yaml +++ b/docs/docs/resources/pipeline-components/pipeline.yaml @@ -27,8 +27,6 @@ component-as-extra-pattern: type: pattern # Implied to be an extra pattern if `label` is defined label: some-role - # to: # While the consumer-app does inherit from kafka-app, it does not need a - # `to` section, hence it does not support it. namespace: namespace # required # Allowed configs: # https://github.com/bakdata/streams-bootstrap/tree/master/charts/consumer-app @@ -246,45 +244,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model namespace: namespace # required values: # required image: exampleImage # Example @@ -308,45 +267,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model # Kafka sink connector - type: kafka-sink-connector name: kafka-sink-connector # required @@ -401,8 +321,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - # The source connector has no `from` section - # from: # Topic(s) into which the component will write output to: topics: @@ -430,45 +348,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model namespace: namespace # required values: # required image: exampleImage # Example @@ -482,8 +361,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - # from: # While the producer-app does inherit from kafka-app, it does not need a - # `from` section, hence it does not support it. # Topic(s) into which the component will write output to: topics: diff --git a/docs/docs/resources/pipeline-components/producer-app.yaml b/docs/docs/resources/pipeline-components/producer-app.yaml index ee80eeb3c..1071ac256 100644 --- a/docs/docs/resources/pipeline-components/producer-app.yaml +++ b/docs/docs/resources/pipeline-components/producer-app.yaml @@ -6,8 +6,6 @@ # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - # from: # While the producer-app does inherit from kafka-app, it does not need a - # `from` section, hence it does not support it. # Topic(s) into which the component will write output to: topics: diff --git a/docs/docs/resources/pipeline-defaults/defaults-consumer-app.yaml b/docs/docs/resources/pipeline-defaults/defaults-consumer-app.yaml index f5a5534c5..7115d8578 100644 --- a/docs/docs/resources/pipeline-defaults/defaults-consumer-app.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults-consumer-app.yaml @@ -3,8 +3,6 @@ # Child of: KafkaApp # More documentation on ConsumerApp: https://github.com/bakdata/streams-bootstrap consumer-app: - # to: # While the consumer-app does inherit from kafka-app, it does not need a - # `to` section, hence it does not support it. # Allowed configs: # https://github.com/bakdata/streams-bootstrap/tree/master/charts/consumer-app values: # required diff --git a/docs/docs/resources/pipeline-defaults/defaults-kafka-app.yaml b/docs/docs/resources/pipeline-defaults/defaults-kafka-app.yaml index 42c18565f..599ab4d5e 100644 --- a/docs/docs/resources/pipeline-defaults/defaults-kafka-app.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults-kafka-app.yaml @@ -6,42 +6,3 @@ kafka-app: # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model diff --git a/docs/docs/resources/pipeline-defaults/defaults-kafka-connector.yaml b/docs/docs/resources/pipeline-defaults/defaults-kafka-connector.yaml index 11468f921..24cfc7557 100644 --- a/docs/docs/resources/pipeline-defaults/defaults-kafka-connector.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults-kafka-connector.yaml @@ -3,48 +3,6 @@ # Parent of: KafkaSinkConnector, KafkaSourceConnector # Child of: PipelineComponent kafka-connector: - # Pipeline prefix that will prefix every component name. If you wish to not - # have any prefix you can specify an empty string. - prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 diff --git a/docs/docs/resources/pipeline-defaults/defaults-kafka-source-connector.yaml b/docs/docs/resources/pipeline-defaults/defaults-kafka-source-connector.yaml index fcd203910..3a413730e 100644 --- a/docs/docs/resources/pipeline-defaults/defaults-kafka-source-connector.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults-kafka-source-connector.yaml @@ -2,5 +2,3 @@ # # Child of: KafkaConnector kafka-source-connector: - # The source connector has no `from` section - # from: diff --git a/docs/docs/resources/pipeline-defaults/defaults-kubernetes-app.yaml b/docs/docs/resources/pipeline-defaults/defaults-kubernetes-app.yaml index 60d0f7fc6..c853f6936 100644 --- a/docs/docs/resources/pipeline-defaults/defaults-kubernetes-app.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults-kubernetes-app.yaml @@ -6,45 +6,6 @@ kubernetes-app: # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model namespace: namespace # required values: # required image: exampleImage # Example diff --git a/docs/docs/resources/pipeline-defaults/defaults-producer-app.yaml b/docs/docs/resources/pipeline-defaults/defaults-producer-app.yaml index 6dbe0e19a..00d4d759c 100644 --- a/docs/docs/resources/pipeline-defaults/defaults-producer-app.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults-producer-app.yaml @@ -3,8 +3,6 @@ # Child of: KafkaApp # More documentation on ProducerApp: https://github.com/bakdata/streams-bootstrap producer-app: - # from: # While the producer-app does inherit from kafka-app, it does not need a - # `from` section, hence it does not support it. # Allowed configs: # https://github.com/bakdata/streams-bootstrap/tree/master/charts/producer-app values: # required diff --git a/docs/docs/resources/pipeline-defaults/defaults.yaml b/docs/docs/resources/pipeline-defaults/defaults.yaml index 489e3eb5b..78854e93b 100644 --- a/docs/docs/resources/pipeline-defaults/defaults.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults.yaml @@ -3,8 +3,6 @@ # Child of: KafkaApp # More documentation on ConsumerApp: https://github.com/bakdata/streams-bootstrap consumer-app: - # to: # While the consumer-app does inherit from kafka-app, it does not need a - # `to` section, hence it does not support it. # Allowed configs: # https://github.com/bakdata/streams-bootstrap/tree/master/charts/consumer-app values: # required @@ -177,92 +175,11 @@ kafka-app: # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model # Kafka connector # # Parent of: KafkaSinkConnector, KafkaSourceConnector # Child of: PipelineComponent kafka-connector: - # Pipeline prefix that will prefix every component name. If you wish to not - # have any prefix you can specify an empty string. - prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 @@ -275,8 +192,6 @@ kafka-sink-connector: # # Child of: KafkaConnector kafka-source-connector: - # The source connector has no `from` section - # from: # Base Kubernetes App # # Parent of: HelmApp @@ -285,45 +200,6 @@ kubernetes-app: # Pipeline prefix that will prefix every component name. If you wish to not # have any prefix you can specify an empty string. prefix: ${pipeline.name}- - from: # Must not be null - topics: # read from topic - ${pipeline.name}-input-topic: - type: input # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra - ${pipeline.name}-input-pattern-topic: - type: pattern # Implied to be an input pattern if `label` is undefined - ${pipeline.name}-extra-pattern-topic: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - components: # read from specific component - account-producer: - type: input # Implied when role is NOT specified - other-producer: - label: some-role # Implies `type` to be extra - component-as-input-pattern: - type: pattern # Implied to be an input pattern if `label` is undefined - component-as-extra-pattern: - type: pattern # Implied to be an extra pattern if `label` is defined - label: some-role - # Topic(s) into which the component will write output - to: - topics: - ${pipeline.name}-output-topic: - type: output # Implied when role is NOT specified - ${pipeline.name}-extra-topic: - label: topic-role # Implies `type` to be extra; Will throw an error if `type` is defined - ${pipeline.name}-error-topic: - type: error - # Currently KPOps supports Avro and JSON schemas. - key_schema: key-schema # must implement SchemaProvider to use - value_schema: value-schema - partitions_count: 1 - replication_factor: 1 - configs: # https://kafka.apache.org/documentation/#topicconfigs - cleanup.policy: compact - models: # SchemaProvider is initiated with the values given here - model: model namespace: namespace # required values: # required image: exampleImage # Example @@ -334,8 +210,6 @@ kubernetes-app: # Child of: KafkaApp # More documentation on ProducerApp: https://github.com/bakdata/streams-bootstrap producer-app: - # from: # While the producer-app does inherit from kafka-app, it does not need a - # `from` section, hence it does not support it. # Allowed configs: # https://github.com/bakdata/streams-bootstrap/tree/master/charts/producer-app values: # required diff --git a/docs/docs/schema/defaults.json b/docs/docs/schema/defaults.json index cbbc79e1f..07f6c2832 100644 --- a/docs/docs/schema/defaults.json +++ b/docs/docs/schema/defaults.json @@ -98,7 +98,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -125,11 +132,6 @@ "description": "Timeout for Helm operations to finish", "title": "Timeout" }, - "to": { - "default": null, - "title": "To", - "type": "null" - }, "type": { "const": "consumer-app", "title": "Type", @@ -142,7 +144,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } @@ -777,7 +778,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -828,7 +836,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } @@ -1528,19 +1535,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "anyOf": [ - { - "$ref": "#/$defs/FromSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) and/or components from which the component will read input", - "title": "From" - }, "name": { "description": "Component name", "title": "Name", @@ -1582,18 +1576,6 @@ "description": "Timeout for Helm operations to finish", "title": "Timeout" }, - "to": { - "anyOf": [ - { - "$ref": "#/$defs/ToSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) into which the component will write output" - }, "type": { "const": "helm-app", "title": "Type", @@ -1980,19 +1962,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "anyOf": [ - { - "$ref": "#/$defs/FromSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) and/or components from which the component will read input", - "title": "From" - }, "name": { "description": "Component name", "title": "Name", @@ -2003,18 +1972,6 @@ "description": "Pipeline prefix that will prefix every component name. If you wish to not have any prefix you can specify an empty string.", "title": "Prefix", "type": "string" - }, - "to": { - "anyOf": [ - { - "$ref": "#/$defs/ToSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) into which the component will write output" } }, "required": [ @@ -2092,19 +2049,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "anyOf": [ - { - "$ref": "#/$defs/FromSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) and/or components from which the component will read input", - "title": "From" - }, "name": { "description": "Component name", "title": "Name", @@ -2302,11 +2246,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "default": null, - "title": "From", - "type": "null" - }, "name": { "description": "Component name", "title": "Name", @@ -2366,19 +2305,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "anyOf": [ - { - "$ref": "#/$defs/FromSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) and/or components from which the component will read input", - "title": "From" - }, "name": { "description": "Component name", "title": "Name", @@ -2395,18 +2321,6 @@ "title": "Prefix", "type": "string" }, - "to": { - "anyOf": [ - { - "$ref": "#/$defs/ToSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) into which the component will write output" - }, "values": { "$ref": "#/$defs/KubernetesAppValues", "description": "Kubernetes app values" @@ -2717,19 +2631,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "anyOf": [ - { - "$ref": "#/$defs/FromSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) and/or components from which the component will read input", - "title": "From" - }, "name": { "description": "Component name", "title": "Name", @@ -2740,18 +2641,6 @@ "description": "Pipeline prefix that will prefix every component name. If you wish to not have any prefix you can specify an empty string.", "title": "Prefix", "type": "string" - }, - "to": { - "anyOf": [ - { - "$ref": "#/$defs/ToSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) into which the component will write output" } }, "required": [ @@ -2970,12 +2859,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "default": null, - "description": "Producer doesn't support FromSection", - "title": "From", - "type": "null" - }, "name": { "description": "Component name", "title": "Name", @@ -2993,7 +2876,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -3044,7 +2934,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } @@ -3959,7 +3848,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -4010,7 +3906,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } @@ -4656,19 +4551,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "anyOf": [ - { - "$ref": "#/$defs/FromSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) and/or components from which the component will read input", - "title": "From" - }, "name": { "description": "Component name", "title": "Name", @@ -4686,7 +4568,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -4713,18 +4602,6 @@ "description": "Timeout for Helm operations to finish", "title": "Timeout" }, - "to": { - "anyOf": [ - { - "$ref": "#/$defs/ToSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) into which the component will write output" - }, "values": { "$ref": "#/$defs/StreamsBootstrapValues", "description": "streams-bootstrap Helm values" @@ -4732,7 +4609,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } diff --git a/docs/docs/schema/pipeline.json b/docs/docs/schema/pipeline.json index 783fcd10c..2b6bdd01a 100644 --- a/docs/docs/schema/pipeline.json +++ b/docs/docs/schema/pipeline.json @@ -98,7 +98,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -125,11 +132,6 @@ "description": "Timeout for Helm operations to finish", "title": "Timeout" }, - "to": { - "default": null, - "title": "To", - "type": "null" - }, "type": { "const": "consumer-app", "title": "Type", @@ -142,7 +144,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } @@ -777,7 +778,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -828,7 +836,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } @@ -1528,19 +1535,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "anyOf": [ - { - "$ref": "#/$defs/FromSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) and/or components from which the component will read input", - "title": "From" - }, "name": { "description": "Component name", "title": "Name", @@ -1582,18 +1576,6 @@ "description": "Timeout for Helm operations to finish", "title": "Timeout" }, - "to": { - "anyOf": [ - { - "$ref": "#/$defs/ToSection" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Topic(s) into which the component will write output" - }, "type": { "const": "helm-app", "title": "Type", @@ -2124,11 +2106,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "default": null, - "title": "From", - "type": "null" - }, "name": { "description": "Component name", "title": "Name", @@ -2668,12 +2645,6 @@ "title": "Enabled", "type": "boolean" }, - "from": { - "default": null, - "description": "Producer doesn't support FromSection", - "title": "From", - "type": "null" - }, "name": { "description": "Component name", "title": "Name", @@ -2691,7 +2662,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -2742,7 +2720,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } @@ -3657,7 +3634,14 @@ "type": "string" }, "repo_config": { - "$ref": "#/$defs/HelmRepoConfig", + "anyOf": [ + { + "$ref": "#/$defs/HelmRepoConfig" + }, + { + "type": "null" + } + ], "default": { "repo_auth_flags": { "ca_file": null, @@ -3708,7 +3692,6 @@ "version": { "default": "3.6.1", "description": "Helm chart version", - "pattern": "^(\\d+)\\.(\\d+)\\.(\\d+)(-[a-zA-Z]+(\\.[a-zA-Z]+)?)?$", "title": "Version", "type": "string" } diff --git a/hooks/gen_docs/gen_docs_components.py b/hooks/gen_docs/gen_docs_components.py index 9847d0ce5..41f1e1a59 100644 --- a/hooks/gen_docs/gen_docs_components.py +++ b/hooks/gen_docs/gen_docs_components.py @@ -12,27 +12,43 @@ from kpops.components.base_components.kafka_connector import KafkaConnector from kpops.components.base_components.pipeline_component import ( PipelineComponent, + leading_keys_first, ) from kpops.core.registry import Registry registry = Registry() registry.discover_components() + +def generic_origin(component: type[PipelineComponent]) -> type[PipelineComponent]: + """Map a parametrized generic component class to its unparametrized origin.""" + origin = component.__pydantic_generic_metadata__["origin"] + if origin is not None and issubclass(origin, PipelineComponent): + return origin + return component + + PATH_DOCS_RESOURCES = ROOT / "docs/docs/resources" KPOPS_COMPONENTS = tuple(registry.components) KPOPS_COMPONENTS_SECTIONS = { component.type: [ field_name - for field_name, field_info in component.model_fields.items() - if not field_info.exclude + for field_name, _ in leading_keys_first( + ( + (field_name, field_info) + for field_name, field_info in component.model_fields.items() + if not field_info.exclude + ), + key=lambda field: field[1].serialization_alias or field[0], + ) ] for component in KPOPS_COMPONENTS } KPOPS_COMPONENTS_INHERITANCE_REF = { component.type: { "bases": [ - base.type + generic_origin(base).type for base in component.__bases__ if issubclass(base, PipelineComponent) ], diff --git a/kpops/components/base_components/__init__.py b/kpops/components/base_components/__init__.py index 39dfabd0c..524971113 100644 --- a/kpops/components/base_components/__init__.py +++ b/kpops/components/base_components/__init__.py @@ -1,5 +1,5 @@ from .helm_app import HelmApp -from .kafka_app import KafkaApp +from .kafka_app import ConsumesTopics, KafkaApp, ProducesTopics from .kafka_connector import ( KafkaConnector, KafkaSinkConnector, @@ -9,6 +9,7 @@ from .pipeline_component import PipelineComponent __all__ = ( + "ConsumesTopics", "HelmApp", "KafkaApp", "KafkaConnector", @@ -16,4 +17,5 @@ "KafkaSourceConnector", "KubernetesApp", "PipelineComponent", + "ProducesTopics", ) diff --git a/kpops/components/base_components/base_defaults_component.py b/kpops/components/base_components/base_defaults_component.py index f33444b4e..44810f354 100644 --- a/kpops/components/base_components/base_defaults_component.py +++ b/kpops/components/base_components/base_defaults_component.py @@ -99,8 +99,9 @@ def parents(cls: type[Self]) -> tuple[type[BaseDefaultsComponent], ...]: # ty: def gen_parents() -> Generator[type[BaseDefaultsComponent]]: for base in cls.mro(): - # skip class itself and non-component ancestors - if base is cls or not issubclass(base, BaseDefaultsComponent): + # skip class itself, capability mixins, non-component ancestors + # and pydantic's parametrized generic classes + if base is cls or not is_component_type(base): continue yield base @@ -192,6 +193,18 @@ def _validate_custom(self) -> None: """Run custom validation on component.""" +def is_component_type(cls: type) -> bool: + """Whether a class is a KPOps component type (pipeline `type`, defaults key, registry entry). + + Excludes capability mixins and pydantic's parametrized generic classes. + """ + if not issubclass(cls, BaseDefaultsComponent): + return False + if cls.__pydantic_generic_metadata__["origin"] is not None: + return False + return not cls.__dict__.get("_kpops_capability", False) + + def defaults_from_yaml(path: Path, key: str) -> dict[str, Any]: """Read component-specific settings from a ``defaults*.yaml`` file and return @default if not found. diff --git a/kpops/components/base_components/cleaner.py b/kpops/components/base_components/cleaner.py index 243a3afa3..a61c930c3 100644 --- a/kpops/components/base_components/cleaner.py +++ b/kpops/components/base_components/cleaner.py @@ -6,11 +6,11 @@ create_helm_name_override, create_helm_release_name, ) -from kpops.components.base_components.helm_app import HelmApp +from kpops.components.base_components.helm_app import HelmApp, HelmAppValues from kpops.config import get_config -class Cleaner(HelmApp, ABC): +class Cleaner[V: HelmAppValues = HelmAppValues](HelmApp[V], ABC): """Generic Helm app for cleaning or resetting.""" suffix: str = "-clean" diff --git a/kpops/components/base_components/helm_app.py b/kpops/components/base_components/helm_app.py index 25a89b497..783b2a7c2 100644 --- a/kpops/components/base_components/helm_app.py +++ b/kpops/components/base_components/helm_app.py @@ -65,7 +65,7 @@ def model_dump(self, **_: Any) -> dict[str, Any]: ) -class HelmApp(KubernetesApp): +class HelmApp[V: HelmAppValues = HelmAppValues](KubernetesApp[V]): """Kubernetes app managed through Helm with an associated Helm chart. :param repo_config: Configuration of the Helm chart repo to be used for @@ -81,17 +81,23 @@ class HelmApp(KubernetesApp): diff_config: SkipGenerate[HelmDiffConfig] = HelmDiffConfig() version: str | None = None timeout: str | None = None - values: HelmAppValues + values: V + + @property + def helm_repo_config(self) -> HelmRepoConfig | None: + """Helm chart repo configuration used for Helm operations.""" + return self.repo_config @cached_property def _helm(self) -> Helm: """Helm object that contains component-specific config such as repo.""" helm = Helm(get_config().helm_config) - if self.repo_config is not None: + repo_config = self.helm_repo_config + if repo_config is not None: helm.add_repo( - self.repo_config.repository_name, - self.repo_config.url, - self.repo_config.repo_auth_flags, + repo_config.repository_name, + repo_config.url, + repo_config.repo_auth_flags, ) return helm @@ -127,9 +133,8 @@ def helm_chart(self) -> str: @property def helm_flags(self) -> HelmFlags: """Return shared flags for Helm commands.""" - auth_flags = ( - self.repo_config.repo_auth_flags.model_dump() if self.repo_config else {} - ) + repo_config = self.helm_repo_config + auth_flags = repo_config.repo_auth_flags.model_dump() if repo_config else {} effective_timeout = ( self.timeout or get_config().helm_config.timeout or HelmFlags().timeout ) diff --git a/kpops/components/base_components/kafka_app.py b/kpops/components/base_components/kafka_app.py index 775e7a663..1c622ba8f 100644 --- a/kpops/components/base_components/kafka_app.py +++ b/kpops/components/base_components/kafka_app.py @@ -1,12 +1,25 @@ from __future__ import annotations from abc import ABC -from typing import override +from collections.abc import Iterator +from typing import ClassVar, override import structlog +from pydantic import AliasChoices, Field from kpops.component_handlers import get_handlers +from kpops.components.base_components.models.from_section import ( + FromSection, + FromTopic, + InputTopicTypes, +) +from kpops.components.base_components.models.to_section import ToSection from kpops.components.base_components.pipeline_component import PipelineComponent +from kpops.components.common.topic import ( + KafkaTopic, + OutputTopicTypes, + TopicConfig, +) log = structlog.get_logger("KafkaApp") @@ -14,6 +27,195 @@ class KafkaApp(PipelineComponent, ABC): """Base component for Kafka-based components.""" + +class ConsumesTopics(KafkaApp, ABC): + """Capability of a Kafka app to read input from topics and components. + + :param from_: Topic(s) and/or components from which the component will read + input, defaults to None + """ + + _kpops_capability: ClassVar[bool] = True + + from_: FromSection | None = Field( + default=None, + serialization_alias="from", + validation_alias=AliasChoices("from", "from_"), + title="From", + ) + + @override + def _apply_topic_sections(self) -> None: + self.set_input_topics() + super()._apply_topic_sections() + + @property + @override + def inputs(self) -> Iterator[KafkaTopic]: + yield from self.input_topics + for labeled_topics in self.extra_input_topics.values(): + yield from labeled_topics + + @property + def input_topics(self) -> list[KafkaTopic]: + """Get all the input topics from config.""" + return [] + + @property + def extra_input_topics(self) -> dict[str, list[KafkaTopic]]: + """Get extra input topics list from config.""" + return {} + + def add_input_topics(self, topics: list[KafkaTopic]) -> None: + """Add given topics to the list of input topics. + + :param topics: Input topics + """ + + def add_extra_input_topics(self, label: str, topics: list[KafkaTopic]) -> None: + """Add given extra topics that share a label to the list of extra input topics. + + :param topics: Extra input topics + :param label: Topic label + """ + + def set_input_pattern(self, name: str) -> None: + """Set input pattern. + + :param name: Input pattern name + """ + + def add_extra_input_pattern(self, label: str, topic: str) -> None: + """Add an input pattern of type extra. + + :param label: Custom identifier belonging to one or multiple topics + :param topic: Topic name + """ + + def set_input_topics(self) -> None: + """Put values of config.from into the streams config section of streams-bootstrap. + + Supports extra_input_topics (topics by label) or input_topics. + """ + if self.from_: + for name, topic in self.from_.topics.items(): + self.apply_from_inputs(name, topic) + + def apply_from_inputs(self, name: str, topic: FromTopic) -> None: + """Add a `from` section input to the component config. + + :param name: Name of the field + :param topic: Value of the field + """ + kafka_topic = KafkaTopic(name=name) + match topic.type: + case None if topic.label: + self.add_extra_input_topics(topic.label, [kafka_topic]) + case InputTopicTypes.PATTERN if topic.label: + self.add_extra_input_pattern(topic.label, name) + case InputTopicTypes.PATTERN: + self.set_input_pattern(name) + case _: + self.add_input_topics([kafka_topic]) + + def weave_from_topics( + self, + to: ToSection | None, + from_topic: FromTopic | None = None, + ) -> None: + """Weave output topics of upstream component or from component into config. + + Override this method to apply custom logic + """ + if from_topic is None: + from_topic = FromTopic(type=InputTopicTypes.INPUT) + if not to: + return + input_topics = [ + topic_name + for topic_name, topic_config in to.topics.items() + if topic_config.type != OutputTopicTypes.ERROR and not topic_config.label + ] + for input_topic in input_topics: + self.apply_from_inputs(input_topic, from_topic) + + +class ProducesTopics(KafkaApp, ABC): + """Capability of a Kafka app to write output to topics. + + :param to: Topic(s) into which the component will write output, + defaults to None + """ + + _kpops_capability: ClassVar[bool] = True + + to: ToSection | None = None + + @override + def _apply_topic_sections(self) -> None: + self.set_output_topics() + super()._apply_topic_sections() + + @property + @override + def outputs(self) -> Iterator[KafkaTopic]: + if output_topic := self.output_topic: + yield output_topic + yield from self.extra_output_topics.values() + + @property + def output_topic(self) -> KafkaTopic | None: + """Get output topic from config.""" + return None + + @property + def extra_output_topics(self) -> dict[str, KafkaTopic]: + """Get extra output topics list from config.""" + return {} + + def set_output_topic(self, topic: KafkaTopic) -> None: + """Set output topic. + + :param topic: Output topic + """ + + def set_error_topic(self, topic: KafkaTopic) -> None: + """Set error topic. + + :param topic: Error topic + """ + + def add_extra_output_topic(self, topic: KafkaTopic, label: str) -> None: + """Add an output topic of type extra. + + :param topic: Output topic + :param label: Label that is unique to the extra output topic + """ + + def set_output_topics(self) -> None: + """Put values of `to` section into the producer config section of streams-bootstrap. + + Supports extra_output_topics (topics by label) or output_topics. + """ + if self.to: + for name, topic in self.to.topics.items(): + self.apply_to_outputs(name, topic) + + def apply_to_outputs(self, name: str, topic: TopicConfig) -> None: + """Add a `to` section input to the component config. + + :param name: Name of the field + :param topic: Value of the field + """ + kafka_topic = KafkaTopic(name=name) + match topic.type: + case None if topic.label: + self.add_extra_output_topic(kafka_topic, topic.label) + case OutputTopicTypes.ERROR: + self.set_error_topic(kafka_topic) + case _: + self.set_output_topic(kafka_topic) + @override async def deploy(self, dry_run: bool) -> None: if self.to: diff --git a/kpops/components/base_components/kafka_connector.py b/kpops/components/base_components/kafka_connector.py index 59bdc7b78..2a1defffb 100644 --- a/kpops/components/base_components/kafka_connector.py +++ b/kpops/components/base_components/kafka_connector.py @@ -1,10 +1,10 @@ from __future__ import annotations from abc import ABC -from typing import Any, NoReturn, override +from typing import Any, override import structlog -from pydantic import AliasChoices, Field, PrivateAttr, ValidationInfo, field_validator +from pydantic import PrivateAttr, ValidationInfo, field_validator from kpops.component_handlers import get_handlers from kpops.component_handlers.kafka_connect.model import ( @@ -12,14 +12,13 @@ KafkaConnectorConfig, KafkaConnectorType, ) -from kpops.components.base_components.models.from_section import FromTopic -from kpops.components.base_components.pipeline_component import PipelineComponent +from kpops.components.base_components.kafka_app import ConsumesTopics, ProducesTopics from kpops.components.common.topic import KafkaTopic log = structlog.get_logger("KafkaConnector") -class KafkaConnector(PipelineComponent, ABC): +class KafkaConnector(ProducesTopics, ABC): """Base class for all Kafka connectors. Should only be used to set defaults @@ -52,13 +51,7 @@ def connector_config_should_have_component_name( @override async def deploy(self, dry_run: bool) -> None: """Deploy Kafka Connector (Source/Sink). Create output topics and register schemas if configured.""" - if self.to: - for topic in self.to.kafka_topics: - await get_handlers().topic_handler.create_topic(topic, dry_run=dry_run) - - if schema_handler := get_handlers().schema_handler: - await schema_handler.submit_schemas(to_section=self.to, dry_run=dry_run) - + await super().deploy(dry_run) await get_handlers().connector_handler.create_connector( self.config, state=self.state, dry_run=dry_run ) @@ -92,21 +85,10 @@ async def clean(self, dry_run: bool) -> None: class KafkaSourceConnector(KafkaConnector): """Kafka source connector model.""" - from_: None = Field( - default=None, - serialization_alias="from", - validation_alias=AliasChoices("from", "from_"), - title="From", - ) _connector_type: KafkaConnectorType = PrivateAttr(KafkaConnectorType.SOURCE) - @override - def apply_from_inputs(self, name: str, topic: FromTopic) -> NoReturn: - msg = "Kafka source connector doesn't support FromSection" - raise NotImplementedError(msg) - -class KafkaSinkConnector(KafkaConnector): +class KafkaSinkConnector(ConsumesTopics, KafkaConnector): """Kafka sink connector model.""" _connector_type: KafkaConnectorType = PrivateAttr(KafkaConnectorType.SINK) diff --git a/kpops/components/base_components/kubernetes_app.py b/kpops/components/base_components/kubernetes_app.py index 88a989f99..243414a9b 100644 --- a/kpops/components/base_components/kubernetes_app.py +++ b/kpops/components/base_components/kubernetes_app.py @@ -25,7 +25,9 @@ class KubernetesAppValues(CamelCaseConfigModel, DescConfigModel): ) -class KubernetesApp(PipelineComponent, ABC): +class KubernetesApp[V: KubernetesAppValues = KubernetesAppValues]( + PipelineComponent, ABC +): """Base class for all Kubernetes apps. All built-in components are Kubernetes apps, except for the Kafka connectors. @@ -35,7 +37,7 @@ class KubernetesApp(PipelineComponent, ABC): """ namespace: str - values: KubernetesAppValues + values: V @override def _validate_custom(self) -> None: diff --git a/kpops/components/base_components/pipeline_component.py b/kpops/components/base_components/pipeline_component.py index 90799ee8b..afb0735ac 100644 --- a/kpops/components/base_components/pipeline_component.py +++ b/kpops/components/base_components/pipeline_component.py @@ -1,13 +1,12 @@ from __future__ import annotations from abc import ABC -from collections.abc import Iterator -from types import NoneType +from collections.abc import Callable, Iterable, Iterator +from operator import itemgetter from typing import Any, ClassVar import pydantic from pydantic import ( - AliasChoices, ConfigDict, Field, SerializationInfo, @@ -17,22 +16,31 @@ from kpops.components.base_components.base_defaults_component import ( BaseDefaultsComponent, ) -from kpops.components.base_components.models.from_section import ( - FromSection, - FromTopic, - InputTopicTypes, -) -from kpops.components.base_components.models.to_section import ( - ToSection, -) -from kpops.components.common.topic import ( - KafkaTopic, - OutputTopicTypes, - TopicConfig, -) +from kpops.components.common.topic import KafkaTopic from kpops.manifests.kubernetes import KubernetesManifest from kpops.utils.pydantic import exclude_by_name, exclude_by_value +GENERATE_LEADING_KEYS = ("enabled", "prefix", "from", "to") + + +def leading_keys_first[T](items: Iterable[T], key: Callable[[T], str] = str) -> list[T]: + """Keep common keys at the top, independent of the MRO-based field order. + + capability fields such as `from`/`to` are declared on mixins late in the MRO. + + :param items: Items to order + :param key: Maps an item to its serialized key name + :returns: The items with leading keys first, remaining order unchanged + """ + return sorted( + items, + key=lambda item: ( + GENERATE_LEADING_KEYS.index(name) + if (name := key(item)) in GENERATE_LEADING_KEYS + else len(GENERATE_LEADING_KEYS) + ), + ) + class PipelineComponent(BaseDefaultsComponent, ABC): """Base class for all components. @@ -41,10 +49,6 @@ class PipelineComponent(BaseDefaultsComponent, ABC): :param prefix: Pipeline prefix that will prefix every component name. If you wish to not have any prefix you can specify an empty string., defaults to "${pipeline.name}-" - :param from_: Topic(s) and/or components from which the component will read - input, defaults to None - :param to: Topic(s) into which the component will write output, - defaults to None """ name: str @@ -54,13 +58,6 @@ class PipelineComponent(BaseDefaultsComponent, ABC): description="Whether the component is enabled and should be included in the pipeline", ) prefix: str = "${pipeline.name}-" - from_: FromSection | None = Field( - default=None, - serialization_alias="from", - validation_alias=AliasChoices("from", "from_"), - title="From", - ) - to: ToSection | None = None model_config: ClassVar[ConfigDict] = ConfigDict( extra="allow", use_enum_values=False @@ -80,12 +77,31 @@ def sort_model( # NOTE: from SerializeAsOptionalModel if info.exclude_none: result = exclude_by_value(result, None) - return {**ordered_fields, **result} + return { + **ordered_fields, + **dict(leading_keys_first(result.items(), key=itemgetter(0))), + } def __init__(self, **kwargs: Any) -> None: super().__init__(**kwargs) - self.set_input_topics() - self.set_output_topics() + self._apply_topic_sections() + + @pydantic.model_validator(mode="before") + @classmethod + def reject_unsupported_topic_sections(cls, data: Any) -> Any: + if isinstance(data, dict): + for key, field_name in ( + ("from", "from_"), + ("from_", "from_"), + ("to", "to"), + ): + if data.get(key) is not None and field_name not in cls.model_fields: + msg = f"Component type '{cls.type}' does not support the `{key.rstrip('_')}` section." + raise ValueError(msg) + return data + + def _apply_topic_sections(self) -> None: + """Apply `from`/`to` sections to the component config.""" @property def id(self) -> str: @@ -98,155 +114,11 @@ def full_name(self) -> str: @property def inputs(self) -> Iterator[KafkaTopic]: - yield from self.input_topics - for labeled_topics in self.extra_input_topics.values(): - yield from labeled_topics + yield from () @property def outputs(self) -> Iterator[KafkaTopic]: - if output_topic := self.output_topic: - yield output_topic - yield from self.extra_output_topics.values() - - @property - def input_topics(self) -> list[KafkaTopic]: - """Get all the input topics from config.""" - return [] - - @property - def extra_input_topics(self) -> dict[str, list[KafkaTopic]]: - """Get extra input topics list from config.""" - return {} - - @property - def output_topic(self) -> KafkaTopic | None: - """Get output topic from config.""" - return None - - @property - def extra_output_topics(self) -> dict[str, KafkaTopic]: - """Get extra output topics list from config.""" - return {} - - def add_input_topics(self, topics: list[KafkaTopic]) -> None: - """Add given topics to the list of input topics. - - :param topics: Input topics - """ - - def add_extra_input_topics(self, label: str, topics: list[KafkaTopic]) -> None: - """Add given extra topics that share a label to the list of extra input topics. - - :param topics: Extra input topics - :param label: Topic label - """ - - def set_input_pattern(self, name: str) -> None: - """Set input pattern. - - :param name: Input pattern name - """ - - def add_extra_input_pattern(self, label: str, topic: str) -> None: - """Add an input pattern of type extra. - - :param label: Custom identifier belonging to one or multiple topics - :param topic: Topic name - """ - - def set_output_topic(self, topic: KafkaTopic) -> None: - """Set output topic. - - :param topic: Output topic - """ - - def set_error_topic(self, topic: KafkaTopic) -> None: - """Set error topic. - - :param topic: Error topic - """ - - def add_extra_output_topic(self, topic: KafkaTopic, label: str) -> None: - """Add an output topic of type extra. - - :param topic: Output topic - :param label: Label that is unique to the extra output topic - """ - - @classmethod - def supports_from_section(cls) -> bool: - return cls.model_fields["from_"].annotation is not NoneType - - def set_input_topics(self) -> None: - """Put values of config.from into the streams config section of streams-bootstrap. - - Supports extra_input_topics (topics by label) or input_topics. - """ - if self.from_: - for name, topic in self.from_.topics.items(): - self.apply_from_inputs(name, topic) - - def apply_from_inputs(self, name: str, topic: FromTopic) -> None: - """Add a `from` section input to the component config. - - :param name: Name of the field - :param topic: Value of the field - """ - kafka_topic = KafkaTopic(name=name) - match topic.type: - case None if topic.label: - self.add_extra_input_topics(topic.label, [kafka_topic]) - case InputTopicTypes.PATTERN if topic.label: - self.add_extra_input_pattern(topic.label, name) - case InputTopicTypes.PATTERN: - self.set_input_pattern(name) - case _: - self.add_input_topics([kafka_topic]) - - def set_output_topics(self) -> None: - """Put values of `to` section into the producer config section of streams-bootstrap. - - Supports extra_output_topics (topics by label) or output_topics. - """ - if self.to: - for name, topic in self.to.topics.items(): - self.apply_to_outputs(name, topic) - - def apply_to_outputs(self, name: str, topic: TopicConfig) -> None: - """Add a `to` section input to the component config. - - :param name: Name of the field - :param topic: Value of the field - """ - kafka_topic = KafkaTopic(name=name) - match topic.type: - case None if topic.label: - self.add_extra_output_topic(kafka_topic, topic.label) - case OutputTopicTypes.ERROR: - self.set_error_topic(kafka_topic) - case _: - self.set_output_topic(kafka_topic) - - def weave_from_topics( - self, - to: ToSection | None, - from_topic: FromTopic | None = None, - ) -> None: - """Weave output topics of upstream component or from component into config. - - Override this method to apply custom logic - """ - if from_topic is None: - from_topic = FromTopic(type=InputTopicTypes.INPUT) - if not to: - return - input_topics = [ - topic_name - for topic_name, topic_config in to.topics.items() - if topic_config.type != OutputTopicTypes.ERROR and not topic_config.label - ] - for input_topic in input_topics: - self.apply_from_inputs(input_topic, from_topic) + yield from () def inflate(self) -> list[PipelineComponent]: """Inflate component. diff --git a/kpops/components/streams_bootstrap/base.py b/kpops/components/streams_bootstrap/base.py index db5ed0adb..9641a5945 100644 --- a/kpops/components/streams_bootstrap/base.py +++ b/kpops/components/streams_bootstrap/base.py @@ -2,14 +2,14 @@ import re from abc import ABC -from typing import Self, override +from typing import Any, Self, override import pydantic import structlog -from pydantic import Field +from pydantic.json_schema import SkipJsonSchema from kpops.component_handlers.helm.model import HelmRepoConfig -from kpops.components.base_components import KafkaApp +from kpops.components.base_components import KafkaApp, ProducesTopics from kpops.components.base_components.cleaner import Cleaner from kpops.components.base_components.helm_app import HelmApp from kpops.components.streams_bootstrap.model import StreamsBootstrapValues @@ -30,7 +30,9 @@ log = structlog.get_logger("StreamsBootstrap") -class StreamsBootstrap(KafkaApp, HelmApp, ABC): +class StreamsBootstrap[V: StreamsBootstrapValues[Any] = StreamsBootstrapValues]( + KafkaApp, HelmApp[V], ABC +): """Base for components with a streams-bootstrap Helm chart. :param values: streams-bootstrap Helm values @@ -39,16 +41,26 @@ class StreamsBootstrap(KafkaApp, HelmApp, ABC): :param version: Helm chart version, defaults to "3.6.1" """ - values: StreamsBootstrapValues - repo_config: SkipGenerate[HelmRepoConfig] = STREAMS_BOOTSTRAP_HELM_REPO - version: str = Field( - default=STREAMS_BOOTSTRAP_VERSION, - pattern=STREAMS_BOOTSTRAP_VERSION_PATTERN, - ) + values: V + repo_config: SkipGenerate[HelmRepoConfig | None] = STREAMS_BOOTSTRAP_HELM_REPO + version: str | SkipJsonSchema[None] = STREAMS_BOOTSTRAP_VERSION + + @property + @override + def helm_repo_config(self) -> HelmRepoConfig: + return self.repo_config or STREAMS_BOOTSTRAP_HELM_REPO + + @property + def repository_name(self) -> str: + return self.helm_repo_config.repository_name @pydantic.field_validator("version", mode="after") @classmethod - def version_validator(cls, version: str) -> str: + def version_validator(cls, version: str | None) -> str: + if version is None: + msg = "The streams-bootstrap component requires a chart version." + raise ValueError(msg) + pattern_match = COMPILED_VERSION_PATTERN.match(version) if not pattern_match: @@ -78,7 +90,7 @@ def warning_for_latest_image_tag(self) -> Self: @override def manifest_deploy(self) -> tuple[KubernetesManifest, ...]: resource = super().manifest_deploy() - if self.to: + if isinstance(self, ProducesTopics) and self.to: resource = resource + tuple( StrimziKafkaTopic.from_topic(topic) for topic in self.to.kafka_topics ) @@ -87,21 +99,20 @@ def manifest_deploy(self) -> tuple[KubernetesManifest, ...]: @override def manifest_destroy(self) -> tuple[KubernetesManifest, ...]: - if self.to: + if isinstance(self, ProducesTopics) and self.to: return tuple( StrimziKafkaTopic.from_topic(topic) for topic in self.to.kafka_topics ) return () -class StreamsBootstrapCleaner(Cleaner, ABC): +class StreamsBootstrapCleaner[V: StreamsBootstrapValues[Any] = StreamsBootstrapValues]( + Cleaner[V], ABC +): """Helm app for resetting and cleaning a streams-bootstrap app.""" - from_: None = None - to: None = None - @classmethod - def from_parent(cls, parent: StreamsBootstrap) -> Self: + def from_parent(cls, parent: StreamsBootstrap[Any]) -> Self: parent_kwargs = parent.model_dump( by_alias=True, exclude_none=True, diff --git a/kpops/components/streams_bootstrap/consumer/consumer_app.py b/kpops/components/streams_bootstrap/consumer/consumer_app.py index af6e16d2b..2f8085d80 100644 --- a/kpops/components/streams_bootstrap/consumer/consumer_app.py +++ b/kpops/components/streams_bootstrap/consumer/consumer_app.py @@ -2,9 +2,10 @@ from typing import override import structlog -from pydantic import Field, ValidationError +from pydantic import ValidationError from kpops.component_handlers.kubernetes.pvc_handler import PVCHandler +from kpops.components.base_components import ConsumesTopics from kpops.components.base_components.helm_app import HelmApp from kpops.components.common.app_type import AppType from kpops.components.common.topic import KafkaTopic @@ -22,15 +23,15 @@ log = structlog.get_logger("ConsumerApp") -class ConsumerAppCleaner(StreamsBootstrapCleaner, StreamsBootstrap): +class ConsumerAppCleaner( + StreamsBootstrapCleaner[ConsumerAppValues], StreamsBootstrap[ConsumerAppValues] +): values: ConsumerAppValues @property @override def helm_chart(self) -> str: - return ( - f"{self.repo_config.repository_name}/{AppType.CLEANUP_CONSUMER_APP.value}" - ) + return f"{self.repository_name}/{AppType.CLEANUP_CONSUMER_APP.value}" @override async def reset(self, dry_run: bool) -> None: @@ -79,18 +80,13 @@ async def clean_pvcs(self, dry_run: bool) -> None: await pvc_handler.delete_pvcs(dry_run) -class ConsumerApp(StreamsBootstrap): +class ConsumerApp(ConsumesTopics, StreamsBootstrap[ConsumerAppValues]): """ConsumerApp component that configures a streams-bootstrap consumer-app. :param values: streams-bootstrap Helm values """ values: ConsumerAppValues - to: None = Field( - default=None, - alias="to", - title="To", - ) @cached_property def _cleaner(self) -> ConsumerAppCleaner: @@ -125,7 +121,7 @@ def add_extra_input_pattern(self, label: str, topic: str) -> None: @property @override def helm_chart(self) -> str: - return f"{self.repo_config.repository_name}/{AppType.CONSUMER_APP.value}" + return f"{self.repository_name}/{AppType.CONSUMER_APP.value}" @override async def destroy(self, dry_run: bool) -> None: diff --git a/kpops/components/streams_bootstrap/consumer/model.py b/kpops/components/streams_bootstrap/consumer/model.py index 0f7333051..616ac5b7b 100644 --- a/kpops/components/streams_bootstrap/consumer/model.py +++ b/kpops/components/streams_bootstrap/consumer/model.py @@ -89,7 +89,9 @@ def add_labeled_input_topics(self, label: str, topics: list[KafkaTopic]) -> None ) -class ConsumerAppValues(StreamsBootstrapValues): +class BaseConsumerAppValues[K: ConsumerConfig = ConsumerConfig]( + StreamsBootstrapValues[K] +): """consumer-app configurations. The attributes correspond to keys and values that are used as values for the streams-bootstrap Helm chart. @@ -103,7 +105,6 @@ class ConsumerAppValues(StreamsBootstrapValues): :param termination_grace_period_seconds: Delay for graceful application shutdown in seconds: https://pracucci.com/graceful-shutdown-of-kubernetes-pods.html """ - kafka: ConsumerConfig = ConsumerConfig() autoscaling: StreamsAppAutoScaling | None = None stateful_set: bool = False persistence: PersistenceConfig | None = None @@ -112,3 +113,20 @@ class ConsumerAppValues(StreamsBootstrapValues): termination_grace_period_seconds: int | None = None model_config: ClassVar[ConfigDict] = ConfigDict(extra="allow") + + +class ConsumerAppValues(BaseConsumerAppValues[ConsumerConfig]): + """consumer-app configurations. + + The attributes correspond to keys and values that are used as values for the streams-bootstrap Helm chart. + + :param kafka: consumer-app kafka section + :param autoscaling: Kubernetes event-driven autoscaling config, defaults to None + :param stateful_set: Whether to use a StatefulSet instead of a Deployment to deploy the consumer app. + :param persistence: Configuration for persistent volume to store the state of the consumer app. + :param prometheus: Configuration for Prometheus JMX Exporter. + :param jmx: Configuration for JMX Exporter. + :param termination_grace_period_seconds: Delay for graceful application shutdown in seconds: https://pracucci.com/graceful-shutdown-of-kubernetes-pods.html + """ + + kafka: ConsumerConfig = ConsumerConfig() diff --git a/kpops/components/streams_bootstrap/consumer_producer/consumer_producer_app.py b/kpops/components/streams_bootstrap/consumer_producer/consumer_producer_app.py index 7d7be2257..471ffd040 100644 --- a/kpops/components/streams_bootstrap/consumer_producer/consumer_producer_app.py +++ b/kpops/components/streams_bootstrap/consumer_producer/consumer_producer_app.py @@ -5,6 +5,7 @@ from pydantic import ValidationError from kpops.component_handlers.kubernetes.pvc_handler import PVCHandler +from kpops.components.base_components import ConsumesTopics, ProducesTopics from kpops.components.base_components.helm_app import HelmApp from kpops.components.common.app_type import AppType from kpops.components.common.topic import KafkaTopic @@ -25,15 +26,16 @@ log = structlog.get_logger("ConsumerProducerApp") -class ConsumerProducerAppCleaner(StreamsBootstrapCleaner, StreamsBootstrap): - from_: None = None - to: None = None +class ConsumerProducerAppCleaner( + StreamsBootstrapCleaner[ConsumerProducerAppValues], + StreamsBootstrap[ConsumerProducerAppValues], +): values: ConsumerProducerAppValues @property @override def helm_chart(self) -> str: - return f"{self.repo_config.repository_name}/{AppType.CLEANUP_CONSUMER_PRODUCER_APP.value}" + return f"{self.repository_name}/{AppType.CLEANUP_CONSUMER_PRODUCER_APP.value}" @override async def reset(self, dry_run: bool) -> None: @@ -85,7 +87,9 @@ async def clean_pvcs(self, dry_run: bool) -> None: await pvc_handler.delete_pvcs(dry_run) -class ConsumerProducerApp(StreamsBootstrap): +class ConsumerProducerApp( + ConsumesTopics, ProducesTopics, StreamsBootstrap[ConsumerProducerAppValues] +): """ConsumerProducerApp component that configures a streams-bootstrap consumerproducer-app. :param values: streams-bootstrap Helm values @@ -148,9 +152,7 @@ def add_extra_output_topic(self, topic: KafkaTopic, label: str) -> None: @property @override def helm_chart(self) -> str: - return ( - f"{self.repo_config.repository_name}/{AppType.CONSUMER_PRODUCER_APP.value}" - ) + return f"{self.repository_name}/{AppType.CONSUMER_PRODUCER_APP.value}" @override async def destroy(self, dry_run: bool) -> None: diff --git a/kpops/components/streams_bootstrap/consumer_producer/model.py b/kpops/components/streams_bootstrap/consumer_producer/model.py index 06223cd90..668d62fe2 100644 --- a/kpops/components/streams_bootstrap/consumer_producer/model.py +++ b/kpops/components/streams_bootstrap/consumer_producer/model.py @@ -1,6 +1,6 @@ from kpops.components.common.topic import KafkaTopicStr from kpops.components.streams_bootstrap.consumer.model import ( - ConsumerAppValues, + BaseConsumerAppValues, ConsumerConfig, ) @@ -16,7 +16,7 @@ class ConsumerProducerConfig(ConsumerConfig): delete_output: bool | None = None -class ConsumerProducerAppValues(ConsumerAppValues): +class ConsumerProducerAppValues(BaseConsumerAppValues[ConsumerProducerConfig]): """consumerproducer-app configurations. The attributes correspond to keys and values that are used as values for the streams-bootstrap Helm chart. diff --git a/kpops/components/streams_bootstrap/model.py b/kpops/components/streams_bootstrap/model.py index 869a3e9aa..3eb1a3967 100644 --- a/kpops/components/streams_bootstrap/model.py +++ b/kpops/components/streams_bootstrap/model.py @@ -68,7 +68,55 @@ class JavaOptions(CamelCaseConfigModel, DescConfigModel): others: list[str] = [] -class StreamsBootstrapValues(SerializeAsOptionalModel, HelmAppValues): +def serialize_labeled_output_topics( + labeled_output_topics: dict[str, KafkaTopic], +) -> dict[str, str]: + return {label: topic.name for label, topic in labeled_output_topics.items()} + + +class KafkaConfig(SerializeAsOptionalModel, CamelCaseConfigModel, DescConfigModel): + """Kafka Streams config. + + :param bootstrap_servers: Brokers + :param schema_registry_url: URL of the Schema Registry, defaults to None + :param labeled_output_topics: Extra output topics + :param output_topic: Output topic, defaults to None + """ + + bootstrap_servers: str = "${config.kafka_brokers}" + schema_registry_url: str | None = Field( + default=None, + validation_alias=AliasChoices( + "schemaRegistryUrl", "schema_registry_url" + ), # TODO: same for other camelcase fields, avoids duplicates during enrichment + title="Schema Registry URL", + ) + labeled_output_topics: SerializeAsOptional[ + Annotated[ + dict[str, KafkaTopicStr], + pydantic.PlainSerializer(serialize_labeled_output_topics), + ] + ] = {} + output_topic: KafkaTopicStr | None = None + + model_config: ClassVar[ConfigDict] = ConfigDict(extra="allow") + + @pydantic.field_validator("labeled_output_topics", mode="before") + @classmethod + def deserialize_labeled_output_topics( + cls, labeled_output_topics: dict[str, str] | Any + ) -> dict[str, KafkaTopic] | Any: + if isinstance(labeled_output_topics, dict): + return { + label: KafkaTopic(name=topic_name) + for label, topic_name in labeled_output_topics.items() + } + return labeled_output_topics + + +class StreamsBootstrapValues[K: KafkaConfig = KafkaConfig]( + SerializeAsOptionalModel, HelmAppValues +): """Base value class for all streams-bootstrap related components. :param image: Docker image of the Kafka producer app. @@ -99,7 +147,7 @@ class StreamsBootstrapValues(SerializeAsOptionalModel, HelmAppValues): ) image_pull_policy: ImagePullPolicy | None = None image_pull_secrets: SerializeAsOptional[list[dict[str, str]]] = [] - kafka: KafkaConfig + kafka: K resources: Resources | None = None ports: SerializeAsOptional[list[PortConfig]] = [] service: ServiceConfig | None = None @@ -126,49 +174,3 @@ def unsupported_attributes(cls, values: Any) -> Any: msg = f"streams-bootstrap v3 no longer supports '{attr}' attribute." raise ValueError(msg) return values - - -def serialize_labeled_output_topics( - labeled_output_topics: dict[str, KafkaTopic], -) -> dict[str, str]: - return {label: topic.name for label, topic in labeled_output_topics.items()} - - -class KafkaConfig(SerializeAsOptionalModel, CamelCaseConfigModel, DescConfigModel): - """Kafka Streams config. - - :param bootstrap_servers: Brokers - :param schema_registry_url: URL of the Schema Registry, defaults to None - :param labeled_output_topics: Extra output topics - :param output_topic: Output topic, defaults to None - """ - - bootstrap_servers: str = "${config.kafka_brokers}" - schema_registry_url: str | None = Field( - default=None, - validation_alias=AliasChoices( - "schemaRegistryUrl", "schema_registry_url" - ), # TODO: same for other camelcase fields, avoids duplicates during enrichment - title="Schema Registry URL", - ) - labeled_output_topics: SerializeAsOptional[ - Annotated[ - dict[str, KafkaTopicStr], - pydantic.PlainSerializer(serialize_labeled_output_topics), - ] - ] = {} - output_topic: KafkaTopicStr | None = None - - model_config: ClassVar[ConfigDict] = ConfigDict(extra="allow") - - @pydantic.field_validator("labeled_output_topics", mode="before") - @classmethod - def deserialize_labeled_output_topics( - cls, labeled_output_topics: dict[str, str] | Any - ) -> dict[str, KafkaTopic] | Any: - if isinstance(labeled_output_topics, dict): - return { - label: KafkaTopic(name=topic_name) - for label, topic_name in labeled_output_topics.items() - } - return labeled_output_topics diff --git a/kpops/components/streams_bootstrap/producer/model.py b/kpops/components/streams_bootstrap/producer/model.py index 0e03ce3c1..e2871317e 100644 --- a/kpops/components/streams_bootstrap/producer/model.py +++ b/kpops/components/streams_bootstrap/producer/model.py @@ -15,7 +15,7 @@ class ProducerConfig(KafkaConfig): """producer-app kafka section.""" -class ProducerAppValues(StreamsBootstrapValues): +class ProducerAppValues(StreamsBootstrapValues[ProducerConfig]): """producer-app configurations. The attributes correspond to keys and values that are used as values for the streams-bootstrap Helm chart. diff --git a/kpops/components/streams_bootstrap/producer/producer_app.py b/kpops/components/streams_bootstrap/producer/producer_app.py index bd0930834..1e3acd0f2 100644 --- a/kpops/components/streams_bootstrap/producer/producer_app.py +++ b/kpops/components/streams_bootstrap/producer/producer_app.py @@ -2,9 +2,10 @@ from typing import override import structlog -from pydantic import Field, ValidationError +from pydantic import ValidationError from kpops.component_handlers.kubernetes.utils import trim +from kpops.components.base_components import ProducesTopics from kpops.components.common.app_type import AppType from kpops.components.common.topic import ( KafkaTopic, @@ -26,15 +27,15 @@ log = structlog.get_logger("ProducerApp") -class ProducerAppCleaner(StreamsBootstrapCleaner, StreamsBootstrap): +class ProducerAppCleaner( + StreamsBootstrapCleaner[ProducerAppValues], StreamsBootstrap[ProducerAppValues] +): values: ProducerAppValues @property @override def helm_chart(self) -> str: - return ( - f"{self.repo_config.repository_name}/{AppType.CLEANUP_PRODUCER_APP.value}" - ) + return f"{self.repository_name}/{AppType.CLEANUP_PRODUCER_APP.value}" @override def manifest_deploy(self) -> tuple[KubernetesManifest, ...]: @@ -52,21 +53,15 @@ def manifest_deploy(self) -> tuple[KubernetesManifest, ...]: ) -class ProducerApp(StreamsBootstrap): +class ProducerApp(ProducesTopics, StreamsBootstrap[ProducerAppValues]): """ProducerApp component that configures a streams-bootstrap producer-app. Note that the producer does not support error topics. :param values: streams-bootstrap Helm values - :param from_: Producer doesn't support FromSection, defaults to None """ values: ProducerAppValues - from_: None = Field( - default=None, - alias="from", - title="From", - ) @property def is_cron_job(self) -> bool: @@ -113,7 +108,7 @@ def add_extra_output_topic(self, topic: KafkaTopic, label: str) -> None: @property @override def helm_chart(self) -> str: - return f"{self.repo_config.repository_name}/{AppType.PRODUCER_APP.value}" + return f"{self.repository_name}/{AppType.PRODUCER_APP.value}" @override async def reset(self, dry_run: bool) -> None: diff --git a/kpops/components/streams_bootstrap/streams/model.py b/kpops/components/streams_bootstrap/streams/model.py index 109e58d06..c96dae5b1 100644 --- a/kpops/components/streams_bootstrap/streams/model.py +++ b/kpops/components/streams_bootstrap/streams/model.py @@ -98,7 +98,7 @@ def add_labeled_input_topics(self, label: str, topics: list[KafkaTopic]) -> None ) -class StreamsAppValues(StreamsBootstrapValues): +class StreamsAppValues(StreamsBootstrapValues[StreamsConfig]): """streams-app configurations. The attributes correspond to keys and values that are used as values for the streams-bootstrap Helm chart. diff --git a/kpops/components/streams_bootstrap/streams/streams_app.py b/kpops/components/streams_bootstrap/streams/streams_app.py index b24973095..275be82d5 100644 --- a/kpops/components/streams_bootstrap/streams/streams_app.py +++ b/kpops/components/streams_bootstrap/streams/streams_app.py @@ -5,6 +5,7 @@ from pydantic import ValidationError from kpops.component_handlers.kubernetes.pvc_handler import PVCHandler +from kpops.components.base_components import ConsumesTopics, ProducesTopics from kpops.components.base_components.helm_app import HelmApp from kpops.components.common.app_type import AppType from kpops.components.common.topic import KafkaTopic @@ -25,15 +26,15 @@ log = structlog.get_logger("StreamsApp") -class StreamsAppCleaner(StreamsBootstrapCleaner, StreamsBootstrap): - from_: None = None - to: None = None +class StreamsAppCleaner( + StreamsBootstrapCleaner[StreamsAppValues], StreamsBootstrap[StreamsAppValues] +): values: StreamsAppValues @property @override def helm_chart(self) -> str: - return f"{self.repo_config.repository_name}/{AppType.CLEANUP_STREAMS_APP.value}" + return f"{self.repository_name}/{AppType.CLEANUP_STREAMS_APP.value}" @override async def reset(self, dry_run: bool) -> None: @@ -85,7 +86,7 @@ async def clean_pvcs(self, dry_run: bool) -> None: await pvc_handler.delete_pvcs(dry_run) -class StreamsApp(StreamsBootstrap): +class StreamsApp(ConsumesTopics, ProducesTopics, StreamsBootstrap[StreamsAppValues]): """StreamsApp component that configures a streams-bootstrap streams-app. :param values: streams-bootstrap Helm values @@ -148,7 +149,7 @@ def add_extra_output_topic(self, topic: KafkaTopic, label: str) -> None: @property @override def helm_chart(self) -> str: - return f"{self.repo_config.repository_name}/{AppType.STREAMS_APP.value}" + return f"{self.repository_name}/{AppType.STREAMS_APP.value}" @override async def destroy(self, dry_run: bool) -> None: diff --git a/kpops/core/registry.py b/kpops/core/registry.py index ce1a01ddf..156c4ecf2 100644 --- a/kpops/core/registry.py +++ b/kpops/core/registry.py @@ -11,7 +11,12 @@ from typing import TYPE_CHECKING, TypeVar import structlog +from pydantic import BaseModel +from kpops.components.base_components.base_defaults_component import ( + BaseDefaultsComponent, + is_component_type, +) from kpops.components.base_components.pipeline_component import PipelineComponent from kpops.const import KPOPS_MODULE from kpops.core.exception import ClassNotFoundError @@ -86,12 +91,24 @@ def _find_classes[PluginT]( ) -> Iterator[type[PluginT]]: for module in modules: for _, _class in inspect.getmembers(module, inspect.isclass): - if not __filter_internal_kpops_classes( - _class.__module__, module.__name__ - ) and issubclass(_class, base): + if ( + not __filter_internal_kpops_classes(_class.__module__, module.__name__) + and issubclass(_class, base) + and not _is_excluded(_class) + ): yield _class +def _is_excluded(_class: type) -> bool: + """Whether a class is not a plugin: a capability mixin or a parametrized generic class.""" + if issubclass(_class, BaseDefaultsComponent): + return not is_component_type(_class) + return ( + issubclass(_class, BaseModel) + and _class.__pydantic_generic_metadata__["origin"] is not None + ) + + def __filter_internal_kpops_classes(class_module: str, module_name: str) -> bool: """Filter out internal kpops classes and components unless specifically requested.""" return class_module.startswith(KPOPS_MODULE) and not module_name.startswith( diff --git a/kpops/pipeline.py b/kpops/pipeline.py index 7a2646119..526f887ec 100644 --- a/kpops/pipeline.py +++ b/kpops/pipeline.py @@ -14,6 +14,10 @@ ) from kpops.component_handlers import ComponentHandlers +from kpops.components.base_components.kafka_app import ( + ConsumesTopics, + ProducesTopics, +) from kpops.components.base_components.pipeline_component import PipelineComponent from kpops.core.exception import KpopsException, ParsingException, ValidationError from kpops.core.registry import Registry @@ -26,6 +30,7 @@ from collections.abc import Awaitable, Coroutine, Iterator from pathlib import Path + from kpops.components.base_components.models.to_section import ToSection from kpops.config import KpopsConfig from kpops.manifests.kubernetes import KubernetesManifest @@ -300,6 +305,10 @@ def __validate_graph(self) -> None: raise ValueError(msg) +def _output_section(component: PipelineComponent) -> ToSection | None: + return component.to if isinstance(component, ProducesTopics) else None + + def create_env_components_index( environment_components: list[dict[str, Any]], ) -> dict[str, dict[str, Any]]: @@ -423,24 +432,30 @@ def _apply_component( return # inflate & enrich components for inflated_component in component.inflate(): # TODO: recursively - if inflated_component.from_: - # read from specified components - for ( - original_from_component_id, - from_topic, - ) in inflated_component.from_.components.items(): - original_from_component = self.pipeline[original_from_component_id] - - inflated_from_component = original_from_component.inflate()[-1] - resolved_from_component = self.pipeline[inflated_from_component.id] - + if isinstance(inflated_component, ConsumesTopics): + if inflated_component.from_: + # read from specified components + for ( + original_from_component_id, + from_topic, + ) in inflated_component.from_.components.items(): + original_from_component = self.pipeline[ + original_from_component_id + ] + + inflated_from_component = original_from_component.inflate()[-1] + resolved_from_component = self.pipeline[ + inflated_from_component.id + ] + + inflated_component.weave_from_topics( + _output_section(resolved_from_component), from_topic + ) + elif self.pipeline: + # read from previous component inflated_component.weave_from_topics( - resolved_from_component.to, from_topic + _output_section(self.pipeline.last) ) - elif self.pipeline and inflated_component.supports_from_section(): - # read from previous component - prev_component = self.pipeline.last - inflated_component.weave_from_topics(prev_component.to) self.pipeline.add(inflated_component) def _enrich_component_with_env( diff --git a/tests/components/streams_bootstrap/test_consumer_app.py b/tests/components/streams_bootstrap/test_consumer_app.py index 6fa27f5e6..99089080a 100644 --- a/tests/components/streams_bootstrap/test_consumer_app.py +++ b/tests/components/streams_bootstrap/test_consumer_app.py @@ -298,7 +298,7 @@ async def test_deploy_order_when_dry_run_is_false( await consumer_app.deploy(dry_run=dry_run) # Ensure no outputs are attached to a ConsumerApp deployment - assert consumer_app.to is None + assert "to" not in type(consumer_app).model_fields assert mock.mock_calls == [ mocker.call.helm_upgrade_install( @@ -620,7 +620,7 @@ async def test_get_input_topics(self) -> None: } # Verify no outputs are registered - assert consumer_app.to is None + assert "to" not in type(consumer_app).model_fields assert not getattr(consumer_app, "outputs", None) or not list( consumer_app.outputs ) diff --git a/tests/components/streams_bootstrap/test_producer_app.py b/tests/components/streams_bootstrap/test_producer_app.py index 2a2ae68ca..a623e2b6a 100644 --- a/tests/components/streams_bootstrap/test_producer_app.py +++ b/tests/components/streams_bootstrap/test_producer_app.py @@ -9,6 +9,7 @@ from kpops.component_handlers.helm.helm import Helm from kpops.component_handlers.helm.model import HelmUpgradeInstallFlags from kpops.component_handlers.helm.utils import create_helm_release_name +from kpops.components.base_components import ConsumesTopics from kpops.components.common.topic import ( KafkaTopic, OutputTopicTypes, @@ -94,7 +95,7 @@ def empty_helm_get_values(self, mocker: MockerFixture) -> MagicMock: return mocker.patch.object(Helm, "get_values", return_value=None) def test_supports_from_section(self) -> None: - assert not ProducerApp.supports_from_section() + assert not issubclass(ProducerApp, ConsumesTopics) def test_helm_name_override(self, producer_app: ProducerApp) -> None: assert len(producer_app.helm_name_override) == 63 @@ -367,7 +368,6 @@ def test_get_output_topics(self) -> None: assert producer_app.values.kafka.labeled_output_topics == { "first-extra-topic": KafkaTopic(name="extra-topic-1") } - assert producer_app.input_topics == [] assert list(producer_app.inputs) == [] assert list(producer_app.outputs) == [ KafkaTopic(name="producer-app-output-topic"), diff --git a/tests/components/streams_bootstrap/test_streams_app.py b/tests/components/streams_bootstrap/test_streams_app.py index c5323322c..eca70a060 100644 --- a/tests/components/streams_bootstrap/test_streams_app.py +++ b/tests/components/streams_bootstrap/test_streams_app.py @@ -229,6 +229,19 @@ def test_no_empty_input_topic(self) -> None: assert "inputPattern" in streams_config assert "extraInputPatterns" not in streams_config + def test_should_raise_validation_error_when_version_is_none(self) -> None: + with pytest.raises(ValidationError, match="requires a chart version"): + StreamsApp.model_validate( + { + "name": STREAMS_APP_NAME, + "namespace": NAMESPACE, + "values": { + "kafka": {"bootstrapServers": "fake-broker:9092"}, + }, + "version": None, + }, + ) + def test_should_validate(self) -> None: # An exception should be raised when both label and type are defined and type is input with pytest.raises( diff --git a/tests/components/streams_bootstrap/test_streams_bootstrap.py b/tests/components/streams_bootstrap/test_streams_bootstrap.py index 1eada00c7..4e6dce0be 100644 --- a/tests/components/streams_bootstrap/test_streams_bootstrap.py +++ b/tests/components/streams_bootstrap/test_streams_bootstrap.py @@ -9,6 +9,7 @@ from kpops.component_handlers.helm.model import ( HelmRepoConfig, HelmUpgradeInstallFlags, + RepoAuthFlags, ) from kpops.component_handlers.helm.utils import create_helm_release_name from kpops.components.common.kubernetes_model import ( @@ -44,6 +45,60 @@ def test_default_configs(self) -> None: assert streams_bootstrap.namespace == "test-namespace" assert streams_bootstrap.values.image_tag is None + def test_should_add_default_helm_repo_when_repo_config_is_none( + self, mocker: MockerFixture + ) -> None: + helm_mock = mocker.patch( + "kpops.components.base_components.helm_app.Helm" + ).return_value + streams_bootstrap = StreamsBootstrap.model_validate( + { + "name": "example-name", + "namespace": "test-namespace", + "values": { + "image": "streamsBootstrap", + "kafka": {"bootstrapServers": "localhost:9092"}, + }, + "repo_config": None, + }, + ) + assert streams_bootstrap.repo_config is None + assert streams_bootstrap.repository_name == "bakdata-streams-bootstrap" + + assert streams_bootstrap._helm is helm_mock + helm_mock.add_repo.assert_called_once_with( + "bakdata-streams-bootstrap", + "https://bakdata.github.io/streams-bootstrap/", + RepoAuthFlags(), + ) + + def test_should_add_configured_helm_repo(self, mocker: MockerFixture) -> None: + helm_mock = mocker.patch( + "kpops.components.base_components.helm_app.Helm" + ).return_value + streams_bootstrap = StreamsBootstrap.model_validate( + { + "name": "example-name", + "namespace": "test-namespace", + "values": { + "image": "streamsBootstrap", + "kafka": {"bootstrapServers": "localhost:9092"}, + }, + "repo_config": HelmRepoConfig( + repository_name="custom-repo", + url="https://example.com/charts/", + ), + }, + ) + assert streams_bootstrap.repository_name == "custom-repo" + + assert streams_bootstrap._helm is helm_mock + helm_mock.add_repo.assert_called_once_with( + "custom-repo", + "https://example.com/charts/", + RepoAuthFlags(), + ) + async def test_should_deploy_streams_bootstrap_app( self, mocker: MockerFixture ) -> None: diff --git a/tests/components/test_component_hierarchy.py b/tests/components/test_component_hierarchy.py new file mode 100644 index 000000000..4831c409f --- /dev/null +++ b/tests/components/test_component_hierarchy.py @@ -0,0 +1,239 @@ +import sys +from typing import Any + +import pytest +from pydantic import ValidationError + +from kpops.components.base_components import ( + ConsumesTopics, + HelmApp, + KafkaSourceConnector, + ProducesTopics, +) +from kpops.components.base_components.pipeline_component import PipelineComponent +from kpops.components.streams_bootstrap.base import StreamsBootstrap +from kpops.components.streams_bootstrap.consumer.consumer_app import ( + ConsumerApp, + ConsumerAppCleaner, +) +from kpops.components.streams_bootstrap.consumer_producer.consumer_producer_app import ( + ConsumerProducerAppCleaner, +) +from kpops.components.streams_bootstrap.model import StreamsBootstrapValues +from kpops.components.streams_bootstrap.producer.model import ( + ProducerAppValues, + ProducerConfig, +) +from kpops.components.streams_bootstrap.producer.producer_app import ( + ProducerApp, + ProducerAppCleaner, +) +from kpops.components.streams_bootstrap.streams.streams_app import StreamsAppCleaner +from kpops.core.registry import Registry, _find_classes, find_class +from tests.components.test_kubernetes_app import DummyKubernetesApp + +registry = Registry() +registry.discover_components() + +EXPECTED_PARENTS: dict[str, tuple[str, ...]] = { + "consumer-app": ( + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), + "consumer-producer-app": ( + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), + "helm-app": ("kubernetes-app", "pipeline-component", "base-defaults-component"), + "kafka-app": ("pipeline-component", "base-defaults-component"), + "kafka-connector": ( + "kafka-app", + "pipeline-component", + "base-defaults-component", + ), + "kafka-sink-connector": ( + "kafka-connector", + "kafka-app", + "pipeline-component", + "base-defaults-component", + ), + "kafka-source-connector": ( + "kafka-connector", + "kafka-app", + "pipeline-component", + "base-defaults-component", + ), + "kubernetes-app": ("pipeline-component", "base-defaults-component"), + "pipeline-component": ("base-defaults-component",), + "producer-app": ( + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), + "streams-app": ( + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), + "streams-bootstrap": ( + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), +} + +EXPECTED_CLEANER_PARENTS: dict[type[PipelineComponent], tuple[str, ...]] = { + StreamsAppCleaner: ( + "streams-bootstrap-cleaner", + "cleaner", + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), + ProducerAppCleaner: ( + "streams-bootstrap-cleaner", + "cleaner", + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), + ConsumerAppCleaner: ( + "streams-bootstrap-cleaner", + "cleaner", + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), + ConsumerProducerAppCleaner: ( + "streams-bootstrap-cleaner", + "cleaner", + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ), +} + + +@pytest.mark.parametrize("component_type", sorted(EXPECTED_PARENTS)) +def test_parents(component_type: str) -> None: + component = registry[component_type] + assert ( + tuple(parent.type for parent in component.parents) + == EXPECTED_PARENTS[component_type] + ) + + +def test_all_components_covered() -> None: + assert {c.type for c in registry.components} == set(EXPECTED_PARENTS) + + +@pytest.mark.parametrize( + ("cleaner", "expected_parents"), + sorted(EXPECTED_CLEANER_PARENTS.items(), key=lambda item: item[0].__name__), +) +def test_cleaner_parents( + cleaner: type[PipelineComponent], expected_parents: tuple[str, ...] +) -> None: + assert tuple(parent.type for parent in cleaner.parents) == expected_parents + + +class CustomStreamsBootstrapApp[ + V: StreamsBootstrapValues[Any] = StreamsBootstrapValues +](StreamsBootstrap[V]): + runs_on_spot: bool = False + + +class CustomProducerApp(ProducerApp, CustomStreamsBootstrapApp[ProducerAppValues]): ... + + +def test_custom_diamond_parents() -> None: + assert tuple(parent.type for parent in CustomProducerApp.parents) == ( + "producer-app", + "custom-streams-bootstrap-app", + "streams-bootstrap", + "kafka-app", + "helm-app", + "kubernetes-app", + "pipeline-component", + "base-defaults-component", + ) + + +def test_custom_diamond_not_registered_as_generic() -> None: + module = sys.modules[__name__] + classes = list(_find_classes([module], base=PipelineComponent)) + assert CustomProducerApp in classes + assert CustomStreamsBootstrapApp in classes + assert not any("[" in cls.__name__ for cls in classes) + assert find_class([module], base=CustomProducerApp) is CustomProducerApp + + +def test_custom_diamond_values() -> None: + component = CustomProducerApp( + name="test", + namespace="test-namespace", + values=ProducerAppValues(image="registry/producer", kafka=ProducerConfig()), + enrich=False, + validate=False, + ) + assert isinstance(component.values, ProducerAppValues) + assert component.runs_on_spot is False + + +@pytest.mark.parametrize( + ("component", "section"), + [ + (HelmApp, "from"), + (HelmApp, "to"), + (DummyKubernetesApp, "from"), + (DummyKubernetesApp, "to"), + (ProducerApp, "from"), + (KafkaSourceConnector, "from"), + (ConsumerApp, "to"), + ], +) +def test_unsupported_topic_section_is_rejected( + component: type[PipelineComponent], section: str +) -> None: + with pytest.raises( + ValidationError, + match=rf"does not support the `{section}` section", + ): + component(name="test", enrich=False, **{section: {"topics": {"topic": {}}}}) + + +def test_capabilities_are_not_component_types() -> None: + capabilities = {ConsumesTopics, ProducesTopics} + registered = set(registry.components) + assert not capabilities & registered + assert len(registered) == 12 + for component in (*registered, *EXPECTED_CLEANER_PARENTS): + assert not capabilities & set(component.parents) diff --git a/tests/components/test_helm_app.py b/tests/components/test_helm_app.py index f9cf391c4..6ab1b98d3 100644 --- a/tests/components/test_helm_app.py +++ b/tests/components/test_helm_app.py @@ -131,8 +131,6 @@ async def test_should_deploy_app_with_local_helm_chart( app_values: HelmAppValues, ) -> None: class AppWithLocalChart(HelmApp): - repo_config: None = None - @property @override def helm_chart(self) -> str: diff --git a/tests/components/test_kafka_sink_connector.py b/tests/components/test_kafka_sink_connector.py index 09cdd0df7..66fb74263 100644 --- a/tests/components/test_kafka_sink_connector.py +++ b/tests/components/test_kafka_sink_connector.py @@ -10,6 +10,7 @@ KafkaConnectorConfig, KafkaConnectorType, ) +from kpops.components.base_components import ConsumesTopics from kpops.components.base_components.kafka_connector import ( KafkaSinkConnector, ) @@ -53,7 +54,7 @@ def connector(self, connector_config: KafkaConnectorConfig) -> KafkaSinkConnecto ) def test_supports_from_section(self) -> None: - assert KafkaSinkConnector.supports_from_section() + assert issubclass(KafkaSinkConnector, ConsumesTopics) def test_connector_config_parsing( self, connector_config: KafkaConnectorConfig diff --git a/tests/components/test_kafka_source_connector.py b/tests/components/test_kafka_source_connector.py index 212238d24..7bac8e728 100644 --- a/tests/components/test_kafka_source_connector.py +++ b/tests/components/test_kafka_source_connector.py @@ -11,6 +11,7 @@ KafkaConnectorConfig, KafkaConnectorType, ) +from kpops.components.base_components import ConsumesTopics from kpops.components.base_components.kafka_connector import ( KafkaSourceConnector, ) @@ -55,17 +56,17 @@ def connector( ) def test_supports_from_section(self) -> None: - assert not KafkaSourceConnector.supports_from_section() + assert not issubclass(KafkaSourceConnector, ConsumesTopics) def test_from_section_raises_exception( self, connector_config: KafkaConnectorConfig, ) -> None: - with pytest.raises(ValidationError): + with pytest.raises(ValidationError, match="does not support"): KafkaSourceConnector( name=CONNECTOR_NAME, config=connector_config, - from_=FromSection( # ty: ignore[invalid-argument-type] + from_=FromSection( topics={ TopicName("connector-topic"): FromTopic( type=InputTopicTypes.INPUT @@ -74,16 +75,6 @@ def test_from_section_raises_exception( ), ) - def test_apply_from_inputs_raises_exception( - self, - connector: KafkaSourceConnector, - ) -> None: - with pytest.raises(NotImplementedError): - connector.apply_from_inputs( - "connector-topic", - FromTopic(type=InputTopicTypes.INPUT), - ) - async def test_deploy_order( self, connector: KafkaSourceConnector,