diff options
| -rw-r--r-- | CHANGELOG.md | 8 | ||||
| -rw-r--r-- | connector.go | 31 | ||||
| -rw-r--r-- | go.mod | 2 | ||||
| -rw-r--r-- | go.sum | 4 | ||||
| -rw-r--r-- | handle.go | 186 | ||||
| -rw-r--r-- | interface.go | 168 | ||||
| -rw-r--r-- | results.go | 2 | ||||
| -rw-r--r-- | statement.go | 47 |
8 files changed, 371 insertions, 77 deletions
diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..18d8dd8 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,8 @@ +<!-- +SPDX-FileCopyrightText: 2026 Stefan Majewsky <majewsky@gmx.net> +SPDX-License-Identifier: Apache-2.0 +--> + +# v1.0.0 (TBD) + +Initial release. diff --git a/connector.go b/connector.go index d7f1242..1914ddb 100644 --- a/connector.go +++ b/connector.go @@ -1,13 +1,28 @@ // SPDX-FileCopyrightText: 2026 Stefan Majewsky <majewsky@gmx.net> // SPDX-License-Identifier: Apache-2.0 -// Package gg_pgx contains integration for using [gg/gsql] (and libraries based on it, such as [Oblast]) with the PostgreSQL driver library [pgx]. -// -// [gg/gsql]: https://pkg.go.dev/go.xyrillian.de/gg/gsql -// [Oblast]: https://pkg.go.dev/go.xyrillian.de/oblast -// [pgx]: https://github.com/jackc/pgx package gg_pgx -// TODO: implement Handle (handle.go is copied from an early prototype in the Oblast benchmark suite and uses the extremely old Wrap() style) -// TODO: implement ConnectionHandle -// TODO: test coverage (via gg/pgruntime) +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "go.xyrillian.de/gg/pgruntime" +) + +// SingleConnector returns a [pgruntime.Connector] that spawns individual database connections. +func SingleConnector() pgruntime.Connector[*Conn] { + return func(ctx context.Context, dbURL string) (*Conn, error) { + conn, err := pgx.Connect(ctx, dbURL) + return maybe(NewConn, conn), err + } +} + +// PoolConnector returns a [pgruntime.Connector] that spawns database connection pools. +func PoolConnector() pgruntime.Connector[*Pool] { + return func(ctx context.Context, dbURL string) (*Pool, error) { + pool, err := pgxpool.New(ctx, dbURL) + return maybe(NewPool, pool), err + } +} @@ -4,7 +4,7 @@ go 1.26 require ( github.com/jackc/pgx/v5 v5.10.0 - go.xyrillian.de/gg v1.11.2-0.20260730175754-81e32cec3629 + go.xyrillian.de/gg v1.11.2-0.20260730205329-a9c88a040a3c ) require ( @@ -16,8 +16,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -go.xyrillian.de/gg v1.11.2-0.20260730175754-81e32cec3629 h1:3IOrqdh3N/X4FLts7prxJ4Sy1IGZsOHl7NOym4hTMdk= -go.xyrillian.de/gg v1.11.2-0.20260730175754-81e32cec3629/go.mod h1:DoO4fQSWIrBRlNlCjVyrYM0kAEBt/Jg2GkMH+cGRZ0k= +go.xyrillian.de/gg v1.11.2-0.20260730205329-a9c88a040a3c h1:jobZ2OZBFxZ3VuXrm3LPzfJioVxBMs1JdQuHslT/Wuw= +go.xyrillian.de/gg v1.11.2-0.20260730205329-a9c88a040a3c/go.mod h1:DoO4fQSWIrBRlNlCjVyrYM0kAEBt/Jg2GkMH+cGRZ0k= golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= @@ -5,85 +5,173 @@ package gg_pgx import ( "context" - "fmt" "strconv" "sync/atomic" "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" + "go.xyrillian.de/gg/errext" "go.xyrillian.de/gg/gsql" ) -type Handle interface { - Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) - Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) - QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +type connHandle struct { + inner *pgx.Conn } -var ( - _ Handle = &pgx.Conn{} - _ Handle = &pgxpool.Conn{} - _ Handle = pgx.Tx(&pgxpool.Tx{}) -) +// GSQLPrepare implements the [gsql.Handle] interface. +func (h connHandle) GSQLPrepare(ctx context.Context, query string, repeated bool) (gsql.Statement, error) { + if !repeated { + return wrappedUnpreparedStatement{query, h.inner}, nil + } + name := getPreparedStatementName() + stmt, err := h.inner.Prepare(ctx, name, query) + return wrappedPreparedStatement{ctx, stmt, h.inner, h.inner.Deallocate}, err +} -func Wrap(h Handle) gsql.Handle { - switch h := h.(type) { - case *pgx.Conn: - return wrappedHandle{h} - case *pgxpool.Conn: - return wrappedHandle{h} - case pgx.Tx: - return wrappedHandle{h} - default: - panic(fmt.Sprintf("unexpected type: %#v", h)) +// GSQLQuery implements the [gsql.Handle] interface. +func (h connHandle) GSQLQuery(ctx context.Context, query string, args []any) (gsql.Rows, error) { + rows, err := h.inner.Query(ctx, query, args...) + return wrappedRows{rows}, err +} + +// GSQLClose implements the [gsql.ConnectionHandle] interface. +func (h connHandle) GSQLClose(ctx context.Context) error { + return h.inner.Close(ctx) +} + +// GSQLTransact implements the [gsql.ConnectionHandle] interface. +func (h connHandle) GSQLTransact(ctx context.Context, action func(tx gsql.Handle) error) error { + tx, err := h.inner.Begin(ctx) + if err != nil { + return err } + return transact(ctx, tx, action) } -var preparedStatementId atomic.Uint64 +//////////////////////////////////////////////////////////////////////////////// -type wrappedHandle struct { - inner Handle +type poolHandle struct { + inner *pgxpool.Pool } // GSQLPrepare implements the [gsql.Handle] interface. -func (h wrappedHandle) GSQLPrepare(ctx context.Context, query string, repeated bool) (gsql.Statement, error) { +func (h poolHandle) GSQLPrepare(ctx context.Context, query string, repeated bool) (gsql.Statement, error) { if !repeated { return wrappedUnpreparedStatement{query, h.inner}, nil } + name := getPreparedStatementName() + + // while we have a prepared statement, we also need to hold an acquired connection + conn, err := h.inner.Acquire(ctx) + if err != nil { + return nil, err + } + stmt, err := conn.Conn().Prepare(ctx, name, query) + if err != nil { + conn.Release() + return nil, err + } + deallocate := func(ctx context.Context, name string) error { + err := conn.Conn().Deallocate(ctx, name) + conn.Release() + return err + } + return wrappedPreparedStatement{ctx, stmt, h.inner, deallocate}, nil +} + +// GSQLQuery implements the [gsql.Handle] interface. +func (h poolHandle) GSQLQuery(ctx context.Context, query string, args []any) (gsql.Rows, error) { + rows, err := h.inner.Query(ctx, query, args...) + return wrappedRows{rows}, err +} + +// GSQLClose implements the [gsql.ConnectionHandle] interface. +func (h poolHandle) GSQLClose(ctx context.Context) error { + h.inner.Close() + return nil +} + +// GSQLTransact implements the [gsql.ConnectionHandle] interface. +func (h poolHandle) GSQLTransact(ctx context.Context, action func(tx gsql.Handle) error) error { + tx, err := h.inner.Begin(ctx) + if err != nil { + return err + } + return transact(ctx, tx, action) +} - name := "oblast_pgx_" + strconv.FormatUint(preparedStatementId.Add(1), 10) - switch inner := h.inner.(type) { - case *pgx.Conn: - stmt, err := inner.Prepare(ctx, name, query) - return wrappedPreparedStatement{ctx, stmt, h.inner}, err - case *pgxpool.Conn: - // pgxpool.Conn does not have Prepare() +//////////////////////////////////////////////////////////////////////////////// + +type poolConnHandle struct { + inner *pgxpool.Conn +} + +// GSQLPrepare implements the [gsql.Handle] interface. +func (h poolConnHandle) GSQLPrepare(ctx context.Context, query string, repeated bool) (gsql.Statement, error) { + if !repeated { return wrappedUnpreparedStatement{query, h.inner}, nil - case pgx.Tx: - stmt, err := inner.Conn().Prepare(ctx, name, query) - return wrappedPreparedStatement{ctx, stmt, h.inner}, err - default: - panic("unreachable") // because of the check in func Wrap() } + name := getPreparedStatementName() + stmt, err := h.inner.Conn().Prepare(ctx, name, query) + return wrappedPreparedStatement{ctx, stmt, h.inner, h.inner.Conn().Deallocate}, err } -// Releases a prepared statement. -func deallocate(ctx context.Context, h Handle, stmt *pgconn.StatementDescription) error { - switch h := h.(type) { - case *pgx.Conn: - return h.Deallocate(ctx, stmt.Name) - case *pgxpool.Conn: - panic("unreachable") // because func GSQLPrepare() does not return a wrappedPreparedStatement for this underlying type - case pgx.Tx: - return h.Conn().Deallocate(ctx, stmt.Name) - default: - panic("unreachable") // because of the check in func Wrap() +// GSQLQuery implements the [gsql.Handle] interface. +func (h poolConnHandle) GSQLQuery(ctx context.Context, query string, args []any) (gsql.Rows, error) { + rows, err := h.inner.Query(ctx, query, args...) + return wrappedRows{rows}, err +} + +// GSQLClose implements the [gsql.ConnectionHandle] interface. +func (h poolConnHandle) GSQLClose(ctx context.Context) error { + h.inner.Release() + return nil +} + +// GSQLTransact implements the [gsql.ConnectionHandle] interface. +func (h poolConnHandle) GSQLTransact(ctx context.Context, action func(tx gsql.Handle) error) error { + tx, err := h.inner.Begin(ctx) + if err != nil { + return err } + return transact(ctx, tx, action) +} + +//////////////////////////////////////////////////////////////////////////////// + +type txHandle struct { + inner pgx.Tx +} + +// GSQLPrepare implements the [gsql.Handle] interface. +func (h txHandle) GSQLPrepare(ctx context.Context, query string, repeated bool) (gsql.Statement, error) { + if !repeated { + return wrappedUnpreparedStatement{query, h.inner}, nil + } + name := getPreparedStatementName() + stmt, err := h.inner.Conn().Prepare(ctx, name, query) + return wrappedPreparedStatement{ctx, stmt, h.inner, h.inner.Conn().Deallocate}, err } // GSQLQuery implements the [gsql.Handle] interface. -func (h wrappedHandle) GSQLQuery(ctx context.Context, query string, args []any) (gsql.Rows, error) { +func (h txHandle) GSQLQuery(ctx context.Context, query string, args []any) (gsql.Rows, error) { rows, err := h.inner.Query(ctx, query, args...) return wrappedRows{rows}, err } + +//////////////////////////////////////////////////////////////////////////////// + +var preparedStatementId atomic.Uint64 + +func getPreparedStatementName() string { + return "oblast_pgx_" + strconv.FormatUint(preparedStatementId.Add(1), 10) +} + +func transact(ctx context.Context, tx pgx.Tx, action func(tx gsql.Handle) error) error { + err := action(NewTx(tx)) + if err == nil { + return errext.WithCleanup(nil, "tx.Commit", tx.Commit(ctx)) + } else { + return errext.WithCleanup(err, "tx.Rollback", tx.Rollback(ctx)) + } +} diff --git a/interface.go b/interface.go new file mode 100644 index 0000000..d4a5b56 --- /dev/null +++ b/interface.go @@ -0,0 +1,168 @@ +// SPDX-FileCopyrightText: 2026 Stefan Majewsky <majewsky@gmx.net> +// SPDX-License-Identifier: Apache-2.0 + +// Package gg_pgx contains integration for using [gg/gsql] (and libraries based on it, such as [Oblast]) with the PostgreSQL driver library [pgx]. +// +// [gg/gsql]: https://pkg.go.dev/go.xyrillian.de/gg/gsql +// [Oblast]: https://pkg.go.dev/go.xyrillian.de/oblast +// [pgx]: https://github.com/jackc/pgx +package gg_pgx + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "go.xyrillian.de/gg/gsql" +) + +// TODO: test coverage (via gg/pgruntime) + +// NOTE: The internal structure of these types follows the pattern established +// by (and explained in) `std.go` of `go.xyrillian.de/gg/gsql`. + +// Conn wraps [*pgx.Conn] into a [gsql.Handle]. +// +// Because this type has [*pgx.Conn] as an embedded field, +// all methods from that type work on this type as well. +type Conn struct { + *pgx.Conn + gsql.ConnectionHandle +} + +// NewConn wraps an instance of [*pgx.Conn] into the [Conn] type that implements [gsql.Handle]. +func NewConn(conn *pgx.Conn) *Conn { + return &Conn{conn, connHandle{conn}} +} + +// Begin is like [pgx.Conn.Begin], but wraps the resulting transaction into a [Handle]. +func (conn *Conn) Begin(ctx context.Context) (*Tx, error) { + tx, err := conn.Conn.Begin(ctx) + return maybeNewTx(tx), err +} + +// BeginTx is like [pgx.Conn.BeginTx], but wraps the resulting transaction into a [Handle]. +func (conn *Conn) BeginTx(ctx context.Context, opts pgx.TxOptions) (*Tx, error) { + tx, err := conn.Conn.BeginTx(ctx, opts) + return maybeNewTx(tx), err +} + +// Pool wraps [*pgxpool.Pool] into a [gsql.Handle]. +// +// Because this type has [*pgxpool.Pool] as an embedded field, +// all methods from that type work on this type as well. +type Pool struct { + *pgxpool.Pool + gsql.ConnectionHandle +} + +// NewPool wraps an instance of [*pgxpool.Pool] into the [Pool] type that implements [gsql.Handle]. +func NewPool(pool *pgxpool.Pool) *Pool { + return &Pool{pool, poolHandle{pool}} +} + +// Acquire is like [pgxpool.Pool.Acquire], but wraps the resulting connection into a [Handle]. +func (pool *Pool) Acquire(ctx context.Context) (*PoolConn, error) { + conn, err := pool.Pool.Acquire(ctx) + return maybe(NewPoolConn, conn), err +} + +// AcquireAllIdle is like [pgxpool.Pool.AcquireAllIdle], but wraps the resulting connections into [Handle] instances. +func (pool *Pool) AcquireAllIdle(ctx context.Context) []*PoolConn { + conns := pool.Pool.AcquireAllIdle(ctx) + result := make([]*PoolConn, len(conns)) + for idx, conn := range conns { + result[idx] = maybe(NewPoolConn, conn) + } + return result +} + +// AcquireFunc is like [pgxpool.Pool.AcquireFunc], but wraps the resulting connection into a [Handle]. +func (pool *Pool) AcquireFunc(ctx context.Context, f func(*PoolConn) error) error { + return pool.Pool.AcquireFunc(ctx, func(conn *pgxpool.Conn) error { + return f(maybe(NewPoolConn, conn)) + }) +} + +// Begin is like [pgxpool.Pool.Begin], but wraps the resulting transaction into a [Handle]. +func (pool *Pool) Begin(ctx context.Context) (*Tx, error) { + tx, err := pool.Pool.Begin(ctx) + return maybeNewTx(tx), err +} + +// BeginTx is like [pgxpool.Pool.BeginTx], but wraps the resulting transaction into a [Handle]. +func (pool *Pool) BeginTx(ctx context.Context, opts pgx.TxOptions) (*Tx, error) { + tx, err := pool.Pool.BeginTx(ctx, opts) + return maybeNewTx(tx), err +} + +// PoolConn wraps [*pgxpool.Conn] into a [gsql.Handle]. +// +// Because this type has [*pgxpool.Conn] as an embedded field, +// all methods from that type work on this type as well. +type PoolConn struct { + *pgxpool.Conn + gsql.ConnectionHandle +} + +// NewPoolConn wraps an instance of [*pgxpool.Conn] into the [PoolConn] type that implements [gsql.Handle]. +func NewPoolConn(pool *pgxpool.Conn) *PoolConn { + return &PoolConn{pool, poolConnHandle{pool}} +} + +// Begin is like [pgxpool.Conn.Begin], but wraps the resulting transaction into a [Handle]. +func (conn *PoolConn) Begin(ctx context.Context) (*Tx, error) { + tx, err := conn.Conn.Begin(ctx) + return maybeNewTx(tx), err +} + +// BeginTx is like [pgxpool.Conn.BeginTx], but wraps the resulting transaction into a [Handle]. +func (conn *PoolConn) BeginTx(ctx context.Context, opts pgx.TxOptions) (*Tx, error) { + tx, err := conn.Conn.BeginTx(ctx, opts) + return maybeNewTx(tx), err +} + +// GetConn is like [pgxpool.Conn.Conn], but wraps the resulting connection into a [Handle]. +// +// This method should be called "Conn", but one of the embedded fields of this type blocks that name. +func (conn *PoolConn) GetConn() *Conn { + return maybe(NewConn, conn.Conn.Conn()) +} + +// Hijack is like [pgxpool.Conn.Conn], but wraps the resulting connection into a [Handle]. +func (conn *PoolConn) Hijack() *Conn { + return maybe(NewConn, conn.Conn.Hijack()) +} + +// Tx wraps [pgx.Tx] into a [gsql.Handle]. +// +// Because this type has [pgx.Tx] as an embedded field, +// all methods from that type work on this type as well. +type Tx struct { + pgx.Tx + gsql.Handle +} + +// NewTx wraps an instance of [pgx.Tx] into the [Tx] type that implements [Handle]. +func NewTx(tx pgx.Tx) *Tx { + return &Tx{tx, txHandle{tx}} +} + +// Conn is like the Conn() method of [pgx.Tx], but wraps the resulting connection into a [Handle]. +func (t *Tx) Conn() *Conn { + return maybe(NewConn, t.Tx.Conn()) +} + +func maybeNewTx(tx pgx.Tx) *Tx { + if tx == nil { + return nil + } + return NewTx(tx) +} + +func maybe[T, U any](wrap func(*T) *U, value *T) *U { + if value == nil { + return nil + } + return wrap(value) +} @@ -31,7 +31,7 @@ func (r wrappedRows) Columns() ([]string, error) { // Close implements the [gsql.Rows] interface. func (r wrappedRows) Close() error { r.inner.Close() - return nil + return r.inner.Err() } // Err implements the [gsql.Rows] interface. diff --git a/statement.go b/statement.go index bb95252..1cda044 100644 --- a/statement.go +++ b/statement.go @@ -7,19 +7,34 @@ import ( "context" "database/sql" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" "go.xyrillian.de/gg/gsql" ) +type pgxExecutor interface { + Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) // TODO: remove after splitting Handle types + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +var ( + _ pgxExecutor = &pgx.Conn{} + _ pgxExecutor = &pgxpool.Conn{} + _ pgxExecutor = pgx.Tx(&pgxpool.Tx{}) +) + type wrappedPreparedStatement struct { - ctx context.Context - statement *pgconn.StatementDescription - handle Handle + ctx context.Context + statement *pgconn.StatementDescription + executor pgxExecutor + deallocate func(ctx context.Context, name string) error } type wrappedUnpreparedStatement struct { - query string - handle Handle + query string + executor pgxExecutor } var ( @@ -27,34 +42,34 @@ var ( _ gsql.Statement = wrappedUnpreparedStatement{} ) -// Close implements the [handle.Statement] interface. +// Close implements the [gsql.Statement] interface. func (s wrappedPreparedStatement) Close() error { - return deallocate(s.ctx, s.handle, s.statement) + return s.deallocate(s.ctx, s.statement.Name) } -// Close implements the [handle.Statement] interface. +// Close implements the [gsql.Statement] interface. func (s wrappedUnpreparedStatement) Close() error { return nil } -// Exec implements the [handle.Statement] interface. +// Exec implements the [gsql.Statement] interface. func (s wrappedPreparedStatement) Exec(ctx context.Context, args []any) (sql.Result, error) { - result, err := s.handle.Exec(ctx, s.statement.Name, args...) + result, err := s.executor.Exec(ctx, s.statement.Name, args...) return wrappedResult{result}, err } -// Exec implements the [handle.Statement] interface. +// Exec implements the [gsql.Statement] interface. func (s wrappedUnpreparedStatement) Exec(ctx context.Context, args []any) (sql.Result, error) { - result, err := s.handle.Exec(ctx, s.query, args...) + result, err := s.executor.Exec(ctx, s.query, args...) return wrappedResult{result}, err } -// QueryRow implements the [handle.Statement] interface. +// QueryRow implements the [gsql.Statement] interface. func (s wrappedPreparedStatement) QueryRow(ctx context.Context, args, slots []any) error { - return s.handle.QueryRow(ctx, s.statement.Name, args...).Scan(slots...) + return s.executor.QueryRow(ctx, s.statement.Name, args...).Scan(slots...) } -// QueryRow implements the [handle.Statement] interface. +// QueryRow implements the [gsql.Statement] interface. func (s wrappedUnpreparedStatement) QueryRow(ctx context.Context, args, slots []any) error { - return s.handle.QueryRow(ctx, s.query, args...).Scan(slots...) + return s.executor.QueryRow(ctx, s.query, args...).Scan(slots...) } |