Fix for duplicate "fields" requirements in query/payload parameters (Webhooks)

Author: unigrated-solutions

UPDATES.md

@@ -1,4 +1,31 @@
-## 7/19/25 Update:
+## 8/14/25 Update:
+   - **Added temporary fix for endpoints that have 'fields' requirements in both the Query Parameters and Payload Parameters**
+   - **To be able to take advantage of the field builder validation for "fields='all'", I added an optional response_fields argument**
+      - This will override any provided "fields" argument even if "fields" isn't a required payload paramater.
+   ```python
+   response = client.post.webhooks(response_fields='all', fields='id,name,primary_phone_number', events=["created","updated"], model="contact", url='https://your-webhook-server.com/events')
+   ```
+   **Returns:** 
+   ```bash
+      "data": {
+         "id": 1234567,
+         "etag": "\"longetagstring\"",
+         "url": "https://your-webhook-server.com/events",
+         "fields": "id,name,primary_phone_number",
+         "shared_secret": "shared-secret",
+         "model": "contact",
+         "status": "pending",
+         "events": [
+            "created"
+         ],
+         "expires_at": "2025-08-17T00:15:35-04:00",
+         "created_at": "2025-08-14T00:15:35-04:00",
+         "updated_at": "2025-08-14T00:15:35-04:00"
+      }
+   ```
+   **Using 'all' will not create a webhook subscription that provides 'all' fields in the actual webhook. For now they still need to be set manually**
+
+## 7/19/25:
    - **Removed model api as submodule**
    - **Clio API Model Generator has been published to pypi and is available via:**
       - pip install clio-api-model-generator

setup.py

@@ -10,7 +10,7 @@
 
 setup(
     name="clio-manage-api-client",
-    version="0.1.2",
+    version="0.1.3",
     author="Unigrated Partners",
     author_email="dev@unigratedpartners.com",
     description="Python Client for the Clio Manage API.",

src/clio_manage_python_client/classes/base.py

@@ -2,6 +2,7 @@
 from typing import Dict, Any, Optional, Type, List, Union, Tuple, get_origin, get_args, Literal
 from datetime import datetime
 import typing_inspect
+from dataclasses import fields, is_dataclass
 
 from ..configs import *
 from ..utils.validate_fields import validate_field_string, build_id_field_string
@@ -213,6 +214,7 @@ def process_model(model: Type, data: Dict[str, Any]) -> Tuple[Dict[str, Any], Di
             errors = {}
 
             for field_name, field_type in model.__annotations__.items():
+                print(field_name, field_type)
                 # Use the provided field name for validation; apply mappings after validation
                 original_field_name = field_name
                 mapped_field_name = MAPPINGS.get(field_name, field_name)
@@ -330,22 +332,50 @@ def _call_endpoint(self, metadata: dict, **kwargs):
         field_model = metadata.get("field_model")
         method = metadata["method"].upper()
 
+        use_field_model_for_fields = True  # default
+
+        # If request_body_model exists, check if it has a field named 'fields'
+        if request_body_model and "fields" in kwargs:
+            for f in fields(request_body_model):
+                # Check top-level
+                if f.name == "fields":
+                    use_field_model_for_fields = False
+                    break
+
+                # Check nested dataclass
+                if is_dataclass(f.type):
+                    nested_names = [sub_f.name for sub_f in fields(f.type)]
+                    if "fields" in nested_names:
+                        use_field_model_for_fields = False
+                        break
+
+        print("Use field model for 'fields'? ->", use_field_model_for_fields)
+                    
         path = self._format_path(path, kwargs)
 
         query_params = {}
         validation_errors = {"query": {}, "data": {}}
 
-        if field_model and "fields" in kwargs.keys():
+        if field_model and "fields" in kwargs and use_field_model_for_fields:
             field_string = kwargs["fields"]
             if field_string == "all_ids":
                 kwargs["fields"] = build_id_field_string(field_model)
             else:
                 response = validate_field_string(field_model, field_string)
                 kwargs["fields"] = response.get("valid_string")            
 
+        # Validate 'response_fields' if provided, but only if we are NOT using the field model
+        if "response_fields" in kwargs and not use_field_model_for_fields:
+            field_string = kwargs["response_fields"]
+            response = validate_field_string(field_model, field_string)
+            kwargs["response_fields"] = response.get("valid_string")
+            
         # Validate query parameters
         if query_model:
             for field, field_type in query_model.__annotations__.items():
+                if field == "fields" and not use_field_model_for_fields:
+                    continue
+        
                 mapped_field = MAPPINGS.get(field, field)
 
                 value = kwargs.pop(mapped_field, kwargs.pop(field, None))
@@ -369,7 +399,12 @@ def _call_endpoint(self, metadata: dict, **kwargs):
             payload = payload_result.get("payload")
             if payload_result.get("errors"):
                 validation_errors["data"].update(payload_result["errors"])
-
+                
+        # Reinject response_fields into query_params if present as "fields"
+        response_fields = kwargs.get("response_fields")
+        if response_fields:
+            query_params["fields"] = response_fields
+            
         # If there are validation errors, return them
         if validation_errors["query"] or validation_errors["data"]:
             return {
@@ -380,8 +415,9 @@ def _call_endpoint(self, metadata: dict, **kwargs):
                 "errors": validation_errors,
             }
 
-        url = BaseRequest.format_url(BaseRequest(self.base_url, self.base_path), path, query_params)
-        return self.request_handler(url, method, query_params, payload, **kwargs)
+        api_url = BaseRequest.format_url(BaseRequest(self.base_url, self.base_path), path, query_params)
+
+        return self.request_handler(api_url, method, query_params, payload, **kwargs)
 
     def __getattr__(self, method_type: str):
         """

src/clio_manage_python_client/client.py

@@ -73,22 +73,21 @@ def __getattr__(self, item: str):
         raise AttributeError(f"'Client' object has no attribute '{item}'")
 
     # Syncronous Requests
-    def _make_request(self, url: str, method: str, params: dict = None, payload: dict = None, return_all=False):
+    def _make_request(self, api_url: str, method: str, params: dict = None, payload: dict = None, return_all=False):
         """
         Makes the actual HTTP request based on the method.
         """
         headers = {
             "Authorization": f"Bearer {self.access_token}",
             "Content-Type": "application/json",
         }
-        print(params)
         params = params or {}
 
         try:
             # Make the request
             response = requests.request(
                 method=method.upper(),
-                url=url,
+                url=api_url,
                 headers=headers,
                 params=params,
                 json=payload,
@@ -99,24 +98,24 @@ def _make_request(self, url: str, method: str, params: dict = None, payload: dic
         except requests.exceptions.RequestException as e:
             raise RuntimeError(f"HTTP request failed: {e}") from e
 
-    def _request_handler(self, url: str, method: str, params: dict = None, payload: dict = None, return_all=False, **kwargs):
+    def _request_handler(self, api_url: str, method: str, params: dict = None, payload: dict = None, return_all=False, **kwargs):
         """
         Handles requests, including support for paginated responses when return_all=True.
         """
-        endpoint = url.split(self.base_url)[-1].split("?")[0]  # Extract endpoint from URL
+        endpoint = api_url.split(self.base_url)[-1].split("?")[0]  # Extract endpoint from URL
         params = params or {}
         # print(f'Return ALl: {return_all}')
         @self.rate_limiter(endpoint)
         def make_request():
             try:
                 if method == "DOWNLOAD":
-                    response_obj = self._download_content(url, params)
+                    response_obj = self._download_content(api_url, params)
                     self.rate_limiter.update_rate_limits(endpoint, response_obj.headers)
                     return response_obj
 
                 if return_all is False:
                     # Single request
-                    response_json, response_obj = self._make_request(url, method, params, payload)
+                    response_json, response_obj = self._make_request(api_url, method, params, payload)
                     self.rate_limiter.update_rate_limits(endpoint, response_obj.headers)
                     if self.response_handler:
                         self.response_handler.add_response(response_obj, kwargs.get('call_metadata'))
@@ -125,7 +124,7 @@ def make_request():
                 # Paginated request
                 all_results = []
                 next_page_token = None
-                current_url = url 
+                current_url = api_url 
 
                 while True: