将 Go (pgx) 与 ClickHouse Managed Postgres 结合使用

CREATE TABLE todos (
    id         bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    title      text NOT NULL,
    done       boolean NOT NULL DEFAULT false,
    created_at timestamptz NOT NULL DEFAULT now()
);

DROP TABLE todos;

package main

import (
	"context"
	"fmt"
	"log"
	"os"

	"github.com/jackc/pgx/v5/pgxpool"
)

// newPool creates a connection pool from DATABASE_URL and verifies that it can connect.
func newPool(ctx context.Context) (*pgxpool.Pool, error) {
	pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		return nil, err
	}

	var tlsVersion string
	err = pool.QueryRow(ctx,
		"SELECT coalesce(version, 'none') FROM pg_stat_ssl WHERE pid = pg_backend_pid()",
	).Scan(&tlsVersion)
	if err != nil {
		pool.Close()
		return nil, fmt.Errorf("connect to Postgres: %w", err)
	}
	log.Printf("Connected to Postgres (TLS: %s)", tlsVersion)
	return pool, nil
}

package main

import (
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"strconv"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

type Todo struct {
	ID        int64     `json:"id" db:"id"`
	Title     string    `json:"title" db:"title"`
	Done      bool      `json:"done" db:"done"`
	CreatedAt time.Time `json:"created_at" db:"created_at"`
}

const todoColumns = "id, title, done, created_at"

// registerTodoRoutes adds the CRUD endpoints for todos to mux.
func registerTodoRoutes(mux *http.ServeMux, pool *pgxpool.Pool) {
	// Read
	mux.HandleFunc("GET /todos", func(w http.ResponseWriter, r *http.Request) {
		rows, _ := pool.Query(r.Context(), "SELECT "+todoColumns+" FROM todos ORDER BY id")
		todos, err := pgx.CollectRows(rows, pgx.RowToStructByName[Todo])
		writeJSON(w, http.StatusOK, todos, err)
	})

	// Create
	mux.HandleFunc("POST /todos", func(w http.ResponseWriter, r *http.Request) {
		var in struct{ Title string }
		if err := json.NewDecoder(r.Body).Decode(&in); err != nil || in.Title == "" {
			http.Error(w, "expected a JSON body with a title", http.StatusBadRequest)
			return
		}
		rows, _ := pool.Query(r.Context(),
			"INSERT INTO todos (title) VALUES ($1) RETURNING "+todoColumns, in.Title)
		todo, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Todo])
		writeJSON(w, http.StatusCreated, todo, err)
	})

	// Update: toggle the done flag
	mux.HandleFunc("PATCH /todos/{id}", func(w http.ResponseWriter, r *http.Request) {
		id, _ := strconv.ParseInt(r.PathValue("id"), 10, 64)
		rows, _ := pool.Query(r.Context(),
			"UPDATE todos SET done = NOT done WHERE id = $1 RETURNING "+todoColumns, id)
		todo, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Todo])
		writeJSON(w, http.StatusOK, todo, err)
	})

	// Delete
	mux.HandleFunc("DELETE /todos/{id}", func(w http.ResponseWriter, r *http.Request) {
		id, _ := strconv.ParseInt(r.PathValue("id"), 10, 64)
		tag, err := pool.Exec(r.Context(), "DELETE FROM todos WHERE id = $1", id)
		if err == nil && tag.RowsAffected() == 0 {
			err = pgx.ErrNoRows
		}
		writeJSON(w, http.StatusNoContent, nil, err)
	})
}

func writeJSON(w http.ResponseWriter, status int, body any, err error) {
	switch {
	case errors.Is(err, pgx.ErrNoRows):
		http.Error(w, "not found", http.StatusNotFound)
	case err != nil:
		log.Println(err)
		http.Error(w, "database error", http.StatusInternalServerError)
	case status == http.StatusNoContent:
		w.WriteHeader(status)
	default:
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(status)
		enc := json.NewEncoder(w)
		enc.SetIndent("", "  ")
		enc.Encode(body)
	}
}

package main

import (
	"context"
	"errors"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
)

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	pool, err := newPool(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer pool.Close()

	mux := http.NewServeMux()
	registerTodoRoutes(mux, pool)
	srv := &http.Server{Addr: ":3114", Handler: mux}

	go func() {
		<-ctx.Done()
		srv.Shutdown(context.Background())
	}()

	log.Println("Listening on http://localhost:3114/todos")
	if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
		log.Fatal(err)
	}
	log.Println("Shutting down")
}