Add postman import spec

.gitignore

@@ -12,9 +12,10 @@ wheels/
 # editor
 .vscode/
 .idea/
+pyrightconfig.json
 
 snapshot_report.html
 *.afdesign
 *.afdesign*
 *.afphoto
-*.afphoto*
+*.afphoto*

src/posting/__main__.py

@@ -9,6 +9,7 @@
 from posting.collection import Collection
 from posting.config import Settings
 from posting.importing.open_api import import_openapi_spec
+from posting.importing.postman import import_postman_spec
 from posting.locations import (
     config_file,
     default_collection_directory,
@@ -100,14 +101,29 @@ def locate(thing_to_locate: str) -> None:
     help="Path to save the imported collection",
     default=None,
 )
-def import_spec(spec_path: str, output: str | None) -> None:
+@click.option(
+    "--type", "-t", default="openapi", help="Specify spec type [openapi, postman]"
+)
+def import_spec(spec_path: str, output: str | None, type: str) -> None:
     """Import an OpenAPI specification into a Posting collection."""
     console = Console()
     console.print(
         "Importing is currently an experimental feature.", style="bold yellow"
     )
     try:
-        collection = import_openapi_spec(spec_path)
+        if type.lower() == "openapi":
+            spec_type = "OpenAPI"
+            collection = import_openapi_spec(spec_path)
+        elif type.lower() == "postman":
+            spec_type = "Postman"
+            console.print(
+                "Importing Postman collection haven't been implemented yet", style="red"
+            )
+            import_postman_spec(spec_path, output)
+            return
+        else:
+            console.print(f"Unknown spec type: {type!r}", style="red")
+            return
 
         if output:
             output_path = Path(output)
@@ -118,7 +134,9 @@ def import_spec(spec_path: str, output: str | None) -> None:
         collection.path = output_path
         collection.save_to_disk(output_path)
 
-        console.print(f"Successfully imported OpenAPI spec to {str(output_path)!r}")
+        console.print(
+            f"Successfully imported {spec_type!r} spec to {str(output_path)!r}"
+        )
     except Exception:
         console.print("An error occurred during the import process.", style="red")
         console.print_exception()

src/posting/collection.py

@@ -282,6 +282,7 @@ class APIInfo(BaseModel):
     termsOfService: HttpUrl | None = None
     contact: Contact | None = None
     license: License | None = None
+    specSchema: str | None = None
     version: str
 
 

src/posting/importing/postman.py

@@ -0,0 +1,180 @@
+from pathlib import Path
+from typing import Any, List, Optional
+import json
+import os
+import re
+import yaml
+
+from pydantic import BaseModel
+
+from rich.console import Console
+
+from posting.collection import APIInfo, Collection
+
+
+class Variable(BaseModel):
+    key: str
+    value: Optional[str] = None
+    src: Optional[str | List[str]] = None
+    fileNotInWorkingDirectoryWarning: Optional[str] = None
+    filesNotInWorkingDirectory: Optional[List[str]] = None
+    type: Optional[str] = None
+    disabled: Optional[bool] = None
+
+
+class Body(BaseModel):
+    mode: str
+    options: Optional[dict] = None
+    raw: Optional[str] = None
+    formdata: Optional[List[Variable]] = None
+
+
+class Url(BaseModel):
+    raw: str
+    host: Optional[List[str]] = None
+    path: Optional[List[str]] = None
+    query: Optional[List[Variable]] = None
+
+
+class PostmanRequest(BaseModel):
+    method: str
+    url: Optional[str | Url] = None
+    header: Optional[List[Variable]] = None
+    description: Optional[str] = None
+    body: Optional[Body] = None
+
+
+class RequestItem(BaseModel):
+    name: str
+    item: Optional[List["RequestItem"]] = None
+    request: Optional[PostmanRequest] = None
+
+
+class PostmanCollection(BaseModel):
+    info: dict[str, str]
+    item: List[RequestItem]
+    variable: List[Variable]
+
+
+def create_env_file(path: Path, env_filename: str, variables: List[Variable]) -> Path:
+    env_content: List[str] = []
+
+    for var in variables:
+        env_content.append(f"{transform_variables(var.key)}={var.value}")
+
+    env_file = path / env_filename
+    env_file.write_text("\n".join(env_content))
+    return env_file
+
+
+def generate_directory_structure(
+    items: List[RequestItem], current_path: str = "", base_path: Path = Path("")
+) -> List[str]:
+    directories = []
+    for item in items:
+        if item.item is not None:
+            folder_name = item.name
+            new_path = f"{current_path}/{folder_name}" if current_path else folder_name
+            full_path = Path(base_path) / new_path
+            os.makedirs(str(full_path), exist_ok=True)
+            directories.append(str(full_path))
+            generate_directory_structure(item.item, new_path, base_path)
+        if item.request is not None:
+            request_name = re.sub(r"[^A-Za-z0-9\.]+", "", item.name)
+            file_name = f"{request_name}.posting.yaml"
+            full_path = Path(base_path) / current_path / file_name
+            create_request_file(full_path, item)
+    return directories
+
+
+# Converts variable names like userId to $USER_ID, or user-id to $USER_ID
+def transform_variables(string):
+    underscore_case = re.sub(r"(?<!^)(?=[A-Z-])", "_", string).replace("-", "")
+    return underscore_case.upper()
+
+
+def transform_url(string):
+    def replace_match(match):
+        value = match.group(1)
+        return f"${transform_variables(value)}"
+
+    transformed = re.sub(r"\{\{([\w-]+)\}\}", replace_match, string)
+    return transformed
+
+
+def create_request_file(file_path: Path, request_data: RequestItem):
+    yaml_content: dict[str, Any] = {
+        "name": request_data.name,
+    }
+
+    if request_data.request is not None:
+        yaml_content["method"] = request_data.request.method
+
+        if request_data.request.header is not None:
+            yaml_content["headers"] = [
+                {"name": header.key, "value": header.value}
+                for header in request_data.request.header
+            ]
+
+        if request_data.request.url is not None:
+            if isinstance(request_data.request.url, Url):
+                yaml_content["url"] = transform_url(request_data.request.url.raw)
+                if request_data.request.url.query is not None:
+                    yaml_content["params"] = [
+                        {"name": param.key, "value": transform_url(param.value)}
+                        for param in request_data.request.url.query
+                    ]
+            else:
+                yaml_content["url"] = transform_url((request_data.request.url))
+
+        if request_data.request.description is not None:
+            yaml_content["description"] = request_data.request.description
+
+        if (
+            request_data.request.body is not None
+            and request_data.request.body.raw is not None
+        ):
+            yaml_content["body"] = {
+                "content": transform_url(request_data.request.body.raw)
+            }
+
+    # Write YAML file
+    with open(file_path, "w") as f:
+        yaml.dump(yaml_content, f, default_flow_style=False)
+
+
+def import_postman_spec(
+    spec_path: str | Path, output_path: str | Path | None
+) -> tuple[Collection, Path, List[str]]:
+    console = Console()
+    console.print(f"Importing Postman spec from {spec_path!r}.")
+
+    spec_path = Path(spec_path)
+    with open(spec_path, "r") as file:
+        spec_dict = json.load(file)
+
+    spec = PostmanCollection(**spec_dict)
+
+    info = APIInfo(
+        title=spec.info["name"],
+        description=spec.info.get("description", "No description"),
+        specSchema=spec.info["schema"],
+        version="2.0.0",
+    )
+
+    base_dir = spec_path.parent
+    if output_path is not None:
+        base_dir = Path(output_path) if isinstance(output_path, str) else output_path
+
+    console.print(f"Output path: {output_path!r}")
+
+    env_file = create_env_file(base_dir, f"{info.title}.env", spec.variable)
+    console.print(f"Created environment file {str(env_file)!r}.")
+
+    main_collection = Collection(path=spec_path.parent, name="Postman Test")
+
+    # Create the directory structure like Postman's request folders
+    directories = generate_directory_structure(spec.item, base_path=base_dir)
+    console.print("Finished importing postman collection.")
+
+    return main_collection, env_file, directories

tests/test_postman_import.py

@@ -0,0 +1,108 @@
+import pytest
+from pathlib import Path
+import json
+import tempfile
+import shutil
+from posting.importing.postman import (
+    Variable,
+    RequestItem,
+    PostmanRequest,
+    Url,
+    import_postman_spec,
+    create_env_file,
+    generate_directory_structure,
+    create_request_file,
+)
+
+
+@pytest.fixture
+def sample_postman_spec():
+    return {
+        "info": {
+            "name": "Test API",
+            "description": "A test API",
+            "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json",
+        },
+        "variable": [{"key": "base_url", "value": "https://api.example.com"}],
+        "item": [
+            {
+                "name": "Users",
+                "item": [
+                    {
+                        "name": "Get User",
+                        "request": {
+                            "method": "GET",
+                            "url": {
+                                "raw": "{{base_url}}/users/1",
+                                "query": [{"key": "include", "value": "posts"}],
+                            },
+                            "header": [{"key": "Accept", "value": "application/json"}],
+                        },
+                    }
+                ],
+            }
+        ],
+    }
+
+
+@pytest.fixture
+def temp_dir():
+    temp_dir = tempfile.mkdtemp()
+    yield Path(temp_dir)
+    shutil.rmtree(temp_dir)
+
+
+def test_import_postman_spec(sample_postman_spec, temp_dir):
+    spec_file = temp_dir / "test_spec.json"
+    with open(spec_file, "w") as f:
+        json.dump(sample_postman_spec, f)
+
+    collection, env_file, directories = import_postman_spec(spec_file, temp_dir)
+
+    assert collection.name == "Postman Test"
+    assert env_file.name == "Test API.env"
+    assert len(directories) == 1
+    assert directories[0].endswith("Users")
+
+
+def test_create_env_file(temp_dir):
+    variables = [Variable(key="base_url", value="https://api.example.com")]
+    env_file = create_env_file(temp_dir, "test.env", variables)
+
+    assert env_file.exists()
+    assert env_file.read_text() == "'base_url'='https://api.example.com'"
+
+
+def test_generate_directory_structure(temp_dir):
+    request_obj = PostmanRequest(method="GET", url=Url(raw="{{base_url}}/users/1"))
+    request_item = RequestItem(name="Get User", request=request_obj)
+    item = RequestItem(name="Users", item=[request_item])
+    items = [item]
+
+    directories = generate_directory_structure(items, base_path=temp_dir)
+
+    assert len(directories) == 1
+    assert directories[0].endswith("Users")
+    assert (temp_dir / "Users" / "Get User.posting.yaml").exists()
+
+
+def test_create_request_file(temp_dir):
+    request_obj = PostmanRequest(
+        method="GET",
+        url=Url(
+            raw="{{base_url}}/users/1", query=[Variable(key="include", value="posts")]
+        ),
+        header=[Variable(key="Accept", value="application/json")],
+    )
+    request_data = RequestItem(name="Get User", request=request_obj)
+
+    file_path = temp_dir / "test_request.posting.yaml"
+    create_request_file(file_path, request_data)
+
+    assert file_path.exists()
+    content = file_path.read_text()
+    assert "name: Get User" in content
+    assert "method: GET" in content
+    assert "url: '{{base_url}}/users/1'" in content
+    assert "headers:" in content
+    assert "params:" in content