# Python

## ChainGPT AI News Generator SDK Documentation

The ChainGPT AI News Generator provides access to curated blockchain and cryptocurrency news with advanced filtering capabilities. You can retrieve the latest news articles, filter by categories, tokens, and search terms, and implement pagination for large datasets.

### Table of Contents

1. Installation
2. Quick Start
3. Initialization and Setup
4. NewsService API Reference
5. Data Models Reference
6. Usage Examples
7. Error Handling
8. Common Category and Token IDs
9. Best Practices

***

### Installation

Install the ChainGPT SDK via pip:

```bash
pip install chaingpt
```

Or add to your `requirements.txt`:

```txt
chaingpt>=1.1.3
```

For environment variable management (recommended):

```bash
pip install python-dotenv
```

***

### Quick Start

```python
import asyncio
import os
from chaingpt.client import ChainGPTClient

async def main():
    # Initialize client
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    # Get latest news
    news = await client.news.get_news(limit=5)
    
    # Display results
    for article in news.data:
        print(f"{article.title} - {article.createdAt}")
    
    await client.close()

if __name__ == "__main__":
    asyncio.run(main())
```

***

### Initialization and Setup

#### Environment Variables

Set up your API key as an environment variable for security:

```bash
export CHAINGPT_API_KEY="your_api_key_here"
```

Or use a `.env` file:

```env
CHAINGPT_API_KEY=your_api_key_here
```

#### Basic Client Setup

```python
import asyncio
import os
from chaingpt.client import ChainGPTClient
from chaingpt.exceptions import ChainGPTError, APIError, ValidationError
from dotenv import load_dotenv  # Optional: for .env file support

# Load environment variables
load_dotenv()

async def main():
    # Initialize the client
    api_key = os.getenv("CHAINGPT_API_KEY")
    if not api_key:
        raise ValueError("CHAINGPT_API_KEY environment variable is required")
    
    client = ChainGPTClient(api_key=api_key)
    
    try:
        # Access the news service
        news_service = client.news
        
        # Your code here
        
    finally:
        # Always close the client
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())
```

***

### NewsService API Reference

#### `get_news()` Method

Retrieves AI-generated news articles with comprehensive filtering and pagination capabilities.

**Method Signature**

```python
async def get_news(
    self,
    category_id: Optional[Union[int, List[int]]] = None,
    sub_category_id: Optional[Union[int, List[int]]] = None,
    token_id: Optional[Union[int, List[int]]] = None,
    search_query: Optional[str] = None,
    fetch_after: Optional[str] = None,
    limit: Optional[int] = 10,
    offset: Optional[int] = 0,
    sort_by: Optional[str] = "createdAt",
) -> GetNewsResponseModel
```

**Parameters**

| Parameter         | Type                              | Default       | Description                                                                              |
| ----------------- | --------------------------------- | ------------- | ---------------------------------------------------------------------------------------- |
| `category_id`     | `Optional[Union[int, List[int]]]` | `None`        | Filter by one or more category IDs. Examples: `8` (NFT), `[8, 12]` (multiple categories) |
| `sub_category_id` | `Optional[Union[int, List[int]]]` | `None`        | Filter by one or more sub-category IDs. Examples: `15` (Ethereum), `[15, 39]` (multiple) |
| `token_id`        | `Optional[Union[int, List[int]]]` | `None`        | Filter by specific token IDs. Examples: `79` (Bitcoin), `[79, 1027]` (multiple tokens)   |
| `search_query`    | `Optional[str]`                   | `None`        | Keyword search for title/description. Examples: `"halving"`, `"defi"`, `"nft"`           |
| `fetch_after`     | `Optional[str]`                   | `None`        | Date filter in `YYYY-MM-DD` format. Only returns articles published after this date      |
| `limit`           | `Optional[int]`                   | `10`          | Maximum articles to return (pagination). Range: 1-100                                    |
| `offset`          | `Optional[int]`                   | `0`           | Number of articles to skip (pagination)                                                  |
| `sort_by`         | `Optional[str]`                   | `"createdAt"` | Sort field. Currently supports `"createdAt"`                                             |

**Return Value**

Returns a `GetNewsResponseModel` object containing:

| Field        | Type                     | Description                        |
| ------------ | ------------------------ | ---------------------------------- |
| `statusCode` | `Optional[int]`          | HTTP status code (200 for success) |
| `message`    | `Optional[str]`          | Response message                   |
| `data`       | `List[NewsArticleModel]` | Array of news articles             |
| `limit`      | `Optional[int]`          | Applied limit parameter            |
| `offset`     | `Optional[int]`          | Applied offset parameter           |
| `total`      | `Optional[int]`          | Total number of articles available |

**Raises**

* `ValidationError`: Invalid parameters or malformed request
* `APIError`: Server-side errors or invalid responses
* `AuthenticationError`: Invalid API key
* `RateLimitError`: Too many requests
* `ChainGPTError`: Other SDK-related errors

***

### Data Models Reference

#### NewsArticleModel

Represents a single news article with complete metadata.

```python
class NewsArticleModel(BaseModel):
    # Basic article information
    id: int                           # Unique article identifier
    title: str                        # Article headline
    description: str                  # Article summary/description
    pubDate: str                      # Publication date (ISO format)
    author: str                       # Article author name
    imageUrl: str                     # Featured image URL
    createdAt: str                    # Creation timestamp
    updatedAt: str                    # Last update timestamp
    
    # Publishing information
    isPublished: bool                 # Publication status
    isFeatured: int                   # Featured article flag (0/1)
    isTopStory: int                   # Top story flag (0/1)
    viewsCount: int                   # Number of article views
    
    # Classification
    categoryId: Optional[int]         # Primary category ID
    subCategoryId: Optional[int]      # Sub-category ID
    tokenId: Optional[int]            # Related token/cryptocurrency ID
    
    # Relationships (populated objects)
    category: Optional[CategoryModel]     # Category details
    subCategory: Optional[SubCategoryModel]  # Sub-category details
    token: Optional[TokenModel]           # Token details
    media: Optional[MediaModel]           # Media/image details
    newsTags: List[Any]                   # Article tags
    
    # Internal fields
    userId: Optional[int]             # Author user ID
    mediaId: int                      # Media file ID
    deletedAt: Optional[str]          # Deletion timestamp (if soft-deleted)
```

#### CategoryModel

```python
class CategoryModel(BaseModel):
    id: int                    # Category identifier
    name: str                  # Category name (e.g., "NFT", "DeFi")
    isBlockchain: bool         # Blockchain-related category flag
    isToken: bool              # Token-specific category flag
    createdAt: str             # Creation timestamp
    updatedAt: str             # Last update timestamp
    deletedAt: Optional[str]   # Deletion timestamp (if applicable)
```

#### SubCategoryModel

```python
class SubCategoryModel(BaseModel):
    id: int                    # Sub-category identifier
    name: str                  # Sub-category name (e.g., "Ethereum", "Bitcoin")
    isBlockchain: bool         # Blockchain-related flag
    isToken: bool              # Token-specific flag
    createdAt: str             # Creation timestamp
    updatedAt: str             # Last update timestamp
    deletedAt: Optional[str]   # Deletion timestamp (if applicable)
```

#### TokenModel

```python
class TokenModel(BaseModel):
    id: int                    # Token identifier
    name: str                  # Token name (e.g., "Bitcoin", "Ethereum")
    isBlockchain: bool         # Blockchain-related flag
    isToken: bool              # Token classification flag
    createdAt: str             # Creation timestamp
    updatedAt: str             # Last update timestamp
    deletedAt: Optional[str]   # Deletion timestamp (if applicable)
```

#### MediaModel

```python
class MediaModel(BaseModel):
    id: int                         # Media file identifier
    fileName: str                   # Original filename
    fileDescriptor: str             # File description
    mimeType: str                   # MIME type (e.g., "image/jpeg")
    fileSize: int                   # File size in bytes
    createdAt: Optional[str]        # Creation timestamp
    updatedAt: Optional[str]        # Last update timestamp
    deletedAt: Optional[str]        # Deletion timestamp (if applicable)
```

#### GetNewsResponseModel

```python
class GetNewsResponseModel(BaseModel):
    statusCode: Optional[int]       # HTTP status code
    message: Optional[str]          # Response message
    data: List[NewsArticleModel]    # Array of news articles
    limit: Optional[int]            # Applied limit parameter
    offset: Optional[int]           # Applied offset parameter
    total: Optional[int]            # Total articles available
```

***

### Usage Examples

#### Basic News Retrieval

```python
async def get_latest_news():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    try:
        # Get latest 5 articles
        news = await client.news.get_news(limit=5)
        
        print(f"Found {news.total} total articles")
        print(f"Showing {len(news.data)} articles:")
        
        for article in news.data:
            print(f"\n📰 {article.title}")
            print(f"   👤 {article.author}")
            print(f"   📅 {article.createdAt}")
            print(f"   👀 {article.viewsCount} views")
            print(f"   📝 {article.description[:100]}...")
            
            # Show category information if available
            if article.category:
                print(f"   🏷️ Category: {article.category.name}")
            if article.subCategory:
                print(f"   🏷️ Sub-category: {article.subCategory.name}")
            if article.token:
                print(f"   🪙 Token: {article.token.name}")
                
    except ChainGPTError as e:
        print(f"Error: {e}")
    finally:
        await client.close()
```

#### Filtering by Category and Sub-Category

```python
async def get_nft_ethereum_news():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    try:
        # Get NFT news related to Ethereum
        news = await client.news.get_news(
            category_id=8,              # NFT category
            sub_category_id=[15, 39],   # Ethereum-related subcategories
            limit=10
        )
        
        print(f"Found {news.total} NFT/Ethereum articles:")
        
        for article in news.data:
            print(f"\n🎨 {article.title}")
            print(f"   📅 Published: {article.pubDate}")
            print(f"   🏷️ Category: {article.category.name if article.category else 'N/A'}")
            print(f"   🏷️ Sub-category: {article.subCategory.name if article.subCategory else 'N/A'}")
            
    except ChainGPTError as e:
        print(f"Error: {e}")
    finally:
        await client.close()
```

#### Search with Keywords

```python
async def search_bitcoin_halving():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    try:
        # Search for Bitcoin halving news
        news = await client.news.get_news(
            search_query="halving",
            token_id=79,  # Bitcoin token ID
            limit=5
        )
        
        print(f"Found {news.total} Bitcoin halving articles:")
        
        for article in news.data:
            print(f"\n₿ {article.title}")
            print(f"   📅 {article.createdAt}")
            print(f"   🪙 Token: {article.token.name if article.token else 'N/A'}")
            print(f"   📝 {article.description}")
            
    except ChainGPTError as e:
        print(f"Error: {e}")
    finally:
        await client.close()
```

#### Date Range Filtering

```python
async def get_recent_news():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    try:
        # Get news from the last month
        from datetime import datetime, timedelta
        
        # Calculate date 30 days ago
        thirty_days_ago = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
        
        news = await client.news.get_news(
            fetch_after=thirty_days_ago,
            limit=10,
            sort_by="createdAt"
        )
        
        print(f"Found {news.total} articles from the last 30 days:")
        
        for article in news.data:
            print(f"\n📰 {article.title}")
            print(f"   📅 {article.createdAt}")
            print(f"   👀 {article.viewsCount} views")
            
    except ChainGPTError as e:
        print(f"Error: {e}")
    finally:
        await client.close()
```

#### Pagination Implementation

```python
async def paginate_news():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    try:
        page_size = 5
        current_page = 0
        
        while True:
            offset = current_page * page_size
            
            news = await client.news.get_news(
                limit=page_size,
                offset=offset
            )
            
            if not news.data:
                print("No more articles to display")
                break
                
            print(f"\n{'='*50}")
            print(f"Page {current_page + 1}")
            print(f"Showing articles {offset + 1}-{offset + len(news.data)} of {news.total}")
            print(f"{'='*50}")
            
            for i, article in enumerate(news.data, 1):
                print(f"\n{offset + i}. {article.title}")
                print(f"    📅 {article.createdAt}")
                print(f"    👀 {article.viewsCount} views")
            
            # User interaction
            user_input = input(f"\nShow next page? (y/n/q to quit): ").lower()
            if user_input in ['n', 'q', 'quit']:
                break
                
            current_page += 1
            
    except ChainGPTError as e:
        print(f"Error: {e}")
    finally:
        await client.close()
```

#### Complex Multi-Filter Query

```python
async def complex_news_search():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    try:
        # Complex query: Multiple categories, tokens, and search terms
        news = await client.news.get_news(
            category_id=[8, 12],        # NFT and DeFi categories
            token_id=[79, 1027],        # Bitcoin and Ethereum
            search_query="market",      # Market-related news
            fetch_after="2024-01-01",   # This year only
            limit=15
        )
        
        print(f"Complex search results: {news.total} articles found")
        print("Filters applied:")
        print("  - Categories: NFT, DeFi")
        print("  - Tokens: Bitcoin, Ethereum")
        print("  - Search: 'market'")
        print("  - Date: After 2024-01-01")
        print(f"{'='*50}")
        
        # Group by category for better display
        by_category = {}
        for article in news.data:
            cat_name = article.category.name if article.category else "Uncategorized"
            if cat_name not in by_category:
                by_category[cat_name] = []
            by_category[cat_name].append(article)
        
        for category, articles in by_category.items():
            print(f"\n📂 {category} ({len(articles)} articles):")
            for article in articles:
                print(f"  • {article.title}")
                print(f"    🪙 {article.token.name if article.token else 'N/A'}")
                print(f"    👀 {article.viewsCount} views")
                
    except ChainGPTError as e:
        print(f"Error: {e}")
    finally:
        await client.close()
```

***

### Error Handling

#### Exception Types

The SDK provides specific exception types for different error scenarios:

```python
from chaingpt.exceptions import (
    ChainGPTError,          # Base exception
    APIError,               # API response errors
    ValidationError,        # Request validation errors
    AuthenticationError,    # Authentication failures
    RateLimitError,         # Rate limit exceeded
    InsufficientCreditsError,  # Insufficient account credits
    NotFoundError,          # Endpoint not found
    ServerError,            # Server-side errors
    TimeoutError,           # Request timeout
    ConfigurationError      # SDK configuration errors
)
```

#### Comprehensive Error Handling

```python
async def robust_news_fetching():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    try:
        news = await client.news.get_news(
            category_id=8,
            limit=10
        )
        
        # Check response status
        if news.statusCode != 200:
            print(f"API returned status {news.statusCode}: {news.message}")
            return
            
        # Process successful response
        for article in news.data:
            print(f"📰 {article.title}")
            
    except ValidationError as e:
        print(f"❌ Validation Error: {e.message}")
        if hasattr(e, 'field') and e.field:
            print(f"   Field: {e.field}")
        # Log details for debugging
        if e.details:
            print(f"   Details: {e.details}")
            
    except AuthenticationError as e:
        print(f"🔐 Authentication Error: {e.message}")
        print("   Please check your API key")
        
    except RateLimitError as e:
        print(f"⏰ Rate Limit Error: {e.message}")
        if hasattr(e, 'retry_after') and e.retry_after:
            print(f"   Retry after: {e.retry_after} seconds")
            
    except InsufficientCreditsError as e:
        print(f"💳 Insufficient Credits: {e.message}")
        print("   Please check your account balance")
        
    except APIError as e:
        print(f"🌐 API Error: {e.message}")
        if e.status_code:
            print(f"   Status Code: {e.status_code}")
        if e.details:
            print(f"   Details: {e.details}")
            
    except TimeoutError as e:
        print(f"⏱️ Timeout Error: {e.message}")
        print("   Try again or increase timeout settings")
        
    except ChainGPTError as e:
        print(f"🔧 SDK Error: {e.message}")
        if e.details:
            print(f"   Details: {e.details}")
            
    except Exception as e:
        print(f"💥 Unexpected Error: {str(e)}")
        print("   Please report this issue")
        
    finally:
        await client.close()
```

#### Error Handling with Retry Logic

```python
import asyncio
from typing import Optional

async def fetch_news_with_retry(
    client: ChainGPTClient,
    max_retries: int = 3,
    delay: float = 1.0
) -> Optional[GetNewsResponseModel]:
    """Fetch news with automatic retry on certain errors."""
    
    for attempt in range(max_retries):
        try:
            news = await client.news.get_news(limit=10)
            return news
            
        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            
            # Wait for the retry-after period if specified
            wait_time = e.retry_after if e.retry_after else delay * (2 ** attempt)
            print(f"Rate limited. Waiting {wait_time} seconds before retry {attempt + 1}/{max_retries}")
            await asyncio.sleep(wait_time)
            
        except (TimeoutError, ServerError) as e:
            if attempt == max_retries - 1:
                raise
                
            wait_time = delay * (2 ** attempt)  # Exponential backoff
            print(f"Temporary error: {e.message}. Retrying in {wait_time} seconds...")
            await asyncio.sleep(wait_time)
            
        except (AuthenticationError, ValidationError, InsufficientCreditsError):
            # Don't retry on these errors
            raise
            
    return None
```

***

### Common Category and Token IDs

#### Categories

| ID | Name | Description                     |
| -- | ---- | ------------------------------- |
| 8  | NFT  | Non-Fungible Token related news |
| 12 | DeFi | Decentralized Finance news      |

#### Sub-Categories (Ethereum-related)

| ID | Name     | Description                   |
| -- | -------- | ----------------------------- |
| 15 | Ethereum | Ethereum blockchain news      |
| 39 | Ethereum | Alternative Ethereum category |

#### Tokens

| ID   | Name     | Symbol | Description             |
| ---- | -------- | ------ | ----------------------- |
| 79   | Bitcoin  | BTC    | Bitcoin cryptocurrency  |
| 1027 | Ethereum | ETH    | Ethereum cryptocurrency |

**Note:** These IDs are examples and may change. Use the API responses to discover current category, sub-category, and token IDs. Consider caching this information for better performance.

#### Discovering Available IDs

```python
async def discover_categories():
    """Example function to discover available categories from API responses."""
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    
    try:
        # Fetch a large sample of news to discover categories
        news = await client.news.get_news(limit=100)
        
        categories = {}
        sub_categories = {}
        tokens = {}
        
        for article in news.data:
            # Collect category information
            if article.category:
                categories[article.category.id] = article.category.name
            
            # Collect sub-category information
            if article.subCategory:
                sub_categories[article.subCategory.id] = article.subCategory.name
                
            # Collect token information
            if article.token:
                tokens[article.token.id] = article.token.name
        
        print("Available Categories:")
        for cat_id, cat_name in sorted(categories.items()):
            print(f"  {cat_id}: {cat_name}")
            
        print("\nAvailable Sub-Categories:")
        for sub_id, sub_name in sorted(sub_categories.items()):
            print(f"  {sub_id}: {sub_name}")
            
        print("\nAvailable Tokens:")
        for token_id, token_name in sorted(tokens.items()):
            print(f"  {token_id}: {token_name}")
            
    except ChainGPTError as e:
        print(f"Error discovering categories: {e}")
    finally:
        await client.close()
```

***

### Best Practices

#### 1. Efficient API Usage

```python
# ✅ Good: Use specific filters to reduce response size
news = await client.news.get_news(
    category_id=8,      # Specific category
    limit=10,           # Reasonable limit
    fetch_after="2024-01-01"  # Recent news only
)

# ❌ Avoid: Fetching all news without filters
news = await client.news.get_news(limit=1000)  # Too large, may timeout
```

#### 2. Proper Pagination

```python
# ✅ Good: Implement proper pagination
async def get_all_news(client, category_id):
    all_articles = []
    offset = 0
    page_size = 50
    
    while True:
        news = await client.news.get_news(
            category_id=category_id,
            limit=page_size,
            offset=offset
        )
        
        if not news.data:
            break
            
        all_articles.extend(news.data)
        offset += page_size
        
        # Respect rate limits
        await asyncio.sleep(0.1)
    
    return all_articles
```

#### 3. Error Handling and Logging

```python
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def fetch_news_safely():
    try:
        news = await client.news.get_news(limit=10)
        logger.info(f"Successfully fetched {len(news.data)} articles")
        return news
        
    except ChainGPTError as e:
        logger.error(f"ChainGPT API error: {e.message}")
        if e.details:
            logger.debug(f"Error details: {e.details}")
        raise
```

#### 4. Resource Management

```python
# ✅ Good: Always close the client
async def process_news():
    client = ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY"))
    try:
        # Your news processing code
        news = await client.news.get_news()
        return news
    finally:
        await client.close()

# ✅ Better: Use async context manager if available
async def process_news_with_context():
    async with ChainGPTClient(api_key=os.getenv("CHAINGPT_API_KEY")) as client:
        news = await client.news.get_news()
        return news
```

#### 5. Caching for Performance

```python
import asyncio
from datetime import datetime, timedelta
from typing import Dict, Optional

class NewsCache:
    def __init__(self, ttl_minutes: int = 5):
        self.cache: Dict[str, tuple] = {}
        self.ttl = timedelta(minutes=ttl_minutes)
    
    def _make_key(self, **kwargs) -> str:
        return str(sorted(kwargs.items()))
    
    def get(self, **kwargs) -> Optional[GetNewsResponseModel]:
        key = self._make_key(**kwargs)
        if key in self.cache:
            data, timestamp = self.cache[key]
            if datetime.now() - timestamp < self.ttl:
                return data
            else:
                del self.cache[key]
        return None
    
    def set(self, data: GetNewsResponseModel, **kwargs):
        key = self._make_key(**kwargs)
        self.cache[key] = (data, datetime.now())

# Usage
cache = NewsCache(ttl_minutes=10)

async def get_cached_news(client, **kwargs):
    # Try cache first
    cached = cache.get(**kwargs)
    if cached:
        return cached
    
    # Fetch from API
    news = await client.news.get_news(**kwargs)
    cache.set(news, **kwargs)
    return news
```

#### 6. Rate Limiting

```python
import asyncio
from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, calls_per_minute: int = 60):
        self.calls_per_minute = calls_per_minute
        self.calls = []
    
    async def wait_if_needed(self):
        now = datetime.now()
        # Remove calls older than 1 minute
        self.calls = [call_time for call_time in self.calls 
                     if now - call_time < timedelta(minutes=1)]
        
        if len(self.calls) >= self.calls_per_minute:
            # Wait until the oldest call is more than 1 minute old
            sleep_time = 60 - (now - self.calls[0]).total_seconds()
            if sleep_time > 0:
                await asyncio.sleep(sleep_time)
        
        self.calls.append(now)

# Usage
rate_limiter = RateLimiter(calls_per_minute=30)

async def rate_limited_request(client, **kwargs):
    await rate_limiter.wait_if_needed()
    return await client.news.get_news(**kwargs)
```

***