support before_request

Author: ponytailer

README.md

@@ -124,6 +124,27 @@ def get_user_post(self, user_id: int, post_id: int) -> PostResponse:
 - For GET and DELETE methods, remaining arguments are sent as query parameters
 - For POST, PUT, and PATCH methods, remaining arguments are sent in the request body as JSON
 
+```python
+
+# you can cal signature by your self, overwrite the function `before_request`
+from pydantic_client.schema import RequestInfo
+
+
+class MyAPIClient(RequestsWebClient):
+    # some code
+
+    def before_request(self, request_params: Dict[str, Any]) -> Dict[str, Any]:
+        # the request params before send: body, header, etc...
+        sign = cal_signature(request_params)
+        request_params["headers].update(dict(signature=sign))
+        return request_params
+
+
+# will send your new request_params
+user = client.get_user("123")
+
+```
+
 
 ### Timing Context Manager
 

pydantic_client/async_client.py

@@ -2,15 +2,14 @@
 
 from pydantic import BaseModel
 
-from .base import BaseWebClient
+from .base import BaseWebClient, RequestInfo
 
 try:
     import aiohttp
 except ImportError:
     raise ImportError("please install aiohttp: `pip install aiohttp`")
 
 
-
 T = TypeVar('T', bound=BaseModel)
 
 
@@ -25,32 +24,15 @@ def __init__(
     ):
         super().__init__(base_url, headers, timeout, session, statsd_address)
 
-    async def _request(
-        self,
-        method: str,
-        path: str,
-        *,
-        params: Optional[Dict[str, Any]] = None,
-        json: Optional[Dict[str, Any]] = None,
-        data: Optional[Dict[str, Any]] = None,
-        headers: Optional[Dict[str, str]] = None,
-        response_model: Optional[Type[T]] = None
-    ) -> Any:
-        url = self._make_url(path)
-        
-        # Merge headers
-        request_headers = self.headers.copy()
-        if headers:
-            request_headers.update(headers)
+    async def _request(self, request_info: RequestInfo) -> Any:
+        request_params = self.dump_request_params(request_info)
+        response_model = request_params.pop("response_model")
+
+        request_params = self.before_request(request_params)
 
         async with aiohttp.ClientSession() as session:
             async with session.request(
-                method=method,
-                url=url,
-                params=params,
-                json=json,
-                data=data,
-                headers=request_headers,
+                **request_params,
                 timeout=aiohttp.ClientTimeout(total=self.timeout)
             ) as response:
                 response.raise_for_status()
@@ -77,37 +59,17 @@ def __init__(
         except ImportError:
             raise ImportError("please install httpx: `pip install httpx`")
 
-    async def _request(
-        self,
-        method: str,
-        path: str,
-        *,
-        params: Optional[Dict[str, Any]] = None,
-        json: Optional[Dict[str, Any]] = None,
-        data: Optional[Dict[str, Any]] = None,
-        headers: Optional[Dict[str, str]] = None,
-        response_model: Optional[Type[T]] = None
-    ) -> Any:
+    async def _request(self, request_info: RequestInfo) -> Any:
         import httpx
-        url = self._make_url(path)
-        
-        # Merge headers
-        request_headers = self.headers.copy()
-        if headers:
-            request_headers.update(headers)
+        request_params = self.dump_request_params(request_info)
+        response_model = request_params.pop("response_model")
+
+        request_params = self.before_request(request_params)
 
         async with httpx.AsyncClient(timeout=self.timeout) as client:
-            response = await client.request(
-                method=method,
-                url=url,
-                params=params,
-                json=json,
-                data=data,
-                headers=request_headers
-            )
+            response = await client.request(**request_params)
             response.raise_for_status()
             data = response.json()
-
             if response_model is not None:
                 return response_model.model_validate(data)
             return data

pydantic_client/base.py

@@ -7,6 +7,7 @@
 from typing import Any, Dict, Optional, Type, TypeVar, List
 
 from pydantic import BaseModel
+from .schema import RequestInfo
 
 
 T = TypeVar('T', bound=BaseModel)
@@ -59,25 +60,32 @@ def from_config(cls, config: Dict[str, Any]) -> 'BaseWebClient':
             timeout=config.get('timeout', 30),
             session=config.get('session', None)
         )
+    
+    def before_request(self, request_params: Dict[str, Any]) -> Dict[str, Any]:
+        """before request, you can do something by yourself
+        such as: cal signature, etc."""
+        return request_params
 
     def _make_url(self, path: str) -> str:
         return f"{self.base_url}/{path.lstrip('/')}"
 
     def span(self, prefix: Optional[str] = None):
         return SpanContext(self, prefix)
+    
+    def dump_request_params(self, request_info: RequestInfo) -> Dict[str, Any]:
+        request_params = request_info.model_dump()
+        url = self._make_url(request_params.pop("path"))
+        # Merge headers
+        request_headers = self.headers.copy()
+        if request_info.headers:
+            request_headers.update(request_info.headers)
+        
+        request_params["headers"] = request_headers
+        request_params["url"] = url
+        return request_params
 
     @abstractmethod
-    def _request(
-        self,
-        method: str,
-        path: str,
-        *,
-        params: Optional[Dict[str, Any]] = None,
-        json: Optional[Dict[str, Any]] = None,
-        data: Optional[Dict[str, Any]] = None,
-        headers: Optional[Dict[str, str]] = None,
-        response_model: Optional[Type[T]] = None
-    ) -> Any:
+    def _request(self, request_info: RequestInfo) -> Any:
         ...
 
 

pydantic_client/decorators.py

@@ -5,11 +5,12 @@
 from pydantic import BaseModel
 
 from .tools.agno import register_agno_tool
+from .schema import RequestInfo
 
 
 def _process_request_params(
     func: Callable, method: str, path: str, form_body: bool, *args, **kwargs
-) -> Dict[str, Any]:
+) -> RequestInfo:
     """
     Extract and process request parameters from function arguments.
 
@@ -21,7 +22,7 @@ def _process_request_params(
         *args, **kwargs: Function arguments
 
     Returns:
-        Dictionary containing processed request parameters
+        RequestInfo containing processed request parameters
     """
     sig = inspect.signature(func)
     bound_args = sig.bind(*args, **kwargs)
@@ -61,7 +62,7 @@ def _process_request_params(
             if method in ["GET", "DELETE"]:
                 query_params[param_name] = param_value
 
-    return {
+    info = {
         "method": method,
         "path": formatted_path,
         "params": query_params if method in ["GET", "DELETE"] else None,
@@ -78,6 +79,7 @@ def _process_request_params(
         "headers": request_headers,
         "response_model": response_model,
     }
+    return RequestInfo.model_validate(info)
 
 
 def rest(
@@ -131,15 +133,15 @@ async def async_wrapped(self, *args, **kwargs):
                 request_params = _process_request_params(
                     func, method, path, form_body, self, *args, **kwargs
                 )
-                return await self._request(**request_params)
+                return await self._request(request_params)
 
             @wraps(func)
             def sync_wrapped(self, *args, **kwargs):
                 """Sync wrapper for handling HTTP requests."""
                 request_params = _process_request_params(
                     func, method, path, form_body, self, *args, **kwargs
                 )
-                return self._request(**request_params)
+                return self._request(request_params)
 
             @wraps(func)
             def choose_wrapper(self, *args, **kwargs):

pydantic_client/schema.py

@@ -0,0 +1,13 @@
+from typing import Dict, Optional, Any, Type
+
+from pydantic import BaseModel
+
+
+class RequestInfo(BaseModel):
+    method: str
+    path: str
+    params: Optional[Dict[str, Any]] = {}
+    json: Optional[Dict[str, Any]] = None
+    data: Optional[Dict[str, Any]] = None
+    headers: Optional[Dict[str, Any]] = {}
+    response_model: Optional[Type[BaseModel]] = None

pydantic_client/sync_client.py

@@ -3,7 +3,7 @@
 import requests
 from pydantic import BaseModel
 
-from .base import BaseWebClient
+from .base import BaseWebClient, RequestInfo
 
 T = TypeVar('T', bound=BaseModel)
 
@@ -21,35 +21,14 @@ def __init__(
         if not self.session:
             self.session = requests.Session()
 
-    def _request(
-        self,
-        method: str,
-        path: str,
-        *,
-        params: Optional[Dict[str, Any]] = None,
-        json: Optional[Dict[str, Any]] = None,
-        data: Optional[Dict[str, Any]] = None,
-        headers: Optional[Dict[str, str]] = None,
-        response_model: Optional[Type[T]] = None
-    ) -> Any:
-        url = self._make_url(path)
-        
-        # Merge headers
-        request_headers = self.headers.copy()
-        if headers:
-            request_headers.update(headers)
-
-        response = requests.request(
-            method=method,
-            url=url,
-            params=params,
-            json=json,
-            data=data,
-            headers=request_headers,
-            timeout=self.timeout
-        )
-        response.raise_for_status()
+    def _request(self, request_info: RequestInfo) -> Any:
+        request_params = self.dump_request_params(request_info)
+        response_model = request_params.pop("response_model")
 
+        request_params = self.before_request(request_params)
+
+        response = requests.request(**request_params, timeout=self.timeout)
+        response.raise_for_status()
         data = response.json()
         if response_model is not None:
             return response_model.model_validate(data)

tests/test_async_new_features.py

@@ -38,12 +38,12 @@ async def test_async_pydantic_model_json_body():
     # For this test, we'll use a simple mock by patching the _request method
 
     class MockAsyncClient(TestAsyncFormBodyClient):
-        async def _request(self, method, path, **kwargs):
+        async def _request(self, request_info):
             # Verify that the correct parameters are passed
-            assert method == "POST"
-            assert path == "/users"
-            assert kwargs.get('json') == {"name": "Test User", "email": "test@example.com"}
-            assert kwargs.get('data') is None  # Should be None for JSON requests
+            assert request_info.method == "POST"
+            assert request_info.path == "/users"
+            assert request_info.json == {"name": "Test User", "email": "test@example.com"}
+            assert request_info.data is None  # Should be None for JSON requests
 
             # Return mock response
             return {"id": "123", "name": "Test User", "email": "test@example.com"}
@@ -60,12 +60,12 @@ async def test_async_pydantic_model_form_body():
     """Test that async client correctly handles Pydantic models as form data"""
 
     class MockAsyncClient(TestAsyncFormBodyClient):
-        async def _request(self, method, path, **kwargs):
+        async def _request(self, request_info):
             # Verify that the correct parameters are passed
-            assert method == "POST"
-            assert path == "/users"
-            assert kwargs.get('data') == {"name": "Test User", "email": "test@example.com"}
-            assert kwargs.get('json') is None  # Should be None for form requests
+            assert request_info.method == "POST"
+            assert request_info.path == "/users"
+            assert request_info.data == {"name": "Test User", "email": "test@example.com"}
+            assert request_info.json is None  # Should be None for form requests
 
             # Return mock response
             return {"id": "123", "name": "Test User", "email": "test@example.com"}
@@ -82,11 +82,11 @@ async def test_async_custom_headers():
     """Test that async client correctly handles custom headers"""
 
     class MockAsyncClient(TestAsyncFormBodyClient):
-        async def _request(self, method, path, **kwargs):
+        async def _request(self, request_info):
             # Verify that the correct parameters are passed
-            assert method == "POST"
-            assert path == "/users/custom"
-            assert kwargs.get('headers') == {"X-Custom-Header": "custom-value", "Authorization": "Bearer token123"}
+            assert request_info.method == "POST"
+            assert request_info.path == "/users/custom"
+            assert request_info.headers == {"X-Custom-Header": "custom-value", "Authorization": "Bearer token123"}
 
             # Return mock response
             return {"id": "123", "name": "Test User", "email": "test@example.com"}