feat: enable Taps and Targets to generate their own Meltano yaml files

cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/tap.py

@@ -38,6 +38,7 @@ class Tap{{ cookiecutter.source_name }}({{ 'SQL' if cookiecutter.stream_type ==
             "auth_token",
             th.StringType,
             required=True,
+            secret=True,  # Flag config as protected.
             description="The token to authenticate against the API service"
         ),
         th.Property(

cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/target.py

@@ -21,6 +21,7 @@ class Target{{ cookiecutter.destination_name }}({{ target_class }}):
         th.Property(
             "sqlalchemy_url",
             th.StringType,
+            secret=True,  # Flag config as protected.
             description="SQLAlchemy connection string",
         ),
         {%- else %}
@@ -34,6 +35,12 @@ class Target{{ cookiecutter.destination_name }}({{ target_class }}):
             th.StringType,
             description="The scheme with which output files will be named"
         ),
+        th.Property(
+            "auth_token",
+            th.StringType,
+            secret=True,  # Flag config as protected.
+            description="The path to the target output file"
+        ),
         {%- endif %}
     ).to_dict()
 

samples/sample_tap_gitlab/gitlab_tap.py

@@ -34,7 +34,7 @@ class SampleTapGitlab(Tap):
 
     name: str = "sample-tap-gitlab"
     config_jsonschema = PropertiesList(
-        Property("auth_token", StringType, required=True),
+        Property("auth_token", StringType, required=True, secret=True),
         Property("project_ids", ArrayType(StringType), required=True),
         Property("group_ids", ArrayType(StringType), required=True),
         Property("start_date", DateTimeType, required=True),

samples/sample_tap_google_analytics/ga_tap.py

@@ -24,7 +24,7 @@ class SampleTapGoogleAnalytics(Tap):
     config_jsonschema = PropertiesList(
         Property("view_id", StringType(), required=True),
         Property("client_email", StringType(), required=True),
-        Property("private_key", StringType(), required=True),
+        Property("private_key", StringType(), required=True, secret=True),
     ).to_dict()
 
     def discover_streams(self) -> List[SampleGoogleAnalyticsStream]:

singer_sdk/helpers/_meltano.py

@@ -0,0 +1,110 @@
+"""Helper functions for Meltano and MeltanoHub interop."""
+
+from __future__ import annotations
+
+from ._typing import (
+    is_array_type,
+    is_boolean_type,
+    is_datetime_type,
+    is_integer_type,
+    is_object_type,
+    is_secret_type,
+    is_string_type,
+)
+
+
+def _to_meltano_kind(jsonschema_type: dict) -> str | None:
+    """Returns a Meltano `kind` indicator for the provided JSON Schema type.
+
+    For reference:
+    https://docs.meltano.com/reference/plugin-definition-syntax#settingskind
+
+    Args:
+        jsonschema_type: JSON Schema type to check.
+
+    Returns:
+        A string representing the meltano 'kind'.
+    """
+    if is_secret_type(jsonschema_type):
+        return "password"
+
+    if is_string_type(jsonschema_type):
+        return "string"
+
+    if is_object_type(jsonschema_type):
+        return "object"
+
+    if is_array_type(jsonschema_type):
+        return "array"
+
+    if is_boolean_type(jsonschema_type):
+        return "boolean"
+
+    if is_datetime_type(jsonschema_type):
+        return "date_iso8601"
+
+    if is_integer_type(jsonschema_type):
+        return "integer"
+
+    return None
+
+
+def meltano_yaml_str(
+    plugin_name: str,
+    capabilities: list[str],
+    config_jsonschema: dict,
+) -> str:
+    """Returns a Meltano plugin definition as a yaml string.
+
+    Args:
+        plugin_name: Name of the plugin.
+        capabilities: List of capabilities.
+        config_jsonschema: JSON Schema of the expected config.
+
+    Returns:
+        A string representing the Meltano plugin Yaml definition.
+    """
+    capabilities_str: str = "\n".join(
+        [" - {capability}" for capability in capabilities]
+    )
+    settings_str: str = "\n".join(
+        [
+            f"""
+- name: {setting_name}
+  label: {setting_name.replace("_", " ").proper()}
+  kind: {_to_meltano_kind(type_dict["type"])},
+  description: {type_dict.get("description", 'null')}
+"""
+            for setting_name, type_dict in config_jsonschema["properties"].items()
+        ]
+    )
+    required_settings = [
+        setting_name
+        for setting_name, type_dict in config_jsonschema.items()
+        if setting_name in config_jsonschema.get("required", [])
+        or type_dict.get("required", False)
+    ]
+    settings_group_validation_str = " - - " + "\n  - ".join(required_settings)
+
+    return f"""
+name: {plugin_name}
+namespace: {plugin_name.replace('-', '_')}
+
+## The following could not be auto-detected:
+# maintenance_status:   #
+# repo:                 #
+# variant:              #
+# label:                #
+# description:          #
+# pip_url:              #
+# domain_url:           #
+# logo_url:             #
+# keywords: []          #
+
+capabilities:
+{capabilities_str}
+settings_group_validation:
+{settings_group_validation_str}
+settings:
+{settings_str}
+    """

singer_sdk/helpers/_typing.py

@@ -54,6 +54,40 @@ def append_type(type_dict: dict, new_type: str) -> dict:
     )
 
 
+def is_secret_type(type_dict: dict) -> bool:
+    """Return True if JSON Schema type definition appears to be a secret.
+
+    Will return true if either `writeOnly` or `sensitive` are true on this type
+    or any of the type's subproperties.
+
+    Args:
+        type_dict: The JSON Schema type to check.
+
+    Raises:
+        ValueError: If type_dict is None or empty.
+
+    Returns:
+        True if we detect any sensitive property nodes.
+    """
+    if not type_dict:
+        raise ValueError(
+            "Could not detect type from empty type_dict. "
+            "Did you forget to define a property in the stream schema?"
+        )
+
+    if type_dict.get("writeOnly") or type_dict.get("sensitive"):
+        return True
+
+    if "properties" in type_dict:
+        # Recursively check subproperties and return True if any child is secret.
+        return any(
+            is_secret_type(child_type_dict)
+            for child_type_dict in type_dict["properties"].values()
+        )
+
+    return False
+
+
 def is_object_type(property_schema: dict) -> Optional[bool]:
     """Return true if the JSON Schema type is an object or None if detection fails."""
     if "anyOf" not in property_schema and "type" not in property_schema:
@@ -152,6 +186,23 @@ def is_string_array_type(type_dict: dict) -> bool:
     return "array" in type_dict["type"] and bool(is_string_type(type_dict["items"]))
 
 
+def is_array_type(type_dict: dict) -> bool:
+    """Return True if JSON Schema type definition is a string array."""
+    if not type_dict:
+        raise ValueError(
+            "Could not detect type from empty type_dict. "
+            "Did you forget to define a property in the stream schema?"
+        )
+
+    if "anyOf" in type_dict:
+        return any([is_array_type(t) for t in type_dict["anyOf"]])
+
+    if "type" not in type_dict:
+        raise ValueError(f"Could not detect type from schema '{type_dict}'")
+
+    return "array" in type_dict["type"]
+
+
 def is_boolean_type(property_schema: dict) -> Optional[bool]:
     """Return true if the JSON Schema type is a boolean or None if detection fails."""
     if "anyOf" not in property_schema and "type" not in property_schema:
@@ -162,6 +213,16 @@ def is_boolean_type(property_schema: dict) -> Optional[bool]:
     return False
 
 
+def is_integer_type(property_schema: dict) -> Optional[bool]:
+    """Return true if the JSON Schema type is a boolean or None if detection fails."""
+    if "anyOf" not in property_schema and "type" not in property_schema:
+        return None  # Could not detect data type
+    for property_type in property_schema.get("anyOf", [property_schema.get("type")]):
+        if "integer" in property_type or property_type == "integer":
+            return True
+    return False
+
+
 def is_string_type(property_schema: dict) -> Optional[bool]:
     """Return true if the JSON Schema type is a boolean or None if detection fails."""
     if "anyOf" not in property_schema and "type" not in property_schema:

singer_sdk/plugin_base.py

@@ -28,6 +28,7 @@
 from singer_sdk.exceptions import ConfigValidationError
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers._compat import metadata
+from singer_sdk.helpers._meltano import meltano_yaml_str
 from singer_sdk.helpers._secrets import SecretString, is_common_secret_key
 from singer_sdk.helpers._util import read_json_file
 from singer_sdk.helpers.capabilities import (
@@ -341,64 +342,79 @@ def print_about(cls: Type["PluginBase"], format: Optional[str] = None) -> None:
 
         if format == "json":
             print(json.dumps(info, indent=2, default=str))
+            return
 
-        elif format == "markdown":
-            max_setting_len = cast(
-                int, max(len(k) for k in info["settings"]["properties"].keys())
-            )
+        if format == "markdown":
+            cls._print_about_markdown(info)
+            return
 
-            # Set table base for markdown
-            table_base = (
-                f"| {'Setting':{max_setting_len}}| Required | Default | Description |\n"
-                f"|:{'-' * max_setting_len}|:--------:|:-------:|:------------|\n"
-            )
+        if format == "meltano":
+            print(meltano_yaml_str(cls.name, cls.capabilities, cls.config_jsonschema))
+            return
 
-            # Empty list for string parts
-            md_list = []
-            # Get required settings for table
-            required_settings = info["settings"].get("required", [])
+        formatted = "\n".join([f"{k.title()}: {v}" for k, v in info.items()])
+        print(formatted)
 
-            # Iterate over Dict to set md
-            md_list.append(
-                f"# `{info['name']}`\n\n"
-                f"{info['description']}\n\n"
-                f"Built with the [Meltano Singer SDK](https://sdk.meltano.com).\n\n"
-            )
-            for key, value in info.items():
-
-                if key == "capabilities":
-                    capabilities = f"## {key.title()}\n\n"
-                    capabilities += "\n".join([f"* `{v}`" for v in value])
-                    capabilities += "\n\n"
-                    md_list.append(capabilities)
-
-                if key == "settings":
-                    setting = f"## {key.title()}\n\n"
-                    for k, v in info["settings"].get("properties", {}).items():
-                        md_description = v.get("description", "").replace("\n", "<BR/>")
-                        table_base += (
-                            f"| {k}{' ' * (max_setting_len - len(k))}"
-                            f"| {'True' if k in required_settings else 'False':8} | "
-                            f"{v.get('default', 'None'):7} | "
-                            f"{md_description:11} |\n"
-                        )
-                    setting += table_base
-                    setting += (
-                        "\n"
-                        + "\n".join(
-                            [
-                                "A full list of supported settings and capabilities "
-                                f"is available by running: `{info['name']} --about`"
-                            ]
-                        )
-                        + "\n"
+    @classmethod
+    def _print_about_markdown(cls: Type["PluginBase"], info: dict) -> None:
+        """Print about info as markdown.
+
+        Args:
+            info: The collected metadata for the class.
+        """
+        max_setting_len = cast(
+            int, max(len(k) for k in info["settings"]["properties"].keys())
+        )
+
+        # Set table base for markdown
+        table_base = (
+            f"| {'Setting':{max_setting_len}}| Required | Default | Description |\n"
+            f"|:{'-' * max_setting_len}|:--------:|:-------:|:------------|\n"
+        )
+
+        # Empty list for string parts
+        md_list = []
+        # Get required settings for table
+        required_settings = info["settings"].get("required", [])
+
+        # Iterate over Dict to set md
+        md_list.append(
+            f"# `{info['name']}`\n\n"
+            f"{info['description']}\n\n"
+            f"Built with the [Meltano Singer SDK](https://sdk.meltano.com).\n\n"
+        )
+        for key, value in info.items():
+
+            if key == "capabilities":
+                capabilities = f"## {key.title()}\n\n"
+                capabilities += "\n".join([f"* `{v}`" for v in value])
+                capabilities += "\n\n"
+                md_list.append(capabilities)
+
+            if key == "settings":
+                setting = f"## {key.title()}\n\n"
+                for k, v in info["settings"].get("properties", {}).items():
+                    md_description = v.get("description", "").replace("\n", "<BR/>")
+                    table_base += (
+                        f"| {k}{' ' * (max_setting_len - len(k))}"
+                        f"| {'True' if k in required_settings else 'False':8} | "
+                        f"{v.get('default', 'None'):7} | "
+                        f"{md_description:11} |\n"
+                    )
+                setting += table_base
+                setting += (
+                    "\n"
+                    + "\n".join(
+                        [
+                            "A full list of supported settings and capabilities "
+                            f"is available by running: `{info['name']} --about`"
+                        ]
                     )
-                    md_list.append(setting)
+                    + "\n"
+                )
+                md_list.append(setting)
 
-            print("".join(md_list))
-        else:
-            formatted = "\n".join([f"{k.title()}: {v}" for k, v in info.items()])
-            print(formatted)
+        print("".join(md_list))
 
     @classproperty
     def cli(cls) -> Callable: