class CQL::Postgres::AsyncEngine

Overview

Runs CQL statements over the native wire driver, with a connection pool. It is an Executor, so any statement can be pointed at it:

engine = CQL::Postgres::AsyncEngine.connect("postgres://app:secret@db/app?pool_size=10")
User.query.where(active: true).using(engine).to_a
engine.transaction { |tx| User.query.where(name: "Ada").using(tx).to_a }
engine.exec("INSERT INTO notes (body) VALUES ($1)", ["first".as(DB::Any)])

using(engine) changes where a statement runs, not its SQL, so the statement's schema must use a PostgreSQL or CockroachDB adapter.

Statements run through CQL::Execution, so telemetry and performance tracking see them like any other.

Included Modules

Defined in:

postgres/async_engine.cr

Constructors

Instance Method Summary

Instance methods inherited from module CQL::Postgres::QueryStreaming

stream(query : String, args : Array(DB::Any) = [] of ::DB::Any, fetch_size : Int32 = 1000, timeout : Time::Span | Nil = nil, & : DB::ResultSet -> T) : T forall T
stream(query : Q, fetch_size : Int32 = 1000, timeout : Time::Span | Nil = nil, & : DB::ResultSet -> T) : T forall Q, T stream

Instance methods inherited from module CQL::Executor

adapter : Adapter adapter, exec_query(& : DB::Connection -> _) exec_query

Constructor Detail

def self.connect(url : String, adapter : CQL::Adapter | Nil = nil, setup : Proc(Connection, Nil) | Nil = nil, reset : String | Nil = nil) : AsyncEngine #

setup runs once on every new pooled connection (e.g. SET search_path); reset is a statement run each time a connection returns to the pool (e.g. DISCARD ALL). Pool sizing and lifetimes come from the URL: pool_size, pool_min_size, max_queries, max_idle_lifetime. The adapter defaults from the URL: cockroachdb:// is CockroachDB, anything else PostgreSQL.


[View source]
def self.new(pool : Pool, adapter : CQL::Adapter = CQL::Adapter::Postgres) #

[View source]

Instance Method Detail

def adapter : CQL::Adapter #

[View source]
def close : Nil #

[View source]
def exec(sql : String, args : Array(DB::Any) = [] of ::DB::Any) : DB::ExecResult #

Runs raw SQL (no rows wanted) as a telemetry-observed statement. Without arguments the text may be a script of several statements.


[View source]
def exec_query(& : DB::Connection -> _) #

[View source]
def pipeline(statements : Array(Tuple(String, Array(DB::Any)))) : Array(PipelineResult | ServerError) #

[View source]
def pipeline(queries : Array(Q)) : Array(PipelineResult | ServerError) forall Q #

Sends several queries in one network round trip (see Session#pipeline). Each query is observed by telemetry like any other (the first one's duration covers the whole round trip); one failing leaves the others unaffected and appears as its ServerError in the returned array.

engine = CQL::Postgres::AsyncEngine.connect(ENV["DATABASE_URL"])
users, posts = engine.pipeline([User.query.where(active: true), Post.query])
users.as(CQL::Postgres::PipelineResult).all(as: User)

[View source]
def pool : Pool #

[View source]
def transaction(isolation : IsolationLevel | Nil = nil, read_only : Bool = false, deferrable : Bool = false, synchronous_commit : Bool | Nil = nil, & : TransactionScope -> T) : T | Nil forall T #

Runs the block inside BEGIN/COMMIT on one connection; it is rolled back when the block raises (raise DB::Rollback to roll back without an error, which makes this return nil). The yielded scope pins statements to that connection (query.using(tx)) and offers tx.savepoint { ... }.

engine = CQL::Postgres::AsyncEngine.connect(ENV["DATABASE_URL"])
engine.transaction(isolation: :serializable, read_only: true, deferrable: true) { |tx| User.query.using(tx).to_a }
engine.transaction(synchronous_commit: false) do |tx| # SET LOCAL: this commit may be lost in a crash
  tx.savepoint { |inner| User.query.where(name: "Ada").using(inner).to_a }
end

synchronous_commit: false makes the commit return before the WAL is flushed: the database stays consistent, but the most recent commits (about three wal_writer_delays' worth) can vanish if the server crashes. true forces on (overriding e.g. remote_apply); the default follows the server's setting.


[View source]
def with_connection(& : Connection -> T) : T forall T #

One pooled connection for the block, returned afterwards.


[View source]