Updating graphiant CLI invoke module

README.md

@@ -224,6 +224,8 @@ graphiant invoke v1_edges_summary_get --kwargs '{"enterprise_id": 123}'
   ```bash
   graphiant invoke v1_edges_summary_post --kwargs '{"v1_edges_summary_post_request": {"filter": {}}}'
   graphiant invoke v1_global_summary_post --kwargs '{"v1_global_summary_post_request": {"ntpType": true}}'
+  graphiant invoke v1_global_content_filters_get
+  graphiant invoke v1_global_domain_categories_get
   ```
 
 - **`graphiant rest`** — Query strings are a single **`--query`** / **`-q`** string: **`key=value`** pairs joined with **`&`**. Values are strings (URL-encode special characters in the shell if needed).

graphiant_cli/sdk_invoke.py

@@ -5,9 +5,10 @@
 import inspect
 import json
 import re
+import types
 from functools import lru_cache
 from pathlib import Path
-from typing import Annotated, Any, Optional, get_args, get_origin
+from typing import Annotated, Any, Optional, Union, get_args, get_origin
 
 from pydantic import BaseModel
 
@@ -71,10 +72,45 @@ def _unwrap_annotated(annotation: Any) -> Any:
     return annotation
 
 
+def _is_union_origin(origin: Any) -> bool:
+    if origin is Union:
+        return True
+    ut = getattr(types, "UnionType", None)
+    return ut is not None and origin is ut
+
+
 def _coerce_value(value: Any, annotation: Any) -> Any:
     if annotation is None or annotation is inspect.Parameter.empty:
         return value
     inner = _unwrap_annotated(annotation)
+    origin = get_origin(inner)
+    args = get_args(inner)
+
+    if _is_union_origin(origin):
+        non_none = [a for a in args if a is not type(None)]
+        if value is None:
+            return None
+        if len(non_none) == 1:
+            return _coerce_value(value, non_none[0])
+        return value
+
+    if origin is list:
+        if not isinstance(value, list) or len(args) != 1:
+            return value
+        elt_ann = _unwrap_annotated(args[0])
+        elt_origin = get_origin(elt_ann)
+        elt_args = get_args(elt_ann)
+        if _is_union_origin(elt_origin):
+            nn = [a for a in elt_args if a is not type(None)]
+            if len(nn) == 1:
+                elt_ann = nn[0]
+        if isinstance(elt_ann, type) and issubclass(elt_ann, BaseModel):
+            return [
+                elt_ann.model_validate(item) if isinstance(item, dict) else item
+                for item in value
+            ]
+        return value
+
     if isinstance(value, dict) and isinstance(inner, type) and issubclass(inner, BaseModel):
         return inner.model_validate(value)
     return value

pyproject.toml

@@ -44,6 +44,10 @@ filelock = ">=3.20.3"
 requires = ["setuptools"]
 build-backend = "setuptools.build_meta"
 
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
+
 [tool.pylint.'MESSAGES CONTROL']
 extension-pkg-whitelist = "pydantic"
 

tests/test_sdk_invoke.py

@@ -1,6 +1,11 @@
 """Tests for SDK method listing (no live API calls)."""
 
-from graphiant_cli.sdk_invoke import list_api_method_rows, list_api_methods
+from typing import Optional
+
+from graphiant_cli.sdk_invoke import _coerce_value, list_api_method_rows, list_api_methods
+from graphiant_sdk.models.v1_global_content_filters_post_request import (
+    V1GlobalContentFiltersPostRequest,
+)
 
 
 def test_list_api_methods_includes_auth_get() -> None:
@@ -18,3 +23,36 @@ def test_list_api_method_rows_includes_raw_http() -> None:
     rows = {name: (verb, path) for name, verb, path in list_api_method_rows("v1_edges_summary")}
     assert rows["v1_edges_summary_get"] == ("GET", "/v1/edges-summary")
     assert rows["v1_edges_summary_post"][0] == "POST"
+
+
+def test_list_api_methods_includes_global_content_filters() -> None:
+    names = list_api_methods("v1_global_content_filters")
+    assert "v1_global_content_filters_get" in names
+    assert "v1_global_content_filters_post" in names
+
+
+def test_list_api_method_rows_global_content_filters_and_rollouts() -> None:
+    rows = {n: (v, p) for n, v, p in list_api_method_rows("v1_global_content_filters")}
+    assert rows["v1_global_content_filters_get"] == ("GET", "/v1/global/content-filters")
+    assert rows["v1_global_content_filters_post"][0] == "POST"
+    roll = {n: (v, p) for n, v, p in list_api_method_rows("v1_software_rollouts")}
+    assert roll["v1_software_rollouts_get"] == ("GET", "/v1/software/rollouts")
+    macsec = {n: (v, p) for n, v, p in list_api_method_rows("v2_monitoring_macsec_device_id_status")}
+    assert macsec["v2_monitoring_macsec_device_id_status_get"] == (
+        "GET",
+        "/v2/monitoring/macsec/{deviceId}/status",
+    )
+
+
+def test_coerce_optional_request_model_from_dict() -> None:
+    """CLI --kwargs JSON dicts become Pydantic bodies (incl. Optional[Model] params)."""
+    ann = Optional[V1GlobalContentFiltersPostRequest]
+    out = _coerce_value({}, ann)
+    assert isinstance(out, V1GlobalContentFiltersPostRequest)
+    out2 = _coerce_value(None, ann)
+    assert out2 is None
+
+
+def test_coerce_request_model_non_optional() -> None:
+    out = _coerce_value({}, V1GlobalContentFiltersPostRequest)
+    assert isinstance(out, V1GlobalContentFiltersPostRequest)