import os
import sys
from typing import Optional, Dict, Any, List
try:
import requests
except ImportError:
print("Erro: requests não instalado.")
print("Execute: pip install requests")
sys.exit(1)
class PGAPIClient:
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url.rstrip('/')
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
'X-API-Key': api_key,
'Content-Type': 'application/json'
})
def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
url = f"{self.base_url}{endpoint}"
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
return response.json()
def health(self) -> Dict[str, Any]:
return self._request('GET', '/health')
def get_account(self) -> Dict[str, Any]:
return self._request('GET', '/v1/account')
def get_usage(self) -> Dict[str, Any]:
return self._request('GET', '/v1/account/usage')
def list_databases(self) -> Dict[str, Any]:
return self._request('GET', '/v1/databases')
def query(self, database: str, sql: str, params: Optional[List] = None) -> Dict[str, Any]:
data = {
'database': database,
'query': sql
}
if params:
data['params'] = params
return self._request('POST', '/v1/query', json=data)
def batch(self, queries: List[Dict[str, Any]]) -> Dict[str, Any]:
return self._request('POST', '/v1/batch', json=queries)
def transaction(self, database: str, queries: List[Dict[str, Any]]) -> Dict[str, Any]:
data = {
'database': database,
'queries': queries
}
return self._request('POST', '/v1/transaction', json=data)
def get_schema(self, database: str) -> Dict[str, Any]:
return self._request('GET', f'/v1/databases/{database}/schema')
def list_tables(self, database: str) -> Dict[str, Any]:
return self._request('GET', f'/v1/databases/{database}/tables')
def print_section(title: str):
print("\n" + "=" * 50)
print(f" {title}")
print("=" * 50 + "\n")
def main():
api_key = os.getenv('API_KEY', 'sk_dev_123456789')
base_url = os.getenv('BASE_URL', 'http://localhost:8580')
print("=" * 50)
print(" PG-API Python Client Example")
print("=" * 50)
print(f"\nConfiguração:")
print(f" Base URL: {base_url}")
print(f" API Key: {api_key[:10]}...")
client = PGAPIClient(base_url, api_key)
try:
print_section("1. Health Check")
result = client.health()
print(f"Status: {result['status']}")
print(f"Versão: {result['version']}")
print_section("2. Informações da Conta")
result = client.get_account()
account = result['data']
print(f"ID: {account['id']}")
print(f"Nome: {account['name']}")
print(f"Role: {account['role']}")
print(f"Databases: {', '.join(account['databases'])}")
print_section("3. Versão do PostgreSQL")
result = client.query('postgres', 'SELECT version()')
print(f"Resultado: {result['data']['rows'][0]['version']}")
print(f"Tempo de execução: {result['metadata']['execution_time_ms']}ms")
print_section("4. Criar Tabela de Exemplo")
result = client.query(
'postgres',
'''CREATE TABLE IF NOT EXISTS python_demo (
id SERIAL PRIMARY KEY,
name VARCHAR(100),
email VARCHAR(100),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)'''
)
print(f"Tabela criada: {result['success']}")
print_section("5. Inserir Dados")
result = client.query(
'postgres',
'INSERT INTO python_demo (name, email) VALUES ($1, $2), ($3, $4) RETURNING *',
['Alice', 'alice@example.com', 'Bob', 'bob@example.com']
)
print(f"Linhas inseridas: {len(result['data']['rows'])}")
for row in result['data']['rows']:
print(f" - {row['name']} ({row['email']})")
print_section("6. Consultar Dados")
result = client.query(
'postgres',
'SELECT * FROM python_demo ORDER BY id'
)
print(f"Total de registros: {result['data']['row_count']}")
for row in result['data']['rows']:
print(f" ID: {row['id']}, Name: {row['name']}, Email: {row['email']}")
print_section("7. Query com Parâmetros")
result = client.query(
'postgres',
'SELECT * FROM python_demo WHERE name = $1',
['Alice']
)
print(f"Encontrado: {result['data']['row_count']} registro(s)")
if result['data']['rows']:
print(f" Dados: {result['data']['rows'][0]}")
print_section("8. Batch de Queries")
batch_queries = [
{
'database': 'postgres',
'query': 'SELECT $1::text as msg',
'params': ['Batch Query 1']
},
{
'database': 'postgres',
'query': 'SELECT $1::text as msg',
'params': ['Batch Query 2']
},
{
'database': 'postgres',
'query': 'SELECT COUNT(*) as count FROM python_demo'
}
]
result = client.batch(batch_queries)
print(f"Queries executadas: {len(result['data'])}")
for i, res in enumerate(result['data']):
print(f" Query {i+1}: {res['row_count']} linha(s)")
print_section("9. Transação")
tx_queries = [
{
'query': 'INSERT INTO python_demo (name, email) VALUES ($1, $2) RETURNING id',
'params': ['Transaction User 1', 'tx1@example.com']
},
{
'query': 'INSERT INTO python_demo (name, email) VALUES ($1, $2) RETURNING id',
'params': ['Transaction User 2', 'tx2@example.com']
}
]
result = client.transaction('postgres', tx_queries)
print(f"Transação commitada: {result['data']['committed']}")
print(f"Queries executadas: {len(result['data']['results'])}")
print_section("10. Listar Tabelas")
result = client.list_tables('postgres')
tables = result['data']['tables']
print(f"Tabelas encontradas: {len(tables)}")
print(f" {', '.join(tables[:5])}{'...' if len(tables) > 5 else ''}")
print_section("11. Limpar Dados de Exemplo")
result = client.query(
'postgres',
"DELETE FROM python_demo WHERE name LIKE $1",
['%User%']
)
print(f"Registros removidos: {result['metadata']['rows_affected']}")
print_section("12. Dropar Tabela")
result = client.query('postgres', 'DROP TABLE IF EXISTS python_demo')
print(f"Tabela removida: {result['success']}")
print_section("13. Estatísticas de Uso")
result = client.get_usage()
usage = result['data']
print(f"Requisições hoje: {usage.get('requests_today', 'N/A')}")
print(f"Conexões ativas: {usage.get('active_connections', 'N/A')}")
print(f"Rate limit restante: {usage.get('rate_limit_remaining', 'N/A')}")
print_section("Exemplos Concluídos! ✅")
print("Todos os exemplos foram executados com sucesso!")
except requests.exceptions.ConnectionError:
print(f"\n❌ Erro: Não foi possível conectar a {base_url}")
print("Verifique se o servidor PG-API está rodando.")
sys.exit(1)
except requests.exceptions.HTTPError as e:
print(f"\n❌ Erro HTTP: {e}")
print(f"Resposta: {e.response.text}")
sys.exit(1)
except Exception as e:
print(f"\n❌ Erro: {e}")
sys.exit(1)
if __name__ == '__main__':
main()