feat/Add new cli to generate schema as yaml

setup.py

@@ -32,8 +32,9 @@ def load_requirements(file: Union[str, Path]) -> List[str]:
     packages=find_packages(),
     entry_points={
         "console_scripts": [
-            "etl-uvicorn=unstructured_platform_plugins.etl_uvicorn.main:cmd",
-            "etl-validate=unstructured_platform_plugins.validate_api:validate_api",
+            "etl-uvicorn=unstructured_platform_plugins.cli.etl_uvicorn.main:cmd",
+            "etl-validate=unstructured_platform_plugins.cli.validate.main:validate_api",
+            "etl-schema=unstructured_platform_plugins.cli.schema_yaml.main:generate_yaml",
         ],
     },
     install_requires=load_requirements("requirements/cli.in")

test/expected_results/simple.yaml

@@ -0,0 +1,22 @@
+inputs:
+  properties:
+    content:
+      properties:
+        name:
+          type: string
+        value:
+          type: number
+      required:
+      - name
+      - value
+      type: object
+  required:
+  - content
+  type: object
+outputs:
+  properties:
+    response:
+      type: object
+  required:
+  - response
+  type: object

test/test_schema.py

@@ -6,7 +6,7 @@
 from pydantic import BaseModel
 
 import unstructured_platform_plugins.schema.json_schema as js
-from unstructured_platform_plugins.etl_uvicorn.utils import get_input_schema
+from unstructured_platform_plugins.cli.utils import get_input_schema
 from unstructured_platform_plugins.schema.model import is_validate_dict
 
 

test/test_utils.py

@@ -4,7 +4,7 @@
 from pydantic import BaseModel
 from uvicorn.importer import import_from_string
 
-from unstructured_platform_plugins.etl_uvicorn import utils
+from unstructured_platform_plugins.cli import utils
 
 
 def test_get_func_simple():

test/test_yaml.py

@@ -0,0 +1,14 @@
+from pathlib import Path
+
+from unstructured_platform_plugins.cli.schema_yaml.yaml_generator import generate_yaml
+
+test_path = Path(__file__).parent
+results_path = test_path / "expected_results"
+
+
+def test_generate_yaml_simple():
+    output = generate_yaml(app="test.assets.typed_dict_response:sample_function")
+    file_path = results_path / "simple.yaml"
+    with file_path.open("r") as f:
+        expected = f.read()
+    assert expected == output

unstructured_platform_plugins/etl_uvicorn/api_generator.py → unstructured_platform_plugins/cli/etl_uvicorn/api_generator.py

@@ -10,7 +10,7 @@
 from pydantic import BaseModel
 from uvicorn.importer import import_from_string
 
-from unstructured_platform_plugins.etl_uvicorn.utils import (
+from unstructured_platform_plugins.cli.utils import (
     get_func,
     get_input_schema,
     get_output_sig,

unstructured_platform_plugins/etl_uvicorn/main.py → unstructured_platform_plugins/cli/etl_uvicorn/main.py

@@ -5,7 +5,7 @@
 from uvicorn.config import LOGGING_CONFIG, Config, RawConfigParser
 from uvicorn.main import main, run
 
-from unstructured_platform_plugins.etl_uvicorn.api_generator import generate_fast_api
+from unstructured_platform_plugins.cli.etl_uvicorn.api_generator import generate_fast_api
 
 
 @dataclass

unstructured_platform_plugins/cli/schema_yaml/main.py

@@ -0,0 +1,20 @@
+from typing import Optional
+
+import click
+
+from unstructured_platform_plugins.cli.schema_yaml.yaml_generator import generate_yaml
+
+
+@click.command()
+@click.argument("app", envvar="UVICORN_APP")
+@click.option(
+    "--method-name",
+    required=False,
+    type=str,
+    default=None,
+    help="If passed in instance is a class, what method to wrap. "
+    "Will fall back to __call__ if none is provided.",
+)
+def yaml(app: str, method_name: Optional[str] = None) -> None:
+    output = generate_yaml(app=app, method_name=method_name)
+    print(output)

unstructured_platform_plugins/cli/schema_yaml/yaml_generator.py

@@ -0,0 +1,14 @@
+from typing import Optional
+
+from uvicorn.importer import import_from_string
+from yaml import Dumper, dump
+
+from unstructured_platform_plugins.cli.utils import get_func, get_schema_dict
+
+
+def generate_yaml(app: str, method_name: Optional[str] = None) -> str:
+    instance = import_from_string(app)
+    func = get_func(instance=instance, method_name=method_name)
+    schemas = get_schema_dict(func)
+    output = dump(schemas, Dumper=Dumper)
+    return output