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,25 @@ 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"
+            collection = import_postman_spec(spec_path, output)
+        else:
+            console.print(f"Unknown spec type: {type!r}", style="red")
+            return
 
         if output:
             output_path = Path(output)
@@ -118,7 +130,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,205 @@
+from pathlib import Path
+from typing import List, Optional
+import json
+import re
+
+from pydantic import BaseModel, Field
+
+from rich.console import Console
+
+from posting.collection import (
+    APIInfo,
+    Collection,
+    FormItem,
+    Header,
+    QueryParam,
+    RequestBody,
+    RequestModel,
+    HttpRequestMethod,
+)
+
+
+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 RawRequestOptions(BaseModel):
+    language: str
+
+
+class RequestOptions(BaseModel):
+    raw: RawRequestOptions
+
+
+class Body(BaseModel):
+    mode: str
+    options: Optional[RequestOptions] = 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: HttpRequestMethod
+    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] = Field(default_factory=dict)
+    variable: List[Variable] = Field(default_factory=list)
+
+    item: List[RequestItem]
+
+
+# Converts variable names like userId to $USER_ID, or user-id to $USER_ID
+def sanitize_variables(string):
+    underscore_case = re.sub(r"(?<!^)(?=[A-Z-])", "_", string).replace("-", "")
+    return underscore_case.upper()
+
+
+def sanitize_str(string):
+    def replace_match(match):
+        value = match.group(1)
+        return f"${sanitize_variables(value)}"
+
+    transformed = re.sub(r"\{\{([\w-]+)\}\}", replace_match, string)
+    return transformed
+
+
+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"{sanitize_variables(var.key)}={var.value}")
+
+    env_file = path / env_filename
+    env_file.write_text("\n".join(env_content))
+    return env_file
+
+
+def import_requests(
+    items: List[RequestItem], base_path: Path = Path("")
+) -> List[RequestModel]:
+    requests: List[RequestModel] = []
+    for item in items:
+        if item.item is not None:
+            requests = requests + import_requests(item.item, base_path)
+        if item.request is not None:
+            file_name = re.sub(r"[^A-Za-z0-9\.]+", "", item.name)
+            requests.append(format_request(file_name, item.request))
+
+    return requests
+
+
+def format_request(name: str, request: PostmanRequest) -> RequestModel:
+    postingRequest = RequestModel(
+        name=name,
+        method=request.method,
+        description=request.description if request.description is not None else "",
+        url=sanitize_str(
+            request.url.raw if isinstance(request.url, Url) else request.url
+        )
+        if request.url is not None
+        else "",
+    )
+
+    if request.header is not None:
+        for header in request.header:
+            postingRequest.headers.append(
+                Header(
+                    name=header.key,
+                    value=header.value if header.value is not None else "",
+                    enabled=True,
+                )
+            )
+
+    if (
+        request.url is not None
+        and isinstance(request.url, Url)
+        and request.url.query is not None
+    ):
+        for param in request.url.query:
+            postingRequest.params.append(
+                QueryParam(
+                    name=param.key,
+                    value=param.value if param.value is not None else "",
+                    enabled=param.disabled if param.disabled is not None else False,
+                )
+            )
+
+    if request.body is not None and request.body.raw is not None:
+        if (
+            request.body.mode == "raw"
+            and request.body.options is not None
+            and request.body.options.raw.language == "json"
+        ):
+            postingRequest.body = RequestBody(content=sanitize_str(request.body.raw))
+        elif request.body.mode == "formdata" and request.body.formdata is not None:
+            form_data: list[FormItem] = [
+                FormItem(
+                    name=data.key,
+                    value=data.value if data.value is not None else "",
+                    enabled=data.disabled is False,
+                )
+                for data in request.body.formdata
+            ]
+            postingRequest.body = RequestBody(form_data=form_data)
+
+    return postingRequest
+
+
+def import_postman_spec(
+    spec_path: str | Path, output_path: str | Path | None
+) -> Collection:
+    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: {str(base_dir)!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=info.title)
+    main_collection.readme = main_collection.generate_readme(info)
+
+    main_collection.requests = import_requests(spec.item, base_dir)
+
+    return main_collection

tests/test_postman_import.py

@@ -0,0 +1,66 @@
+import json
+from pathlib import Path
+from unittest.mock import patch, mock_open
+
+import pytest
+from posting.importing.postman import import_postman_spec, PostmanCollection
+from posting.collection import Collection
+
+
+@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": "baseUrl", "value": "https://api.example.com"}],
+        "item": [
+            {
+                "name": "Get Users",
+                "request": {
+                    "method": "GET",
+                    "url": {
+                        "raw": "{{baseUrl}}/users",
+                        "host": ["{{baseUrl}}"],
+                        "path": ["users"],
+                    },
+                },
+            }
+        ],
+    }
+
+
+@pytest.fixture
+def mock_spec_file(sample_postman_spec):
+    return mock_open(read_data=json.dumps(sample_postman_spec))
+
+
+def test_import_postman_spec(sample_postman_spec, mock_spec_file):
+    spec_path = Path("/path/to/spec.json")
+    output_path = Path("/path/to/output")
+
+    with patch("builtins.open", mock_spec_file), patch(
+        "posting.importing.postman.create_env_file"
+    ) as mock_create_env, patch(
+        "posting.collection.Collection.generate_readme"
+    ) as mock_generate_readme:
+        mock_create_env.return_value = output_path / "Test API.env"
+        mock_generate_readme.return_value = "# Test API"
+
+        result = import_postman_spec(spec_path, output_path)
+
+        assert isinstance(result, Collection)
+        assert result.name == "Test API"
+        assert len(result.requests) == 1
+        assert result.requests[0].name == "GetUsers"
+        assert result.requests[0].method == "GET"
+        assert result.requests[0].url == "$BASE_URL/users"
+
+        mock_create_env.assert_called_once_with(
+            output_path,
+            "Test API.env",
+            PostmanCollection(**sample_postman_spec).variable,
+        )
+        mock_generate_readme.assert_called_once()