-
-
import asyncio
+
+
+
+
+
+
+
+
+
+
import asyncio
import PostPyro
async def main():
- # Connect to database (returns a Pool)
pool = await PostPyro.connect("postgresql://user:password@localhost:5432/mydb")
- # Create a table
await pool.execute("""
CREATE TABLE users (
id SERIAL PRIMARY KEY,
@@ -278,840 +267,445 @@ Quick Start
)
""")
- # Insert data
affected = await pool.execute(
"INSERT INTO users (name, email, age) VALUES ($1, $2, $3)",
["Alice", "alice@example.com", 30]
)
print(f"Inserted {affected} rows")
- # Close the pool
await pool.close()
asyncio.run(main())
-
+
-
-
# Query multiple rows
+
+
# Query multiple rows
users = await pool.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 (raises ProgrammingError on zero matches;
-# extra matches beyond the first are silently discarded)
+# Query a single row - raises ProgrammingError on zero matches;
+# extra matches beyond the first are silently discarded
user = await pool.query_one("SELECT * FROM users WHERE id = $1", [1])
print(f"Found user: {user['name']}")
# There's no execute_batch() - run each statement individually
-names_and_ages = [("Bob", 25), ("Charlie", 35), ("Diana", 28)]
-inserted = 0
-for name, age in names_and_ages:
- inserted += await pool.execute(
+for name, age in [("Bob", 25), ("Charlie", 35), ("Diana", 28)]:
+ await pool.execute(
"INSERT INTO users (name, age) VALUES ($1, $2)", [name, age]
)
-print(f"Inserted {inserted} rows")
-# Pool status check
if not pool.is_closed():
- print("✅ Pool is open")
-
+ print("Pool is open")
+
-
-
# pool.transaction() is a real BEGIN round-trip, so it must be
+
+
# pool.transaction() is a real BEGIN round-trip, so it must be
# awaited before it can be used as a context manager -
# `async with pool.transaction():` raises TypeError.
tx = await pool.transaction()
async with tx:
await tx.execute("INSERT INTO users (name) VALUES ($1)", ["Alice"])
await tx.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = $1", [1])
- # Automatically commits on success, rolls back on exception
+ # Commits automatically on success, rolls back on exception
-# Manual transaction management (commit()/rollback() also
-# work outside a `with` block)
+# Manual transaction management also works outside a `with` block
tx = await pool.transaction()
try:
await tx.execute("INSERT INTO orders (user_id, amount) VALUES ($1, $2)", [1, 99.99])
- await tx.execute("UPDATE inventory SET stock = stock - 1 WHERE product_id = $1", [123])
await tx.commit()
except Exception as e:
await tx.rollback()
- print(f"Transaction failed: {e}")
-
-# Query within a transaction
-tx = await pool.transaction()
-async with tx:
- users = await tx.query("SELECT * FROM users WHERE created_today = true")
- for user in users:
- await tx.execute("UPDATE users SET welcomed = true WHERE id = $1", [user['id']])
-
+ print(f"Transaction failed: {e}")
+
-
-
# bind_params only auto-converts bool/int/float/str/None -
-# anything else (date, datetime, uuid, dict/list for JSON) has
-# to be passed as a string with an explicit Postgres cast.
-import json
+
+
# Non-primitive types bind natively - no str() + ::type cast needed.
import uuid
-from datetime import date, datetime
-
-await pool.execute("""
- INSERT INTO complex_data (
- json_col, uuid_col, date_col, timestamp_col
- ) VALUES ($1::jsonb, $2::uuid, $3::date, $4::timestamp)
-""", [
- json.dumps({"name": "John", "scores": [85, 92, 78]}), # JSON
- str(uuid.uuid4()), # UUID
- date.today().isoformat(), # Date
- datetime.now().isoformat(), # Timestamp
-])
-
-# Reading back decodes to the right Python type automatically -
-# this direction IS automatic (see the Type System reference below).
-
-# Convert Row to dict for pandas integration
-import pandas as pd
-rows = await pool.query("SELECT * FROM users")
-df = pd.DataFrame([row.to_dict() for row in rows])
-
+from datetime import datetime, timezone
+from decimal import Decimal
+
+await pool.execute(
+ "INSERT INTO events (at, uid, price, meta, blob) VALUES ($1,$2,$3,$4,$5)",
+ [
+ datetime.now(timezone.utc), # -> TIMESTAMPTZ
+ uuid.uuid4(), # -> UUID
+ Decimal("9.99"), # -> NUMERIC, exact
+ {"ok": True, "tags": ["a", "b"]},# -> JSONB
+ b"\x00payload", # -> BYTEA
+ ],
+)
+
+# A homogeneous list/tuple binds as a real Postgres array
+await pool.execute(
+ "INSERT INTO tags (id, names, scores) VALUES ($1, $2, $3)",
+ [1, ["red", "blue"], [10, 20, None]], # None -> NULL array element
+)
+
+# Reading back decodes to the right Python type automatically
+row = await pool.query_one("SELECT * FROM events WHERE id = $1", [1])
+assert isinstance(row["at"], datetime)