Postgres Laravel Package

Getting Started

Install and connect. composer require amphp/postgres pulls in amphp/amp, amphp/sql, amphp/sql-common, and revolt/event-loop. There is no C extension and no ext-pgsql/ext-pq requirement — the client speaks the Postgres wire protocol over plain PHP sockets, which also means pg_*() functions and PDO are simply not in the picture.

Build a config, build a pool once per process, and hand the pool around:

<?php
require __DIR__ . '/vendor/autoload.php';

use Amp\Postgres\PostgresConfig;
use Amp\Postgres\PostgresConnectionPool;

$config = new PostgresConfig(
    host: '127.0.0.1',
    port: 5432,
    user: 'app',
    password: getenv('DB_PASSWORD'),
    database: 'app',
    maxConnections: 10,
);

$pool = PostgresConnectionPool::create($config);

$result = $pool->execute('SELECT id, email FROM users WHERE id = $1', [42]);

foreach ($result as $row) {
    printf("%d %s\n", $row['id'], $row['email']);
}

PostgresConfig::fromString('host=… user=… password=… dbname=…') exists if you'd rather drive it from an env var; it's a small hand-written parser, so check the class for exactly which keys it honours before you copy a libpq URI into it.

Where to look first. The README plus five files cover 95% of day-to-day use: src/PostgresConfig.php (all connection knobs), src/PostgresConnectionPool.php, src/PostgresConnection.php (the actual API surface: execute, query, prepare, beginTransaction), src/PostgresResult.php / src/PostgresRow.php, and src/PostgresTransaction.php. Then skim amphp/sql's interfaces (SqlExecutor, SqlLink, SqlResult) because the pool implements them — that's also how you write code that's testable and swappable.

First real use case. The sweet spot is a long-lived async process: an amphp/http-server API, a queue worker, a WebSocket fan-out service, or a CLI that has to hit the database N times concurrently. Start with the boring version — one request handler doing one execute() — then immediately convert a "loop over ids and query each" job into a Future\await() batch (see below). That single change is the reason to adopt this package over PDO; almost everything else is a lateral move.

If your first instinct is "drop this under Eloquent/Doctrine", stop: there's no PDO driver here and no ORM will see it. You're writing SQL.

Implementation Patterns

One pool per process, injected, never per-request. The pool is the connection manager — lazily opening connections up to maxConnections, reaping them after idleTimeout. Creating a pool per request is the single most common port-from-PDO mistake and it turns your async app into a connection-storm generator. Construct it in your bootstrap, pass it into handlers/repositories, and warm it with a fire-and-forget Amp\async(fn () => $pool->execute('SELECT 1')) if you care about p99 on the first few requests.

Concurrency is the point. Sequential awaits in a loop are the anti-pattern. Batch instead:

use Amp\Future;
use function Amp\async;

$futures = [];
foreach ($userIds as $id) {
    $futures[$id] = async(fn () => $pool->execute(
        'SELECT id, email FROM users WHERE id = $1', [$id]
    ));
}
$results = Future\await($futures);          // all dispatched, all awaited
$results = Future\await($futures, $cancellation); // or bounded with a cancellation/timeout

Cap the fan-out with Amp\Sync\LocalSemaphore (or chunk the ids) — Future\await() over 10k ids will happily queue 10k queries behind 10 connections and blow up memory while providing zero extra throughput. A semaphore sized to maxConnections * 2 is usually the right shape.

Streaming reads for anything unbounded. Use query() and iterate; rows are pulled through an amphp/pipeline and you get backpressure for free:

foreach ($pool->query('SELECT * FROM events WHERE created_at > now() - interval \'1 day\'') as $row) {
    $sink->push($row['payload']);   // memory stays flat
}

Keep the foreach tight. If you break early or store every row, you either hold the connection with unread tuples or you've just rebuilt the buffered version with extra steps.

Transactions pin a connection — use the transaction object, exclusively.

$tx = $pool->beginTransaction();
try {
    $tx->execute('UPDATE accounts SET balance = balance - $1 WHERE id = $2', [$amt, $from]);
    $tx->execute('UPDATE accounts SET balance = balance + $1 WHERE id = $2', [$amt, $to]);
    $tx->commit();
} catch (\Throwable $e) {
    $tx->rollback();
    throw $e;
}

Calling $pool->execute() inside that try block gets a different connection from the pool and runs outside your transaction. This is silent, passes tests under low concurrency, and corrupts data in production. Grep for it during review; it's worth a lint rule.

Session state belongs on a dedicated connection. SET, advisory locks, LISTEN, temp tables — none of these survive a pool hop. Use PostgresConnection::connect($config) directly, keep that object for the process lifetime, and close it on shutdown. For per-transaction timeouts prefer SET LOCAL statement_timeout = '5s' as the first statement inside $tx, which is connection-pinned by construction.

Thin repository layer, SQL in one place. A class per aggregate, methods returning readonly DTOs, $row['col'] mapped explicitly. You get three things you won't get from the driver: a place to put RETURNING, a seam for tests (interface + in-memory fake), and protection from the "row is ArrayAccess so let's pass it around" drift that makes every later schema change a scavenger hunt.

Writes with RETURNING, always. There's no meaningful last-insert-id in Postgres and getLastInsertId() won't help you:

$result = $pool->execute(
    'INSERT INTO users (email, meta) VALUES ($1, $2::jsonb) RETURNING id, created_at',
    [$email, json_encode($meta)]
);
$row = $result->current();          // or iterate / iterator_to_array()
$id = $row['id'];

For bulk writes, skip the loop-with-async and generate one statement: either multi-row VALUES ($1,$2),($3,$4)… or INSERT … SELECT * FROM unnest($1::text[], $2::int[]) — the latter keeps parameter count flat and is dramatically faster for a few thousand rows. COPY isn't exposed by this client, so for genuine bulk loads shell out to psql \copy from a worker instead of fighting it.

Error handling keyed on SQLSTATE. Wrap PostgresQueryError at the repository boundary and translate the handful of codes you actually act on: 23505 unique violation → 409/upsert retry, 23503 FK violation → 422, 40001 serialization failure → retry the transaction, 57014 query canceled → timeout response. Everything else should bubble up as a 500 with the query logged (never the parameters with PII).

Test against real Postgres. There's no fake backend and no in-memory mode. Use a Docker-based Postgres in CI, truncate between tests, and keep the repository interface as the mock seam for unit tests. Async tests run fine under amphp/phpunit-util (AsyncTestCase) if you need to await in test bodies.

Gotchas and Tips

• Placeholders are $1, $2, not ? and not :name. Every PDO snippet you copy will fail. Named parameters don't exist; if you want them, build the mapping in your repository layer.
• "could not determine data type of parameter $1" — happens with null params, empty arrays, and INSERT INTO t VALUES ($1) style statements where the planner has nothing to infer from. Fix with an explicit cast ($1::text, $1::jsonb, $1::uuid[]). Cheap habit: cast every placeholder in dynamic queries.
• Type round-tripping is partial. Expect native int/float/bool/null/string for the basic scalars, and expect numeric, json/jsonb, timestamps, intervals, and arrays to arrive as strings. Don't assume — var_dump one row per table once, write the decoding into your DTO constructor, and move on. If you ever see 't'/'f' in a boolean column, something got cast to text; fix it in SQL (::boolean), not in PHP.
• Pool exhaustion is the failure mode you'll actually hit. Default maxConnections is small; a handler that awaits an HTTP call while holding a transaction can deadlock the whole process under load. Rule: never await anything unrelated between beginTransaction() and commit(). If you must, restructure into two transactions.
• idleTimeout reaping plus server-side idle_in_transaction_session_timeout/PgBouncer means connections die under you. The pool doesn't transparently replay a failed query. Retry idempotent reads once on a connection-closed error; never blind-retry writes outside a transaction you control.
• PgBouncer in transaction mode will bite. Anything that relies on server-side prepared statements or session state can produce "prepared statement already exists" or missing-SET bugs. Use session pooling for this client, or keep statements trivially simple and set state per-transaction.
• Timeouts. There's no universal per-query timeout knob; the reliable levers are connectTimeout in the config, SET LOCAL statement_timeout inside transactions, and wrapping fan-out in a TimeoutCancellation at the Future\await() level. Add a watchdog at the call site rather than assuming cancellation aborts a running server query — Postgres needs a separate cancel connection for that, so a "cancelled" client may leave the statement running briefly.
• TLS defaults deserve a look. Check PostgresConfig for the SSL options your version supports and confirm whether sslmode=require is verifying the certificate. In managed-Postgres environments, don't assume; verify.
• Blocking calls poison everything. No sleep(), no sync file_get_contents() against a remote URL, no PDO fallback inside an async handler — one blocking call stalls the event loop and therefore every query in flight in that process. This is the number-one production incident with any Amp client.
• query() results are lazy; exceptions surface during iteration, not at the call site. Wrap the foreach, not just the call.
• Column collisions in joins silently overwrite. SELECT a.*, b.* FROM … gives you the last column of each name. Alias explicitly (b.id AS b_id); ArrayAccess won't warn you.
• No migrations, no schema tooling, no query builder, no ORM. Run migrations out-of-band (psql, Phinx, a Doctrine migrations step in deploy) — not from inside your async app, where a lock wait will just stall the loop.
• Version alignment matters. The 2.x line is Amp v3 / fibers; 1.x is the old Amp v2 / Promise world. Mixing them across amphp/* packages produces bizarre autoload-level breakage. Pin ^2 and let Composer resolve the rest together.
• When debugging, log SQL + getRowCount() + elapsed time at the repository boundary, and log the PostgresQueryError diagnostics (SQLSTATE, detail, hint, constraint name) rather than just getMessage(). The hint field routinely spells out the fix, especially for the parameter-type errors above.