PostPyro

High-Performance PostgreSQL Driver for Python Built with Rust

PostPyro combines the safety and performance of Rust with the simplicity of Python. Built with PyO3 and tokio-postgres, it provides DB-API 2.0 compliance while delivering superior performance through native Rust implementation.

Get Started PyPI Package
🔥 Rust Powered
⚡ High Performance
🛡️ Memory Safe
📦 Zero Dependencies
quick_example.py 
import PostPyro as pg

# Connect to PostgreSQL
conn = pg.Connection("postgresql://user:pass@localhost/db")

# Query data
users = conn.query("SELECT * FROM users WHERE active = $1", [True])
for user in users:
    print(f"User: {user['name']} ({user['email']})")

# Execute updates
affected = conn.execute(
    "UPDATE users SET last_login = NOW() WHERE id = $1", 
    [user_id]
)

# Use transactions
with conn.begin() as tx:
    tx.execute("INSERT INTO orders (user_id) VALUES ($1)", [user_id])
    tx.execute("UPDATE inventory SET stock = stock - 1")

conn.close()

Key Features

🔥

High Performance

Rust-powered backend with zero-copy data handling for maximum speed and efficiency.

🛡️

Memory Safe

Rust's ownership system prevents memory leaks and segfaults, ensuring reliability.

🌐

Full PostgreSQL Support

Complete support for all PostgreSQL data types, arrays, JSON, UUIDs, and network types.

⚡

Tokio Async I/O

Native async I/O under the hood powered by tokio-postgres for excellent performance.

🔒

Type Safety

Comprehensive type checking and automatic conversion between Python and PostgreSQL types.

🎯

DB-API 2.0 Compliant

Standard Python database interface ensures compatibility with existing code.

Installation

📦 Install from PyPI

pip install PostPyro

🔧 Build from Source

# Install Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

# Clone and build
git clone https://github.com/magi8101/PostPyro.git
cd PostPyro
pip install -e .

Requirements

  • Python 3.8+
  • PostgreSQL 9.6+ (server)
  • Rust 1.70+ (for building from source)

Quick Start

import PostPyro as pg

# Connect to database
conn = pg.Connection("postgresql://user:password@localhost:5432/mydb")

# Create a table
conn.execute("""
    CREATE TABLE users (
        id SERIAL PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        age INTEGER
    )
""")

# Insert data
affected = conn.execute(
    "INSERT INTO users (name, email, age) VALUES ($1, $2, $3)",
    ["Alice", "alice@example.com", 30]
)
print(f"Inserted {affected} rows")

# Close connection
conn.close()
# Query multiple rows
users = conn.query("SELECT * FROM users WHERE age > $1", [25])
for user in users:
    print(f"ID: {user['id']}, Name: {user['name']}, Age: {user['age']}")

# Query single row
user = conn.query_one("SELECT * FROM users WHERE id = $1", [1])
print(f"Found user: {user['name']}")

# Batch operations
queries = [
    "INSERT INTO users (name, age) VALUES ('Bob', 25)",
    "INSERT INTO users (name, age) VALUES ('Charlie', 35)",
    "INSERT INTO users (name, age) VALUES ('Diana', 28)"
]
results = conn.execute_batch(queries)
print(f"Batch inserted {sum(results)} rows")

# Health check
if conn.ping():
    print("✅ Connection is healthy")
# Automatic transaction with context manager
with conn.begin() as tx:
    tx.execute("INSERT INTO users (name) VALUES ($1)", ["Alice"])
    tx.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = $1", [1])
    # Automatically commits on success, rolls back on exception

# Manual transaction management
tx = conn.begin()
try:
    tx.execute("INSERT INTO orders (user_id, amount) VALUES ($1, $2)", [1, 99.99])
    tx.execute("UPDATE inventory SET stock = stock - 1 WHERE product_id = $1", [123])
    tx.commit()
except Exception as e:
    tx.rollback()
    print(f"Transaction failed: {e}")

# Query within transaction
with conn.begin() as tx:
    users = tx.query("SELECT * FROM users WHERE created_today = true")
    for user in users:
        tx.execute("UPDATE users SET welcomed = true WHERE id = $1", [user['id']])
# Prepared statements for better performance
stmt_id = conn.prepare("SELECT * FROM users WHERE department = $1")

# Connection info and status
info = conn.info()
print(f"Connection closed: {info['closed']}")
print(f"Connection healthy: {info['healthy']}")

# Working with complex types
from datetime import datetime, date
import uuid

conn.execute("""
    INSERT INTO complex_data (
        json_col, array_col, uuid_col, date_col, timestamp_col
    ) VALUES ($1, $2, $3, $4, $5)
""", [
    {"name": "John", "scores": [85, 92, 78]},  # JSON
    [1, 2, 3, 4, 5],                          # Array
    uuid.uuid4(),                             # UUID
    date.today(),                             # Date
    datetime.now()                            # Timestamp
])

# Convert Row to dict for pandas integration
import pandas as pd
rows = conn.query("SELECT * FROM users")
df = pd.DataFrame([row.to_dict() for row in rows])

API Reference

Module Functions

pg.connect(connection_string)

Create a new database connection using the connect function.

Parameters:
  • connection_string (str): PostgreSQL connection string
Returns:

Connection object

Example:
conn = pg.connect("postgresql://user:pass@localhost:5432/mydb")
pg.Connection(connection_string)

Create a new database connection using the Connection class.

Parameters:
  • connection_string (str): PostgreSQL connection string
Example:
conn = pg.Connection("postgresql://user:pass@localhost:5432/mydb")
pg.get_version()

Get the PostPyro driver version.

Returns:

Version string (e.g., "0.1.2")

Example:
version = pg.get_version()
print(f"PostPyro version: {version}")

Connection Class

conn.execute(query, params=None)

Execute INSERT, UPDATE, DELETE, or DDL statements.

Parameters:
  • query (str): SQL query string
  • params (list, optional): Query parameters using $1, $2, ... placeholders
Returns:

Number of rows affected (int)

Examples:
# INSERT
affected = conn.execute("INSERT INTO users (name, age) VALUES ($1, $2)", ["Alice", 30])

# UPDATE  
affected = conn.execute("UPDATE users SET age = $1 WHERE name = $2", [31, "Alice"])

# DELETE
affected = conn.execute("DELETE FROM users WHERE age < $1", [18])
conn.query(query, params=None)

Execute SELECT queries and return all matching rows.

Parameters:
  • query (str): SQL SELECT statement
  • params (list, optional): Query parameters
Returns:

List of Row objects

Example:
rows = conn.query("SELECT id, name, age FROM users WHERE age > $1", [25])
for row in rows:
    print(f"ID: {row['id']}, Name: {row['name']}, Age: {row['age']}")
conn.query_one(query, params=None)

Execute SELECT query and return exactly one row.

Parameters:
  • query (str): SQL SELECT statement
  • params (list, optional): Query parameters
Returns:

Single Row object

Raises:

Error if zero or multiple rows returned

Example:
user = conn.query_one("SELECT * FROM users WHERE id = $1", [1])
print(f"User name: {user['name']}")
conn.begin()

Begin a new transaction and return a Transaction object.

Returns:

Transaction object (context manager)

Example:
with conn.begin() as tx:
    tx.execute("INSERT INTO users (name) VALUES ($1)", ["Alice"])
    tx.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = $1", [1])
    # Automatically commits on success, rolls back on exception

Row Class

Represents a single row from a query result with dict-like interface.

row[key]

Access column values by index or name.

Example:
row = conn.query_one("SELECT id, name, email FROM users WHERE id = $1", [1])

# Access by column name
print(row['name'])
print(row['email'])

# Access by index  
print(row[0])  # id
print(row[1])  # name
row.to_dict()

Convert row to a Python dictionary.

Returns:

Dictionary with column names as keys

Example:
user_dict = row.to_dict()
print(user_dict)  # {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}

Transaction Class

Represents a database transaction with automatic rollback on errors.

tx.execute(query, params=None)

Execute a statement within the transaction.

Example:
with conn.begin() as tx:
    tx.execute("INSERT INTO users (name) VALUES ($1)", ["Alice"])
    tx.execute("UPDATE accounts SET balance = balance - $1 WHERE id = $2", [100, 1])
tx.query(query, params=None)

Execute a query within the transaction.

Example:
with conn.begin() as tx:
    users = tx.query("SELECT * FROM users WHERE created_today = true")
    for user in users:
        tx.execute("UPDATE users SET welcomed = true WHERE id = $1", [user['id']])

Error Handling

PostPyro provides comprehensive PostgreSQL error mapping with specific exception types.

Exception Hierarchy

DatabaseError                    # Base database error
├── InterfaceError              # Driver interface problems
├── DataError                   # Data processing errors
├── OperationalError            # Database operation errors
├── IntegrityError              # Constraint violations
├── InternalError               # Internal database errors
├── ProgrammingError            # SQL programming errors
└── NotSupportedError           # Unsupported operations
Error Handling Example:
import PostPyro as pg

try:
    conn = pg.Connection("postgresql://user:pass@localhost/db")
    conn.execute("INSERT INTO users (email) VALUES ($1)", ["invalid-email"])
    
except pg.IntegrityError as e:
    print(f"Constraint violation: {e}")
    
except pg.OperationalError as e:
    print(f"Database operation failed: {e}")
    
except pg.ProgrammingError as e:
    print(f"SQL syntax error: {e}")
    
except pg.DatabaseError as e:
    print(f"General database error: {e}")

Type System

PostPyro automatically converts between Python and PostgreSQL types.

PostgreSQL Type Python Type Example
BOOLEAN bool True, False
INTEGER int 42, -123
REAL float 3.14, 2.718
TEXT str "Hello World"
BYTEA bytes b"binary data"
DATE datetime.date date(2023, 12, 25)
TIMESTAMP datetime.datetime datetime(2023, 12, 25, 14, 30)
UUID uuid.UUID uuid.uuid4()
JSON dict, list {"key": "value"}
ARRAY list [1, 2, 3]

Examples

🐼 Pandas Integration

import pandas as pd
import PostPyro as pg

conn = pg.Connection("postgresql://user:pass@localhost/db")

# Query to DataFrame
rows = conn.query("SELECT * FROM sales_data")
df = pd.DataFrame([row.to_dict() for row in rows])

print(df.head())

🚀 FastAPI Integration

from fastapi import FastAPI, HTTPException
import PostPyro as pg

app = FastAPI()
conn = pg.Connection("postgresql://user:pass@localhost/db")

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        user = conn.query_one(
            "SELECT id, name, email FROM users WHERE id = $1", 
            [user_id]
        )
        return user.to_dict()
    except pg.DatabaseError:
        raise HTTPException(status_code=404, detail="User not found")

🔧 Connection Pool Pattern

import PostPyro as pg
from queue import Queue

class ConnectionPool:
    def __init__(self, connection_string, pool_size=5):
        self.pool = Queue(maxsize=pool_size)
        
        # Initialize pool
        for _ in range(pool_size):
            conn = pg.Connection(connection_string)
            self.pool.put(conn)
    
    def get_connection(self):
        return self.pool.get(timeout=10)
    
    def return_connection(self, conn):
        if not conn.is_closed():
            self.pool.put(conn)

💾 Batch Processing

def bulk_insert_users(conn, users_data):
    # Method 1: Single transaction
    with conn.begin() as tx:
        for user_data in users_data:
            tx.execute(
                "INSERT INTO users (name, email, age) VALUES ($1, $2, $3)",
                [user_data['name'], user_data['email'], user_data['age']]
            )
    
    # Method 2: Batch execution
    queries = []
    for user_data in users_data:
        queries.append(
            f"INSERT INTO users (name, email, age) VALUES "
            f"('{user_data['name']}', '{user_data['email']}', {user_data['age']})"
        )
    
    conn.execute_batch(queries)

Performance

Why PostPyro is Faster

🦀

Rust Backend

Native performance without Python interpreter overhead

⚡

Zero-Copy Operations

Direct memory mapping between PostgreSQL and Python

🌊

Async I/O

Tokio-powered async networking under the hood

📦

No Dependencies

No external Python dependencies to slow things down

Driver Comparison

Feature PostPyro psycopg2 asyncpg psycopg3
Language Rust + Python C + Python Cython C + Python
Performance ⭐⭐⭐⭐⭐ ⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐⭐
Memory Safety ✅ Rust ❌ Manual C ⚠️ Cython ❌ Manual C
Installation 📦 Wheel 🔧 Compilation 📦 Wheel 🔧 Compilation
Dependencies 🎯 Zero 📦 Many 📦 Few 📦 Many
API Simplicity ✅ Simple ⚠️ Complex ⚠️ Async Only ⚠️ Complex

Performance Benchmark

import time
import PostPyro as pg

conn = pg.Connection("postgresql://user:pass@localhost/db")

# Benchmark: Insert 10,000 records
start_time = time.time()

queries = []
for i in range(10000):
    queries.append(f"INSERT INTO benchmark (value) VALUES ({i})")

conn.execute_batch(queries)

elapsed = time.time() - start_time
print(f"Inserted 10,000 records in {elapsed:.2f} seconds")
print(f"Rate: {10000/elapsed:.0f} inserts/second")