Metadata-Version: 2.1
Name: divine-thegraph-token-api
Version: 0.1.1
Summary: Clean Python client for The Graph Token API with elegant EVM/SVM separation and comprehensive test coverage
Author-email: DIVINE <admin@divine.sh>
License: MIT
Project-URL: Homepage, https://github.com/codebydivine/requests
Project-URL: Issues, https://github.com/codebydivine/requests/issues
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Operating System :: OS Independent
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.13
Description-Content-Type: text/markdown
Requires-Dist: divine-requests>=0.1.1
Requires-Dist: divine-type-enforcer>=0.1.1
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: pytest-anyio>=0.21; extra == "dev"
Requires-Dist: python-dotenv>=1.0.0; extra == "dev"

# The Graph Token API Client

A comprehensive, type-safe Python client for The Graph Token API with elegant EVM/SVM separation. Built with `divine-requests` for reliable HTTP handling and `divine-type-enforcer` for robust runtime type validation.

## Features

- 🏗️ **Elegant Architecture**: Clean separation between EVM and SVM (Solana) blockchains
- ⚙️ **Network as Class Variable**: Pass network as constructor parameter, not environment variable
- 🔧 **Factory Pattern**: Easy client creation with `evm()` and `svm()` methods
- 🧹 **Clean API**: Intuitive method names without blockchain suffixes
- 🔄 **Auto Network Injection**: Network automatically injected into all API calls
- 📦 **Type Safety**: Full TypeScript-style type hints and runtime validation
- 🔒 **Async Context Managers**: Proper resource management with async/await
- 🚀 **High Performance**: Built on divine-requests for optimal HTTP performance

## Installation

```bash
pip install thegraph-client
```

## Quick Start

### Factory Pattern (Recommended)

```python
import anyio
from thegraph_client import TheGraphTokenAPI, NetworkId, SolanaNetworkId

async def main():
    # Initialize the main API client
    api = TheGraphTokenAPI(api_key="your_bearer_token")
    
    # Create EVM client for specific network
    async with api.evm(NetworkId.MAINNET) as mainnet:
        # Get NFT ownerships
        nfts = await mainnet.get_nft_ownerships(
            address="0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
        )
        print(f"Found {len(nfts.data)} NFTs")
        
        # Get token balances
        balances = await mainnet.get_balances(
            address="0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
        )
        print(f"Token balances: {len(balances.data)}")
    
    # Create SVM client for Solana
    async with api.svm(SolanaNetworkId.SOLANA) as solana:
        # Get SPL token balances
        balances = await solana.get_balances(
            token_account="4ct7br2vTPzfdmY3S5HLtTxcGSBfn6pnw98hsS6v359A"
        )
        print(f"SPL balances: {len(balances.data)}")

anyio.run(main)
```

### Direct Instantiation

```python
import anyio
from thegraph_client import EVMTokenAPI, SVMTokenAPI, NetworkId

async def main():
    # Direct EVM client
    async with EVMTokenAPI(
        network=NetworkId.BASE,
        api_key="your_bearer_token"
    ) as base_api:
        transfers = await base_api.get_transfers()
        print(f"Base transfers: {len(transfers.data)}")
    
    # Direct SVM client
    async with SVMTokenAPI(api_key="your_bearer_token") as svm_api:
        swaps = await svm_api.get_swaps(program_id="RAYDIUM")
        print(f"Raydium swaps: {len(swaps.data)}")

anyio.run(main)
```

### Class Methods

```python
import anyio
from thegraph_client import TheGraphTokenAPI, NetworkId

async def main():
    # Create EVM client directly
    async with TheGraphTokenAPI.create_evm_client(
        network=NetworkId.ARBITRUM_ONE,
        api_key="your_bearer_token"
    ) as arbitrum:
        pools = await arbitrum.get_pools(protocol="uniswap_v3")
        print(f"Uniswap V3 pools: {len(pools.data)}")

anyio.run(main)
```

## API Reference

### TheGraphTokenAPI (Main Client)

The main entry point that provides factory methods for creating specialized clients.

#### Factory Methods

- `evm(network: Union[NetworkId, str]) -> EVMTokenAPI` - Create EVM client
- `svm(network: Union[SolanaNetworkId, str] = "solana") -> SVMTokenAPI` - Create SVM client

#### Class Methods

- `create_evm_client(network, api_key=None, base_url=None) -> EVMTokenAPI`
- `create_svm_client(network="solana", api_key=None, base_url=None) -> SVMTokenAPI`

#### Monitoring Methods

- `get_health() -> str` - API health check
- `get_version() -> VersionResponse` - API version info
- `get_networks() -> NetworksResponse` - Available networks

---

## EVMTokenAPI

Client for EVM-compatible blockchains (Ethereum, Polygon, Base, Arbitrum, etc.).

### Supported Networks

```python
from thegraph_client import NetworkId

# Available networks
NetworkId.MAINNET          # Ethereum Mainnet
NetworkId.POLYGON          # Polygon
NetworkId.BASE             # Base
NetworkId.ARBITRUM_ONE     # Arbitrum One
NetworkId.OPTIMISM         # Optimism
NetworkId.AVALANCHE        # Avalanche C-Chain
NetworkId.BSC              # BNB Smart Chain
NetworkId.FANTOM           # Fantom
NetworkId.CRONOS           # Cronos
NetworkId.GNOSIS           # Gnosis Chain
NetworkId.CELO             # Celo
NetworkId.MOONBEAM         # Moonbeam
```

### NFT Methods

#### `get_nft_ownerships(address, token_standard=None, limit=10, page=1)`
Get NFT ownerships for an EVM address.

**Parameters:**
- `address` (str): EVM address to query
- `token_standard` (TokenStandard, optional): Filter by ERC721 or ERC1155
- `limit` (int): Maximum results (1-1000, default 10)
- `page` (int): Page number (default 1)

**Returns:** `NFTOwnershipsResponse`

```python
async with api.evm(NetworkId.MAINNET) as mainnet:
    nfts = await mainnet.get_nft_ownerships(
        address="0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
        token_standard="ERC721",
        limit=50
    )
    for nft in nfts.data:
        print(f"{nft['name']} #{nft['token_id']}")
```

#### `get_nft_collection(contract)`
Get NFT collection metadata by contract address.

**Parameters:**
- `contract` (str): NFT contract address

**Returns:** `NFTCollectionsResponse`

```python
collection = await mainnet.get_nft_collection("0xBC4CA0EdA7647A8aB7C2061c2E118A18a936f13D")
print(f"Collection: {collection.data[0]['name']}")
```

#### `get_nft_item(contract, token_id)`
Get specific NFT item metadata.

**Parameters:**
- `contract` (str): NFT contract address
- `token_id` (str): NFT token ID

**Returns:** `NFTItemsResponse`

```python
nft = await mainnet.get_nft_item(
    contract="0xBC4CA0EdA7647A8aB7C2061c2E118A18a936f13D",
    token_id="1"
)
print(f"NFT: {nft.data[0]['name']}")
```

#### `get_nft_activities(contract, any_address=None, from_address=None, to_address=None, start_time=None, end_time=None, order_by="timestamp", order_direction="desc", limit=10, page=1)`
Get NFT activities (transfers, mints, burns).

**Parameters:**
- `contract` (str): NFT contract address (required)
- `any_address` (str, optional): Filter by any address (from or to)
- `from_address` (str, optional): Filter by from address
- `to_address` (str, optional): Filter by to address
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `order_by` (OrderBy): Field to order by
- `order_direction` (OrderDirection): Order direction
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `NFTActivitiesResponse`

```python
activities = await mainnet.get_nft_activities(
    contract="0xBC4CA0EdA7647A8aB7C2061c2E118A18a936f13D",
    any_address="0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
    limit=20
)
```

#### `get_nft_holders(contract)`
Get NFT holders for a contract.

**Parameters:**
- `contract` (str): NFT contract address

**Returns:** `NFTHoldersResponse`

```python
holders = await mainnet.get_nft_holders("0xBC4CA0EdA7647A8aB7C2061c2E118A18a936f13D")
print(f"Total holders: {len(holders.data)}")
```

#### `get_nft_sales(contract, token_id=None, any_address=None, offerer=None, recipient=None, start_time=None, end_time=None, order_by="timestamp", order_direction="desc", limit=10, page=1)`
Get NFT marketplace sales.

**Parameters:**
- `contract` (str): NFT contract address (required)
- `token_id` (str, optional): Filter by specific token ID
- `any_address` (str, optional): Filter by any address
- `offerer` (str, optional): Filter by offerer address
- `recipient` (str, optional): Filter by recipient address
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `order_by` (OrderBy): Field to order by
- `order_direction` (OrderDirection): Order direction
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `NFTSalesResponse`

### Token & Balance Methods

#### `get_balances(address, contract=None, limit=10, page=1)`
Get ERC-20 and native token balances.

**Parameters:**
- `address` (str): EVM address to query
- `contract` (str, optional): Filter by specific contract
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `BalancesResponse`

```python
balances = await mainnet.get_balances(
    address="0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
    limit=100
)
for balance in balances.data:
    print(f"{balance['symbol']}: {balance['value']}")
```

#### `get_token(contract)`
Get ERC-20 token contract metadata.

**Parameters:**
- `contract` (str): Token contract address

**Returns:** `TokensResponse`

```python
token = await mainnet.get_token("0xA0b86a33E6441Ccf00F4F1B4371a8e4b7B34b4E3")
print(f"Token: {token.data[0]['name']} ({token.data[0]['symbol']})")
```

#### `get_token_holders(contract, order_by="value", order_direction="desc", limit=10, page=1)`
Get ERC-20 token holder balances.

**Parameters:**
- `contract` (str): Token contract address
- `order_by` (OrderBy): Field to order by
- `order_direction` (OrderDirection): Order direction
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `TokenHoldersResponse`

```python
holders = await mainnet.get_token_holders(
    contract="0xA0b86a33E6441Ccf00F4F1B4371a8e4b7B34b4E3",
    limit=50
)
```

### Transfer Methods

#### `get_transfers(from_address=None, to_address=None, contract=None, transaction_id=None, start_time=None, end_time=None, order_by="timestamp", order_direction="desc", limit=10, page=1)`
Get ERC-20 and native token transfer events.

**Parameters:**
- `from_address` (str, optional): Filter by from address
- `to_address` (str, optional): Filter by to address
- `contract` (str, optional): Filter by contract address
- `transaction_id` (str, optional): Filter by transaction hash
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `order_by` (OrderBy): Field to order by
- `order_direction` (OrderDirection): Order direction
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `TransfersResponse`

```python
transfers = await mainnet.get_transfers(
    from_address="0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
    limit=100
)
```

### DEX & Trading Methods

#### `get_swaps(pool=None, caller=None, sender=None, recipient=None, protocol=None, transaction_id=None, start_time=None, end_time=None, order_by="timestamp", order_direction="desc", limit=10, page=1)`
Get EVM DEX swap transactions.

**Parameters:**
- `pool` (str, optional): Filter by pool address
- `caller` (str, optional): Filter by caller address
- `sender` (str, optional): Filter by sender address
- `recipient` (str, optional): Filter by recipient address
- `protocol` (Protocol, optional): Filter by DEX protocol (uniswap_v2, uniswap_v3)
- `transaction_id` (str, optional): Filter by transaction hash
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `order_by` (OrderBy): Field to order by
- `order_direction` (OrderDirection): Order direction
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `SwapsResponse`

```python
swaps = await mainnet.get_swaps(
    protocol="uniswap_v3",
    limit=50
)
```

#### `get_pools(pool=None, factory=None, token=None, symbol=None, protocol=None, limit=10, page=1)`
Get EVM DEX liquidity pools.

**Parameters:**
- `pool` (str, optional): Filter by pool address
- `factory` (str, optional): Filter by factory address
- `token` (str, optional): Filter by token address
- `symbol` (str, optional): Filter by symbol
- `protocol` (Protocol, optional): Filter by DEX protocol
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `PoolsResponse`

```python
pools = await mainnet.get_pools(
    token="0xA0b86a33E6441Ccf00F4F1B4371a8e4b7B34b4E3",
    protocol="uniswap_v3"
)
```

### OHLC & Price Data

#### `get_ohlc_pools(pool, interval="1h", start_time=None, end_time=None, limit=10, page=1)`
Get OHLC data for EVM DEX pools.

**Parameters:**
- `pool` (str): Pool address (required)
- `interval` (Interval): Time interval (1h, 4h, 1d, 1w)
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `OHLCResponse`

```python
ohlc = await mainnet.get_ohlc_pools(
    pool="0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640",
    interval="4h",
    limit=100
)
```

#### `get_ohlc_prices(token, interval="1h", start_time=None, end_time=None, limit=10, page=1)`
Get OHLC price data for EVM tokens.

**Parameters:**
- `token` (str): Token address (required)
- `interval` (Interval): Time interval (1h, 4h, 1d, 1w)
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `OHLCResponse`

```python
prices = await mainnet.get_ohlc_prices(
    token="0xA0b86a33E6441Ccf00F4F1B4371a8e4b7B34b4E3",
    interval="1d"
)
```

### Historical Data

#### `get_historical_balances(address, contracts=None, interval="1h", start_time=None, end_time=None, limit=10, page=1)`
Get historical balance data for EVM addresses.

**Parameters:**
- `address` (str): EVM address to query (required)
- `contracts` (List[str], optional): List of contract addresses to filter by
- `interval` (Interval): Time interval (1h, 4h, 1d, 1w)
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `HistoricalBalancesResponse`

```python
historical = await mainnet.get_historical_balances(
    address="0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
    contracts=["0xA0b86a33E6441Ccf00F4F1B4371a8e4b7B34b4E3"],
    interval="1d"
)
```

---

## SVMTokenAPI

Client for Solana Virtual Machine (SVM) operations.

### Supported Networks

```python
from thegraph_client import SolanaNetworkId

# Available networks
SolanaNetworkId.SOLANA  # Solana Mainnet
```

### Balance Methods

#### `get_balances(token_account=None, mint=None, program_id=None, limit=10, page=1)`
Get Solana SPL token balances.

**Parameters:**
- `token_account` (str, optional): Filter by token account address
- `mint` (str, optional): Filter by mint address
- `program_id` (SolanaPrograms, optional): Filter by program ID
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `SolanaBalancesResponse`

```python
async with api.svm() as solana:
    balances = await solana.get_balances(
        token_account="4ct7br2vTPzfdmY3S5HLtTxcGSBfn6pnw98hsS6v359A",
        limit=50
    )
    for balance in balances.data:
        print(f"Mint: {balance['mint']}, Amount: {balance['amount']}")
```

### Transfer Methods

#### `get_transfers(signature=None, program_id=None, mint=None, authority=None, source=None, destination=None, start_time=None, end_time=None, order_by="timestamp", order_direction="desc", limit=10, page=1)`
Get Solana SPL token transfer events.

**Parameters:**
- `signature` (str, optional): Filter by transaction signature
- `program_id` (SolanaPrograms, optional): Filter by program ID
- `mint` (str, optional): Filter by mint address
- `authority` (str, optional): Filter by authority address
- `source` (str, optional): Filter by source address
- `destination` (str, optional): Filter by destination address
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `order_by` (OrderBy): Field to order by
- `order_direction` (OrderDirection): Order direction
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `SolanaTransfersResponse`

```python
transfers = await solana.get_transfers(
    mint="So11111111111111111111111111111111111111112",  # SOL
    limit=100
)
```

### Swap Methods

#### `get_swaps(program_id, amm=None, amm_pool=None, user=None, input_mint=None, output_mint=None, signature=None, start_time=None, end_time=None, order_by="timestamp", order_direction="desc", limit=10, page=1)`
Get Solana DEX swap transactions.

**Parameters:**
- `program_id` (SwapPrograms): Filter by swap program ID (required)
- `amm` (str, optional): Filter by AMM address
- `amm_pool` (str, optional): Filter by AMM pool address
- `user` (str, optional): Filter by user address
- `input_mint` (str, optional): Filter by input mint address
- `output_mint` (str, optional): Filter by output mint address
- `signature` (str, optional): Filter by transaction signature
- `start_time` (int, optional): Start time as UNIX timestamp
- `end_time` (int, optional): End time as UNIX timestamp
- `order_by` (OrderBy): Field to order by
- `order_direction` (OrderDirection): Order direction
- `limit` (int): Maximum results
- `page` (int): Page number

**Returns:** `SolanaSwapsResponse`

```python
from thegraph_client import SwapPrograms

swaps = await solana.get_swaps(
    program_id=SwapPrograms.RAYDIUM,
    user="9WzDXwBbmkg8ZTbNMqUxvQRAyrZzDsGYdLVL9zYtAWWM",
    limit=50
)
```

### Available Swap Programs

```python
from thegraph_client import SwapPrograms

SwapPrograms.RAYDIUM      # Raydium DEX
SwapPrograms.JUPITER      # Jupiter Aggregator
SwapPrograms.ORCA         # Orca DEX
```

---

## Type System

The client includes comprehensive TypeScript-style type definitions with runtime validation:

### Enums

```python
from thegraph_client import (
    NetworkId,           # EVM networks
    SolanaNetworkId,     # SVM networks
    TokenStandard,       # ERC721, ERC1155
    ActivityType,        # NFT activity types
    OrderDirection,      # ASC, DESC
    OrderBy,            # Ordering fields
    Interval,           # Time intervals
    Protocol,           # DEX protocols
    SolanaPrograms,     # Solana programs
    SwapPrograms        # Swap protocols
)
```

### Response Types

All API methods return typed response objects with full IntelliSense support:

```python
# Example response structure
balances: BalancesResponse = await evm.get_balances("0x...")
for balance in balances.data:
    print(f"Symbol: {balance['symbol']}")
    print(f"Value: {balance['value']}")
    print(f"Contract: {balance['contract']}")
```

### Validation

All responses are validated at runtime using `divine-type-enforcer`, ensuring type safety and catching API changes early.

## Authentication

Set your API key in one of these ways:

### Environment Variable
```bash
export THEGRAPH_API_KEY="your_bearer_token"
```

### Constructor Parameter
```python
api = TheGraphTokenAPI(api_key="your_bearer_token")
```

### Per-Client Basis
```python
evm = EVMTokenAPI(network=NetworkId.MAINNET, api_key="your_bearer_token")
svm = SVMTokenAPI(api_key="your_bearer_token")
```

## Error Handling

The client provides comprehensive error handling with typed exceptions:

```python
from thegraph_client import TheGraphTokenAPI
from divine_requests import NetworkingError

async def main():
    api = TheGraphTokenAPI(api_key="your_key")
    
    try:
        async with api.evm(NetworkId.MAINNET) as mainnet:
            balances = await mainnet.get_balances("0xinvalid")
    except NetworkingError as e:
        print(f"Network error: {e}")
    except ValueError as e:
        print(f"Validation error: {e}")
```

## Advanced Usage

### Multiple Networks

```python
import anyio

async def multi_network_example():
    api = TheGraphTokenAPI(api_key="your_key")
    
    # Query multiple networks simultaneously
    mainnet = api.evm(NetworkId.MAINNET)
    polygon = api.evm(NetworkId.POLYGON)
    base = api.evm(NetworkId.BASE)
    
    async with mainnet, polygon, base:
        # Run queries in parallel using anyio task groups
        async with anyio.create_task_group() as tg:
            tg.start_soon(mainnet.get_balances, "0x...")
            tg.start_soon(polygon.get_balances, "0x...")
            tg.start_soon(base.get_balances, "0x...")
```

### Custom Configuration

```python
# Custom base URL
api = TheGraphTokenAPI(
    api_key="your_key",
    base_url="https://custom-api.example.com"
)

# Per-network configuration
mainnet = EVMTokenAPI(
    network=NetworkId.MAINNET,
    api_key="mainnet_key",
    base_url="https://mainnet-api.example.com"
)
```

### Pagination

```python
async def get_all_nfts(address: str):
    async with api.evm(NetworkId.MAINNET) as mainnet:
        all_nfts = []
        page = 1
        
        while True:
            response = await mainnet.get_nft_ownerships(
                address=address,
                limit=1000,
                page=page
            )
            
            all_nfts.extend(response.data)
            
            if len(response.data) < 1000:
                break
                
            page += 1
        
        return all_nfts
```

## Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

## License

This project is licensed under the MIT License.
