feat: support nested response (#131)

* support nested response

* change annotation to English

Author: ponytailer

.gitignore

@@ -158,3 +158,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
+.vscode/

README.md

@@ -18,10 +18,10 @@ supporting both synchronous and asynchronous operations.
 - ⚡ **Async/Sync support**: Work with both synchronous and asynchronous HTTP operations
 - 🎯 **Decorator-based API**: Clean, intuitive API definition with decorators
 - 📝 **OpenAPI/Swagger support**: Generate client code from OpenAPI specifications
-- 🛡️ **Automatic validation**: Request/response validation with Pydantic models
+- 🛡️ **Mock API Responses**: This is useful for testing or development purposes.
 - ⚡ **Timing context manager**: Use `with client.span(prefix="myapi"):` to log timing for any API call, sync or async
-- 🔧 **Flexible configuration**: Easy client configuration with headers, timeouts, and more
 - 🔧 **convert api to llm tools**: API2Tools, support `agno`, others coming soon...
+- 🌟 **Nested Response Extraction**: Extract and parse deeply nested API responses using JSON path expressions
 
 ## TODO
 
@@ -44,6 +44,7 @@ See the [`example/`](./example/) directory for real-world usage of this library,
 - `example_httpx.py`: Async usage with HttpxWebClient
 - `example_aiohttp.py`: Async usage with AiohttpWebClient
 - `example_tools.py`: How to register and use Agno tools
+- `example_nested_response.py`: How to extract data from nested API responses
 
 
 ## Quick Start
@@ -169,6 +170,52 @@ user = client.get_user("123")
 
 ```
 
+## Handling Nested API Responses
+
+Many APIs return deeply nested JSON structures. Use the `response_extract_path` parameter to extract and parse specific data from complex API responses:
+
+```python
+from typing import List
+from pydantic import BaseModel
+from pydantic_client import RequestsWebClient, get
+
+class User(BaseModel):
+    id: str
+    name: str
+    email: str
+
+class MyClient(RequestsWebClient):
+    @get("/users/complex", response_extract_path="$.data.users")
+    def get_users_nested(self) -> List[User]:
+        """
+        Extracts the users list from a nested response structure
+        
+        Example response:
+        {
+            "status": "success",
+            "data": {
+                "users": [
+                    {"id": "1", "name": "Alice", "email": "alice@example.com"},
+                    {"id": "2", "name": "Bob", "email": "bob@example.com"}
+                ],
+                "total": 2
+            }
+        }
+        """
+        pass
+    
+    @get("/search", response_extract_path="$.results[0]")
+    def search_first_result(self) -> User:
+        """
+        Get just the first search result from an array
+        """
+        pass
+```
+
+The `response_extract_path` parameter defines where to find the data in the response. It supports:
+- Array indexing with square brackets: `$.results[0]` -> `User`
+- Optional `$` prefix for root object: `$.data.users` -> `list[User]`
+
 ## Mock API Responses
 
 You can configure the client to return mock responses instead of making actual API calls. This is useful for testing or development purposes.
@@ -178,10 +225,12 @@ You can configure the client to return mock responses instead of making actual A
 ```python
 from pydantic_client import RequestsWebClient, get
 
+
 class UserResponse(BaseModel):
     id: int
     name: str
 
+
 class MyClient(RequestsWebClient):
     @get("/users/{user_id}")
     def get_user(self, user_id: int) -> UserResponse:

example/example_nested_path.py

@@ -0,0 +1,124 @@
+from typing import List, Optional
+from pydantic import BaseModel
+from pydantic_client import RequestsWebClient, get
+
+
+class User(BaseModel):
+    id: str
+    name: str
+    email: Optional[str] = None
+
+
+class NestedPathClient(RequestsWebClient):
+    @get("/users/complex", response_extract_path="$.data.users")
+    def get_users_nested(self) -> List[User]:
+        """
+        Get users from a nested response structure where the users are in data.users
+        
+        Example response:
+        {
+            "status": "success",
+            "data": {
+                "users": [
+                    {"id": "1", "name": "Alice", "email": "alice@example.com"},
+                    {"id": "2", "name": "Bob", "email": "bob@example.com"}
+                ],
+                "total": 2
+            }
+        }
+        """
+        ...
+    
+    @get("/user/{user_id}", response_extract_path="data.user")
+    def get_user_by_id(self, user_id: str) -> User:
+        """
+        Get a single user from a nested path
+        
+        Example response:
+        {
+            "status": "success",
+            "data": {
+                "user": {"id": "1", "name": "Alice", "email": "alice@example.com"}
+            }
+        }
+        """
+        ...
+        
+    @get("/search", response_extract_path="$.results[0]")
+    def search_first_result(self) -> User:
+        """
+        Get just the first search result from an array
+        
+        Example response:
+        {
+            "query": "user search",
+            "results": [
+                {"id": "1", "name": "Alice", "email": "alice@example.com"},
+                {"id": "2", "name": "Bob", "email": "bob@example.com"}
+            ]
+        }
+        """
+        ...
+
+
+def main():
+    """
+    This is an example of how to use the response_extract_path parameter.
+    In a real application, you would make actual API calls.
+    """
+    # Create client
+    client = NestedPathClient(base_url="https://api.example.com")
+    
+    # Setup mock responses for demonstration
+    client.set_mock_config(mock_config=[
+        {
+            "name": "get_users_nested",
+            "output": {
+                "status": "success",
+                "data": {
+                    "users": [
+                        {"id": "1", "name": "Alice", "email": "alice@example.com"},
+                        {"id": "2", "name": "Bob", "email": "bob@example.com"}
+                    ],
+                    "total": 2
+                }
+            }
+        },
+        {
+            "name": "get_user_by_id",
+            "output": {
+                "status": "success",
+                "data": {
+                    "user": {"id": "1", "name": "Alice", "email": "alice@example.com"}
+                }
+            }
+        },
+        {
+            "name": "search_first_result",
+            "output": {
+                "query": "user search",
+                "results": [
+                    {"id": "1", "name": "Alice", "email": "alice@example.com"},
+                    {"id": "2", "name": "Bob", "email": "bob@example.com"}
+                ]
+            }
+        }
+    ])
+    
+    # Get all users from a nested structure
+    users = client.get_users_nested()
+    print(f"Got {len(users)} users:")
+    for user in users:
+        print(f"  - {user.name} ({user.email})")
+    
+    # Get a single user by ID
+    user = client.get_user_by_id("1")
+    print(f"\nGot user: {user.name} ({user.id})")
+    
+    # Get just the first search result
+    first_result = client.search_first_result()
+    print(f"\nFirst search result: {first_result.name}")
+
+
+if __name__ == "__main__":
+    main()

pydantic_client/__init__.py

@@ -12,5 +12,5 @@
     "post",
     "put",
     "patch",
-    "delete",
+    "delete"
 ]

pydantic_client/async_client.py

@@ -30,12 +30,13 @@ def __init__(
     async def _request(self, request_info: RequestInfo) -> Any:
         # Check if there's a mock response for this method
         mock_response = self._get_mock_response(request_info)
-        if mock_response:
+        if mock_response is not None:
             return mock_response
 
         import aiohttp
         request_params = self.dump_request_params(request_info)
         response_model = request_params.pop("response_model")
+        extract_path = request_params.pop("response_extract_path", None)  # Get response extraction path parameter
 
         request_params = self.before_request(request_params)
 
@@ -49,9 +50,16 @@ async def _request(self, request_info: RequestInfo) -> Any:
                 return await response.text()
             elif response_model is bytes:
                 return await response.content.read()
-            elif not response_model:
-                return await response.json()
-            return response_model.model_validate(await response.json(), from_attributes=True)
+            
+            json_data = await response.json()
+            if extract_path:
+                return self._extract_nested_data(json_data, extract_path, response_model)
+            elif not response_model or response_model is dict or getattr(response_model, '__module__', None) == 'inspect':
+                return json_data
+            elif hasattr(response_model, 'model_validate'):
+                return response_model.model_validate(json_data, from_attributes=True)
+            else:
+                return json_data
 
 
 class HttpxWebClient(BaseWebClient):
@@ -73,13 +81,14 @@ def __init__(
     async def _request(self, request_info: RequestInfo) -> Any:
         # Check if there's a mock response for this method
         mock_response = self._get_mock_response(request_info)
-        if mock_response:
+        if mock_response is not None:
             return mock_response
 
-        # 没有mock数据，继续进行正常请求
+        # No mock data, continue with the normal request
         import httpx
         request_params = self.dump_request_params(request_info)
         response_model = request_params.pop("response_model")
+        extract_path = request_params.pop("response_extract_path", None)  # Get response extraction path parameter
 
         request_params = self.before_request(request_params)
 
@@ -91,6 +100,13 @@ async def _request(self, request_info: RequestInfo) -> Any:
                 return response.text
             elif response_model is bytes:
                 return response.content
-            elif not response_model:
-                return response.json()
-            return response_model.model_validate(response.json(), from_attributes=True)
+                
+            json_data = response.json()
+            if extract_path:
+                return self._extract_nested_data(json_data, extract_path, response_model)
+            elif not response_model or response_model is dict or getattr(response_model, '__module__', None) == 'inspect':
+                return json_data
+            elif hasattr(response_model, 'model_validate'):
+                return response_model.model_validate(json_data, from_attributes=True)
+            else:
+                return json_data

pydantic_client/base.py

@@ -3,9 +3,10 @@
 import logging
 import time
 import statsd
+import re
 
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Optional, TypeVar, List
+from typing import Any, Dict, Optional, TypeVar, List, Type, get_origin, get_args
 
 from pydantic import BaseModel
 from .schema import RequestInfo
@@ -92,6 +93,7 @@ def dump_request_params(self, request_info: RequestInfo) -> Dict[str, Any]:
         request_params["headers"] = request_headers
         request_params["url"] = url
         request_params.pop("function_name", None)
+        # response_extract_path 会在 _request 方法中处理，所以保留
         return request_params
 
     def set_mock_config(
@@ -146,17 +148,23 @@ def set_mock_config(
     def _get_mock_response(self, request_info: RequestInfo) -> Optional[Any]:
         """Get mock response for a method if available in mock config"""
         if not self._mock_config:
-            return
+            return None
 
         name = request_info.function_name
 
         if name not in self._mock_config:
             logger.warning(f"No mock found for method: {name}")
-            return
+            return None
 
         mock_response = self._mock_config[name]
         response_model = request_info.response_model
-        if response_model and not isinstance(response_model, type) or response_model in (str, bytes):
+
+
+        extract_path = request_info.response_extract_path
+
+        if extract_path:
+            return self._extract_nested_data(mock_response, extract_path, response_model)
+        elif response_model and not isinstance(response_model, type) or response_model in (str, bytes):
             return mock_response
         elif response_model:
             return response_model.model_validate(mock_response, from_attributes=True)
@@ -165,7 +173,50 @@ def _get_mock_response(self, request_info: RequestInfo) -> Optional[Any]:
     @abstractmethod
     def _request(self, request_info: RequestInfo) -> Any:
         ...
-    
+        
+    def _extract_nested_data(self, data: Dict[str, Any], path: str, model_type: Type) -> Any:
+        """
+        Extract and parse data from nested response data
+        
+        Args:
+            data: JSON response data
+            path: JSON path expression, e.g. "$.data.user" or "$.data.items[0]"
+            model_type: Pydantic model type for parsing the extracted data
+        """
+        # Preprocess path, remove $ prefix
+        if path.startswith('$'):
+            path = path[2:].lstrip('.')
+            
+        # Extract all components from the path
+        path_components = re.split(r'\.|\[|\]', path)
+        path_components = [p for p in path_components if p]
+        
+        current = data
+        for component in path_components:
+            if component.isdigit():
+                idx = int(component)
+                if isinstance(current, list) and 0 <= idx < len(current):
+                    current = current[idx]
+                else:
+                    return None
+            elif isinstance(current, dict):
+                current = current.get(component)
+                if current is None:
+                    return None
+            else:
+                return None
+        
+        if current is not None:
+            # Handle list types
+            origin = get_origin(model_type)
+            
+            if origin is list or origin is List:
+                item_type = get_args(model_type)[0]
+                if isinstance(current, list):
+                    return [item_type.model_validate(item) for item in current]
+            elif hasattr(model_type, 'model_validate'):
+                return model_type.model_validate(current)
+        return current
 
     def register_agno_tools(self, agent):
         """