init decorator and tests for checking server_url

Author: Coniferish

_test_unstructured_client/test_check_url_protocol.py

@@ -0,0 +1,71 @@
+import os
+import pytest
+
+from unstructured_client import UnstructuredClient
+
+
+def get_api_key():
+    api_key = os.getenv("UNS_API_KEY")
+    if api_key is None:
+        raise ValueError("""UNS_API_KEY environment variable not set. 
+Set it in your current shell session with `export UNS_API_KEY=<api_key>`""")
+    return api_key
+
+
+@pytest.mark.parametrize(
+    ("server_url"),
+    [
+        ("https://unstructured-000mock.api.unstructuredapp.io"), # correct url
+        ("unstructured-000mock.api.unstructuredapp.io"),
+        ("http://unstructured-000mock.api.unstructuredapp.io/general/v0/general"),
+        ("https://unstructured-000mock.api.unstructuredapp.io/general/v0/general"),
+        ("unstructured-000mock.api.unstructuredapp.io/general/v0/general"),
+    ]
+)
+def test_clean_server_url_on_paid_api_url(server_url: str):
+    client = UnstructuredClient(
+        server_url=server_url,
+        api_key_auth=get_api_key(),
+    )
+    assert client.general.sdk_configuration.server_url == "https://unstructured-000mock.api.unstructuredapp.io"
+
+
+@pytest.mark.parametrize(
+    ("server_url"),
+    [
+        ("http://localhost:8000"), # correct url
+        ("https://localhost:8000"),
+        ("localhost:8000"),
+        ("localhost:8000/general/v0/general"),
+        ("https://localhost:8000/general/v0/general"),
+    ]
+)
+def test_clean_server_url_on_localhost(server_url: str):
+    client = UnstructuredClient(
+        server_url=server_url,
+        api_key_auth=get_api_key(),
+    )
+    assert client.general.sdk_configuration.server_url == "http://localhost:8000"
+
+
+def test_clean_server_url_on_empty_string():
+    client = UnstructuredClient(
+        server_url="",
+        api_key_auth=get_api_key(),
+    )
+    assert client.general.sdk_configuration.server_url == ""
+
+@pytest.mark.parametrize(
+    ("server_url"),
+    [
+        ("https://unstructured-000mock.api.unstructuredapp.io"),
+        ("unstructured-000mock.api.unstructuredapp.io/general/v0/general"),
+    ]
+)
+def test_clean_server_url_with_positional_arguments(server_url: str):
+    client = UnstructuredClient(
+        get_api_key(),
+        "",
+        server_url,
+    )
+    assert client.general.sdk_configuration.server_url == "https://unstructured-000mock.api.unstructuredapp.io"

src/unstructured_client/sdk.py

@@ -6,13 +6,15 @@
 from typing import Callable, Dict, Union
 from unstructured_client import utils
 from unstructured_client.models import shared
+from unstructured_client.utils._decorators import clean_server_url
 
 class UnstructuredClient:
     r"""Unstructured Pipeline API: Partition documents with the Unstructured library"""
     general: General
 
     sdk_configuration: SDKConfiguration
 
+    @clean_server_url
     def __init__(self,
                  api_key_auth: Union[str, Callable[[], str]],
                  server: str = None,

src/unstructured_client/utils/_decorators.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import functools
+import inspect
+from typing import Any, Callable, Dict, Optional
+from typing_extensions import ParamSpec, Concatenate
+from urllib.parse import urlparse, urlunparse, ParseResult
+
+
+_P = ParamSpec("_P")
+
+
+def clean_server_url(func: Callable[_P, None]) -> Callable[_P, None]:
+
+    @functools.wraps(func)
+    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
+
+        def get_call_args_applying_defaults() -> Dict[str, Any]:
+            """Map both explicit and default arguments of decorated func call by param name."""
+            sig = inspect.signature(func)
+            call_args: Dict[str, Any] = dict(
+                **dict(zip(sig.parameters, args)), **kwargs
+            )
+            for param in sig.parameters.values():
+                if param.name not in call_args and param.default is not param.empty:
+                    call_args[param.name] = param.default
+            return call_args
+
+        call_args = get_call_args_applying_defaults()
+
+        server_url: Optional[str] = call_args.get("server_url")
+
+        if server_url:
+            # -- add a url scheme if not present (urllib.parse does not work reliably without it)
+            if "http" not in server_url:
+                server_url = "http://" + server_url
+
+            parsed_url: ParseResult = urlparse(server_url)
+
+            if "api.unstructuredapp.io" in server_url:
+                if parsed_url.scheme != "https":
+                    parsed_url = parsed_url._replace(scheme="https")
+            else:
+                # -- if not a paid api url, assume the api is hosted locally and the scheme is "http"
+                if parsed_url.scheme != "http":
+                    parsed_url = parsed_url._replace(scheme="http")
+
+            # -- path should always be empty
+            cleaned_url = parsed_url._replace(path="")
+            call_args["server_url"] = urlunparse(cleaned_url)
+
+        # -- call_args contains all args and kwargs. If users define some parameters using
+        # -- kwargs, param definitions would be duplicated. Pass only the `self`
+        # -- param as an arg and keep the rest in kwargs to prevent duplicates.
+        self_arg = (call_args.pop("self"),)
+        
+        return func(*self_arg, **call_args) # type: ignore
+
+    return wrapper