NAME EV::Pg - asynchronous PostgreSQL client using libpq and EV SYNOPSIS use v5.10; use EV; use EV::Pg; my $pg = EV::Pg->new( conninfo => 'dbname=mydb', on_error => sub { die "PG error: $_[0]\n" }, ); $pg->on_connect(sub { $pg->query_params( 'select $1::int + $2::int', [10, 20], sub { my ($rows, $err) = @_; die $err if $err; say $rows->[0][0]; # 30 EV::break; }, ); }); EV::run; DESCRIPTION EV::Pg is a non-blocking PostgreSQL client that integrates with the EV event loop. It drives the libpq async API ("PQsendQuery", "PQconsumeInput", "PQgetResult") via "ev_io" watchers on the libpq socket, so the event loop never blocks on database I/O. Features: parameterized queries, prepared statements, pipeline mode, single-row mode, chunked rows (libpq >= 17), COPY IN/OUT, LISTEN/NOTIFY, async cancel (libpq >= 17), structured error fields, protocol tracing, and notice handling. CALLBACKS Query callbacks receive "($result)" on success, "(undef, $error)" on error: SELECT / single-row mode "(\@rows)" where each row is an arrayref of column values. "undef" columns map to Perl "undef". INSERT / UPDATE / DELETE "($cmd_tuples)" -- the string returned by "PQcmdTuples" (e.g. "1", "0"). Describe "(\%meta)" with keys "nfields", "nparams", and (when non-zero) "fields" (arrayref of "{name, type}" hashes) and "paramtypes" (arrayref of OIDs). COPY "("COPY_IN")", "("COPY_OUT")", or "("COPY_BOTH")". Pipeline sync "(1)". Error "(undef, $error_message)". Exceptions thrown inside callbacks are caught and emitted as warnings. CONSTRUCTOR new my $pg = EV::Pg->new(%args); Arguments: conninfo libpq connection string. If provided, "connect" is called immediately. conninfo_params Hashref of connection parameters (e.g. "{ host => 'localhost', dbname => 'mydb', port => '5432' }"). Alternative to "conninfo". If provided, "connect_params" is called immediately. expand_dbname If true and "conninfo_params" is used, the "dbname" value is parsed as a connection string (allowing "dbname => 'postgresql://...'"). on_connect Callback invoked (with no arguments) when the connection is established. on_error Callback invoked as "($error_message)" on connection-level errors. Defaults to "sub { die @_ }". on_notify Callback invoked as "($channel, $payload, $backend_pid)" on LISTEN/NOTIFY messages. on_notice Callback invoked as "($message)" on PostgreSQL notice/warning messages. on_drain Callback invoked (with no arguments) when the send buffer has been flushed during COPY IN. Useful for resuming "put_copy_data" after it returns 0. keep_alive When true, the connection keeps "EV::run" alive even when no queries are pending. See "keep_alive". loop An EV loop object. Defaults to "EV::default_loop". CONNECTION METHODS connect $pg->connect($conninfo); Initiates an asynchronous connection. The "on_connect" handler fires on success; "on_error" fires on failure. connect_params $pg->connect_params(\%params); $pg->connect_params(\%params, $expand_dbname); Initiates an asynchronous connection using keyword/value parameters instead of a connection string. $expand_dbname allows the "dbname" parameter to contain a full connection URI. reset $pg->reset; Drops the current connection and reconnects using the original conninfo. Pending callbacks receive "(undef, "connection reset")". Alias: "reconnect". finish $pg->finish; Closes the connection. Pending callbacks receive "(undef, "connection finished")". Alias: "disconnect". is_connected my $bool = $pg->is_connected; Returns 1 if connected and ready for queries. status my $st = $pg->status; Returns the libpq connection status ("CONNECTION_OK" or "CONNECTION_BAD"). QUERY METHODS query $pg->query($sql, sub { my ($result, $err) = @_; }); Sends a simple query. Not allowed in pipeline mode -- use "query_params" instead. Multi-statement strings (e.g. "SELECT 1; SELECT 2") are supported but only the last result is delivered to the callback. PostgreSQL stops executing after the first error, so errors always appear as the last result. query_params $pg->query_params($sql, \@params, sub { my ($result, $err) = @_; }); Sends a parameterized query. Parameters are referenced in SQL as $1, $2, etc. "undef" values are sent as SQL NULL. prepare $pg->prepare($name, $sql, sub { my ($result, $err) = @_; }); Creates a prepared statement. The callback receives an empty string ("") on success. Alias: "prep". query_prepared $pg->query_prepared($name, \@params, sub { my ($result, $err) = @_; }); Executes a prepared statement. Alias: "qx". describe_prepared $pg->describe_prepared($name, sub { my ($meta, $err) = @_; }); Describes a prepared statement. The callback receives a hashref with keys "nfields" and "nparams". When "nfields" is non-zero, a "fields" key is also present (arrayref of "{name, type}" hashes). When "nparams" is non-zero, a "paramtypes" key is also present (arrayref of OIDs). describe_portal $pg->describe_portal($name, sub { my ($meta, $err) = @_; }); Describes a portal. The callback receives the same hashref structure as "describe_prepared". set_single_row_mode my $ok = $pg->set_single_row_mode; Switches the most recently sent query to single-row mode. Returns 1 on success, 0 on failure (e.g. no query pending). The callback fires once per row with "(\@rows)" where @rows is an arrayref containing a single row (e.g. "[[$col1, $col2, ...]]"), then a final empty "(\@rows)" (where @rows has zero elements) for the completion. set_chunked_rows_mode my $ok = $pg->set_chunked_rows_mode($chunk_size); Switches the most recently sent query to chunked rows mode, delivering up to $chunk_size rows at a time (requires libpq >= 17). Like single-row mode, but with lower per-callback overhead for large result sets. Returns 1 on success, 0 on failure. close_prepared $pg->close_prepared($name, sub { my ($result, $err) = @_; }); Closes (deallocates) a prepared statement at protocol level (requires libpq >= 17). The callback receives an empty string ("") on success. Works in pipeline mode, unlike "DEALLOCATE" SQL. close_portal $pg->close_portal($name, sub { my ($result, $err) = @_; }); Closes a portal at protocol level (requires libpq >= 17). The callback receives an empty string ("") on success. cancel my $err = $pg->cancel; Sends a cancel request using the legacy "PQcancel" API. This is a blocking call. Returns "undef" on success, an error string on failure. cancel_async $pg->cancel_async(sub { my ($err) = @_; }); Sends an asynchronous cancel request using the "PQcancelConn" API (requires libpq >= 17). The callback receives no arguments on success, or an error string on failure. Croaks if libpq was built without async cancel support ("LIBPQ_HAS_ASYNC_CANCEL"). pending_count my $n = $pg->pending_count; Returns the number of callbacks in the queue. keep_alive $pg->keep_alive(1); my $bool = $pg->keep_alive; When true, the connection's read watcher keeps "EV::run" alive even when no queries are pending. Useful when waiting for server-side "NOTIFY" events via "on_notify" — without this flag the event loop would exit after the "LISTEN" query completes. skip_pending $pg->skip_pending; Cancels all pending callbacks, invoking each with "(undef, "skipped")". PIPELINE METHODS enter_pipeline $pg->enter_pipeline; Enters pipeline mode. Queries are batched and sent without waiting for individual results. exit_pipeline $pg->exit_pipeline; Exits pipeline mode. Croaks if the pipeline is not idle (has pending queries). pipeline_sync $pg->pipeline_sync(sub { my ($ok) = @_; }); Sends a pipeline sync point. The callback fires with "(1)" when all preceding queries have completed. Alias: "sync". send_pipeline_sync $pg->send_pipeline_sync(sub { my ($ok) = @_; }); Like "pipeline_sync" but does not flush the send buffer (requires libpq >= 17). Useful for batching multiple sync points before a single manual flush via "send_flush_request". send_flush_request $pg->send_flush_request; Sends a flush request, asking the server to deliver results for queries sent so far. Alias: "flush". pipeline_status my $st = $pg->pipeline_status; Returns "PQ_PIPELINE_OFF", "PQ_PIPELINE_ON", or "PQ_PIPELINE_ABORTED". COPY METHODS put_copy_data my $ok = $pg->put_copy_data($data); Sends data to the server during a COPY IN operation. Returns 1 on success (data flushed or flush scheduled), 0 if the send buffer is full (wait for writability and retry), or -1 on error. put_copy_end my $ok = $pg->put_copy_end; my $ok = $pg->put_copy_end($errmsg); Ends a COPY IN operation. Pass an error message to abort the COPY. Returns 1 on success, 0 if the send buffer is full (retry after writability), or -1 on error. get_copy_data my $row = $pg->get_copy_data; Retrieves a row during COPY OUT. Returns the row data as a string, -1 when the COPY is complete, or "undef" if no data is available yet. HANDLER METHODS Each handler method is a getter/setter. Called with an argument, it sets the handler and returns the new value (or "undef" if cleared). Called without arguments, it returns the current handler. on_connect Called with no arguments on successful connection. on_error Called as "($error_message)" on connection-level errors. on_notify Called as "($channel, $payload, $backend_pid)" on LISTEN/NOTIFY. on_notice Called as "($message)" on server notice/warning messages. on_drain Called with no arguments when the libpq send buffer has been fully flushed during a COPY IN operation. Use this to resume sending data after "put_copy_data" returns 0 (buffer full). CONNECTION INFO String accessors ("db", "user", "host", "port", "error_message", "parameter_status", "ssl_attribute") return "undef" when not connected. Integer accessors return a default value (typically 0 or -1). Methods that require an active connection ("client_encoding", "set_client_encoding", "set_error_verbosity", "set_error_context_visibility", "conninfo") croak when not connected. error_message Last error message. Alias: "errstr". transaction_status Returns "PQTRANS_IDLE", "PQTRANS_ACTIVE", "PQTRANS_INTRANS", "PQTRANS_INERROR", or "PQTRANS_UNKNOWN". Alias: "txn_status". parameter_status my $val = $pg->parameter_status($name); Returns a server parameter (e.g. "server_version", "client_encoding"). backend_pid Backend process ID. Alias: "pid". server_version Server version as an integer (e.g. 180000 for 18.0). protocol_version Protocol version (typically 3). db Database name. user Connected user name. host Server host. hostaddr Server IP address. port Server port. socket The underlying file descriptor. ssl_in_use Returns 1 if the connection uses SSL. ssl_attribute my $val = $pg->ssl_attribute($name); Returns an SSL attribute (e.g. "protocol", "cipher"). ssl_attribute_names my $names = $pg->ssl_attribute_names; Returns an arrayref of available SSL attribute names, or "undef" if the connection does not use SSL. client_encoding Returns the current client encoding name. set_client_encoding $pg->set_client_encoding($encoding); Sets the client encoding (e.g. "UTF8", "SQL_ASCII"). This is a synchronous (blocking) call that stalls the event loop for one server round trip. Best called right after "on_connect" fires, before any queries are dispatched. Croaks if there are pending queries or on failure. set_error_verbosity my $old = $pg->set_error_verbosity($level); Sets error verbosity. Returns the previous setting. set_error_context_visibility my $old = $pg->set_error_context_visibility($level); Sets error context visibility. $level is one of "PQSHOW_CONTEXT_NEVER", "PQSHOW_CONTEXT_ERRORS" (default), or "PQSHOW_CONTEXT_ALWAYS". Returns the previous setting. error_fields my $fields = $pg->error_fields; Returns a hashref of structured error fields from the most recent "PGRES_FATAL_ERROR" result, or "undef" if no error has occurred. The value persists until the next fatal error; it is not cleared by successful queries. Keys (present only when non-NULL in the server response): sqlstate severity primary detail hint position context schema table column datatype constraint internal_position internal_query source_file source_line source_function result_meta my $meta = $pg->result_meta; Returns a hashref of metadata from the most recent query result, or "undef" if no result has been delivered. The value persists until the next result that carries metadata; it is not cleared by errors or commands that produce no columns. Keys: nfields number of columns cmd_status command status string (e.g. "SELECT 3", "INSERT 0 1") inserted_oid OID of inserted row (only present when valid) fields arrayref of column metadata hashrefs: name, type (OID), ftable (OID), ftablecol, fformat (0=text, 1=binary), fsize, fmod conninfo my $info = $pg->conninfo; Returns a hashref of the connection parameters actually used by the live connection (keyword => value pairs). connection_used_password my $bool = $pg->connection_used_password; Returns 1 if the connection authenticated with a password. connection_used_gssapi my $bool = $pg->connection_used_gssapi; Returns 1 if the connection used GSSAPI authentication. connection_needs_password my $bool = $pg->connection_needs_password; Returns 1 if the server requested a password during authentication. trace $pg->trace($filename); Enables libpq protocol tracing to the specified file. Useful for debugging wire-level issues. untrace $pg->untrace; Disables protocol tracing and closes the trace file. set_trace_flags $pg->set_trace_flags($flags); Sets trace output flags (requires libpq >= 14). $flags is a bitmask of "PQTRACE_SUPPRESS_TIMESTAMPS" and "PQTRACE_REGRESS_MODE". UTILITY METHODS escape_literal my $quoted = $pg->escape_literal($string); Returns a string literal escaped for use in SQL. Alias: "quote". escape_identifier my $quoted = $pg->escape_identifier($string); Returns an identifier escaped for use in SQL. Alias: "quote_id". escape_bytea my $escaped = $pg->escape_bytea($binary); Escapes binary data for use in a bytea column. encrypt_password my $hash = $pg->encrypt_password($password, $user); my $hash = $pg->encrypt_password($password, $user, $algorithm); Encrypts a password for use with "ALTER ROLE ... PASSWORD". $algorithm is optional; defaults to the server's "password_encryption" setting (typically "scram-sha-256"). unescape_bytea my $binary = EV::Pg->unescape_bytea($escaped); Class method. Unescapes bytea data. lib_version my $ver = EV::Pg->lib_version; Class method. Returns the libpq version as an integer. conninfo_parse my $params = EV::Pg->conninfo_parse($conninfo); Class method. Parses a connection string and returns a hashref of the recognized keyword/value pairs. Croaks if the string is invalid. Useful for validating connection strings before connecting. ALIASES Short aliases for common methods: q query qp query_params qx query_prepared prep prepare reconnect reset disconnect finish flush send_flush_request sync pipeline_sync quote escape_literal quote_id escape_identifier errstr error_message txn_status transaction_status pid backend_pid EXPORT TAGS :status PGRES_* result status constants :conn CONNECTION_OK, CONNECTION_BAD :transaction PQTRANS_* transaction status constants :pipeline PQ_PIPELINE_* pipeline status constants :verbosity PQERRORS_* verbosity constants :context PQSHOW_CONTEXT_* context visibility constants :trace PQTRACE_* trace flag constants :all all of the above BENCHMARK 500k queries over Unix socket, PostgreSQL 18, libpq 18: Workload EV::Pg sequential EV::Pg pipeline DBD::Pg sync DBD::Pg async+EV SELECT 73,109 q/s 124,092 q/s 56,496 q/s 48,744 q/s INSERT 58,534 q/s 84,467 q/s 39,068 q/s 41,559 q/s UPSERT 26,342 q/s 34,223 q/s 28,134 q/s 27,155 q/s Sequential mode uses prepared statements (parse once, bind+execute per call). Pipeline mode batches queries with "pipeline_sync" every 1000 queries. See bench/bench.pl to reproduce. REQUIREMENTS libpq >= 14 (PostgreSQL client library) and EV. Some features (chunked rows, close prepared/portal, no-flush pipeline sync, async cancel) require libpq >= 17. SEE ALSO EV, DBD::Pg, Mojo::Pg, AnyEvent::Pg LICENSE This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself. POD ERRORS Hey! The above document had some coding errors, which are explained below: Around line 461: Non-ASCII character seen before =encoding in '—'. Assuming UTF-8