pg-api 0.1.0

A high-performance PostgreSQL REST API driver with rate limiting, connection pooling, and observability
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

#!/usr/bin/env python3
"""
Example Python client for pg-api
"""

import requests
import json
from typing import List, Dict, Any, Optional

class PgApiClient:
    """Simple Python client for pg-api"""
    
    def __init__(self, base_url: str, api_token: str):
        self.base_url = base_url.rstrip('/')
        self.api_token = api_token
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_token}',
            'Content-Type': 'application/json'
        })
    
    def query(self, query: str, params: Optional[List] = None, database: str = "main") -> Dict[str, Any]:
        """Execute a single query"""
        response = self.session.post(
            f"{self.base_url}/v1/query",
            json={
                "query": query,
                "params": params or [],
                "database": database
            }
        )
        response.raise_for_status()
        return response.json()
    
    def batch(self, queries: List[Dict], database: str = "main") -> List[Dict[str, Any]]:
        """Execute multiple queries in batch"""
        response = self.session.post(
            f"{self.base_url}/v1/batch",
            json={
                "queries": queries,
                "database": database
            }
        )
        response.raise_for_status()
        return response.json()
    
    def transaction(self, queries: List[Dict], database: str = "main") -> List[Dict[str, Any]]:
        """Execute queries in a transaction"""
        response = self.session.post(
            f"{self.base_url}/v1/transaction",
            json={
                "queries": queries,
                "database": database
            }
        )
        response.raise_for_status()
        return response.json()
    
    def list_databases(self) -> List[str]:
        """List all databases"""
        response = self.session.get(f"{self.base_url}/v1/databases")
        response.raise_for_status()
        return response.json()
    
    def list_tables(self, database: str) -> List[Dict[str, Any]]:
        """List tables in a database"""
        response = self.session.get(f"{self.base_url}/v1/databases/{database}/tables")
        response.raise_for_status()
        return response.json()
    
    def get_schema(self, database: str) -> Dict[str, Any]:
        """Get full database schema"""
        response = self.session.get(f"{self.base_url}/v1/databases/{database}/schema")
        response.raise_for_status()
        return response.json()
    
    def health_check(self) -> Dict[str, Any]:
        """Check API health"""
        response = requests.get(f"{self.base_url}/health")
        response.raise_for_status()
        return response.json()


def main():
    # Initialize client
    client = PgApiClient(
        base_url="http://localhost:8580",
        api_token="sk_live_your_token_here"
    )
    
    # Check health
    print("Health check:", client.health_check())
    
    # Simple query
    result = client.query(
        "SELECT * FROM users WHERE active = $1 LIMIT 5",
        params=[True]
    )
    print("\nActive users:", json.dumps(result, indent=2))
    
    # Batch operations
    batch_results = client.batch([
        {
            "query": "SELECT COUNT(*) as total FROM users",
            "params": []
        },
        {
            "query": "SELECT COUNT(*) as active FROM users WHERE active = true",
            "params": []
        }
    ])
    print("\nBatch results:", json.dumps(batch_results, indent=2))
    
    # List databases and tables
    databases = client.list_databases()
    print("\nDatabases:", databases)
    
    if databases:
        tables = client.list_tables(databases[0])
        print(f"\nTables in {databases[0]}:", json.dumps(tables, indent=2))


if __name__ == "__main__":
    main()