gistfile1.txt

import requests
from typing import TypeVar, Generic, Type, Optional, Dict, Any, List
from urllib.parse import urljoin
from pydantic import BaseModel, ConfigDict

T = TypeVar('T', bound=BaseModel)


class APIClient(Generic[T]):
    """A generic API client wrapper using Pydantic for validation."""
    
    def __init__(self, base_url: str, timeout: int = 10):
        """Initialize the API client.
        
        Args:
            base_url: The base URL of the API
            timeout: Request timeout in seconds (default: 10)
        """
        self.base_url = base_url
        self.timeout = timeout
        self.session = requests.Session()
    
    def set_headers(self, headers: Dict[str, str]):
        """Set default headers for all requests.
        
        Args:
            headers: Dictionary of headers to set
        """
        self.session.headers.update(headers)
    
    def set_auth(self, auth: tuple):
        """Set authentication for requests.
        
        Args:
            auth: Tuple of (username, password) for basic auth
        """
        self.session.auth = auth
    
    def get(self, endpoint: str, model: Type[T], 
            params: Optional[Dict[str, Any]] = None) -> T:
        """Make a GET request and validate response with Pydantic model.
        
        Args:
            endpoint: The API endpoint (relative to base_url)
            model: Pydantic model to validate response against
            params: Query parameters
            
        Returns:
            Validated model instance
            
        Raises:
            requests.RequestException: If the request fails
            ValueError: If validation fails
        """
        url = urljoin(self.base_url, endpoint)
        response = self.session.get(url, timeout=self.timeout, params=params)
        response.raise_for_status()
        return model(**response.json())
    
    def get_list(self, endpoint: str, model: Type[T],
                 params: Optional[Dict[str, Any]] = None) -> List[T]:
        """Make a GET request expecting a list of items.
        
        Args:
            endpoint: The API endpoint
            model: Pydantic model for list items
            params: Query parameters
            
        Returns:
            List of validated model instances
        """
        url = urljoin(self.base_url, endpoint)
        response = self.session.get(url, timeout=self.timeout, params=params)
        response.raise_for_status()
        data = response.json()
        
        # Handle both list and dict responses
        if isinstance(data, list):
            return [model(**item) for item in data]
        elif isinstance(data, dict) and "results" in data:
            return [model(**item) for item in data["results"]]
        else:
            raise ValueError(f"Unexpected response format: {type(data)}")
    
    def post(self, endpoint: str, model: Type[T],
             json_data: Optional[Dict[str, Any]] = None,
             data: Optional[Dict[str, Any]] = None) -> T:
        """Make a POST request and validate response.
        
        Args:
            endpoint: The API endpoint
            model: Pydantic model to validate response
            json_data: JSON body data
            data: Form body data
            
        Returns:
            Validated model instance
        """
        url = urljoin(self.base_url, endpoint)
        response = self.session.post(url, timeout=self.timeout, 
                                     json=json_data, data=data)
        response.raise_for_status()
        return model(**response.json())
    
    def put(self, endpoint: str, model: Type[T],
            json_data: Optional[Dict[str, Any]] = None) -> T:
        """Make a PUT request and validate response.
        
        Args:
            endpoint: The API endpoint
            model: Pydantic model to validate response
            json_data: JSON body data
            
        Returns:
            Validated model instance
        """
        url = urljoin(self.base_url, endpoint)
        response = self.session.put(url, timeout=self.timeout, json=json_data)
        response.raise_for_status()
        return model(**response.json())
    
    def delete(self, endpoint: str, 
               model: Optional[Type[T]] = None) -> Optional[T]:
        """Make a DELETE request.
        
        Args:
            endpoint: The API endpoint
            model: Optional Pydantic model to validate response
            
        Returns:
            Validated model instance or None if no response
        """
        url = urljoin(self.base_url, endpoint)
        response = self.session.delete(url, timeout=self.timeout)
        response.raise_for_status()
        
        if response.content and model:
            return model(**response.json())
        return None
    
    def close(self):
        """Close the session."""
        self.session.close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.close()


# Example usage with JSONPlaceholder API
class Post(BaseModel):
    """Example Pydantic model."""
    userId: int
    id: int
    title: str
    body: str


class User(BaseModel):
    """Example Pydantic model."""
    id: int
    name: str
    email: str
    username: str


if __name__ == "__main__":
    # Example: Using with JSONPlaceholder API
    client = APIClient(base_url="https://jsonplaceholder.typicode.com")
    
    # GET single resource
    post = client.get("posts/1", Post)
    print(f"Post: {post.title}")
    
    # GET list of resources
    posts = client.get_list("posts", Post, params={"_limit": 5})
    print(f"First 5 posts: {len(posts)}")
    
    # GET user
    user = client.get("users/1", User)
    print(f"User: {user.name} ({user.email})")
    
    client.close()