Skip to content

Commit

Permalink
Feat: Adding method to generate an example config based on the connec…
Browse files Browse the repository at this point in the history
…tor spec. (#80)
  • Loading branch information
tinomerl authored Mar 22, 2024
1 parent 2d90641 commit c8ceafb
Show file tree
Hide file tree
Showing 4 changed files with 193 additions and 124 deletions.
48 changes: 48 additions & 0 deletions airbyte/sources/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,15 @@
import tempfile
import warnings
from contextlib import contextmanager, suppress
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast

import jsonschema
import pendulum
import yaml
from rich import print
from rich.syntax import Syntax
from typing_extensions import Literal

from airbyte_protocol.models import (
AirbyteCatalog,
Expand Down Expand Up @@ -252,6 +255,51 @@ def _get_spec(self, *, force_refresh: bool = False) -> ConnectorSpecification:
log_text=self._last_log_messages,
)

@property
def config_spec(self) -> dict[str, Any]:
"""Generate a configuration spec for this connector, as a JSON Schema definition.
This function generates a JSON Schema dictionary with configuration specs for the
current connector, as a dictionary.
Returns:
dict: The JSON Schema configuration spec as a dictionary.
"""
return self._get_spec(force_refresh=True).connectionSpecification

def print_config_spec(
self,
format: Literal["yaml", "json"] = "yaml", # noqa: A002
*,
output_file: Path | str | None = None,
) -> None:
"""Print the configuration spec for this connector.
Args:
- format: The format to print the spec in. Must be "yaml" or "json".
- output_file: Optional. If set, the spec will be written to the given file path. Otherwise,
it will be printed to the console.
"""
if format not in ["yaml", "json"]:
raise exc.AirbyteLibInputError(
message="Invalid format. Expected 'yaml' or 'json'",
input_value=format,
)
if isinstance(output_file, str):
output_file = Path(output_file)

if format == "yaml":
content = yaml.dump(self.config_spec, indent=2)
elif format == "json":
content = json.dumps(self.config_spec, indent=2)

if output_file:
output_file.write_text(content)
return

syntax_highlighted = Syntax(content, format)
print(syntax_highlighted)

@property
def _yaml_spec(self) -> str:
"""Get the spec as a yaml string.
Expand Down
Loading

0 comments on commit c8ceafb

Please sign in to comment.