Database Access in C++

TL;DR

SQLite is ideal for embedded/local storage. PostgreSQL via pqxx (C++ wrapper around libpq) suits server applications. Always use prepared statements — never build SQL with string concatenation.

SQLite with C API

cpp

Godbolt

#include <sqlite3.h>

sqlite3* db;
sqlite3_open("app.db", &db);

// Create table
sqlite3_exec(db,
    "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)",
    nullptr, nullptr, nullptr);

// Prepared statement — parameterized query
sqlite3_stmt* stmt;
sqlite3_prepare_v2(db,
    "INSERT INTO users (name) VALUES (?)", -1, &stmt, nullptr);

sqlite3_bind_text(stmt, 1, "Alice", -1, SQLITE_STATIC);
sqlite3_step(stmt);
sqlite3_finalize(stmt);

// Query
sqlite3_prepare_v2(db, "SELECT id, name FROM users", -1, &stmt, nullptr);
while (sqlite3_step(stmt) == SQLITE_ROW) {
    int id = sqlite3_column_int(stmt, 0);
    const char* name = reinterpret_cast<const char*>(sqlite3_column_text(stmt, 1));
    std::println("{}: {}", id, name);
}
sqlite3_finalize(stmt);
sqlite3_close(db);

SQLite with SQLiteCpp (RAII wrapper)

cpp

Godbolt

#include <SQLiteCpp/SQLiteCpp.h>

// RAII — no manual open/close
SQLite::Database db{"app.db", SQLite::OPEN_READWRITE | SQLite::OPEN_CREATE};

db.exec("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)");

// Prepared statement
SQLite::Statement insert{db, "INSERT INTO users (name) VALUES (?)"};
insert.bind(1, "Alice");
insert.exec();

// Query with structured binding
SQLite::Statement query{db, "SELECT id, name FROM users WHERE id > ?"};
query.bind(1, 0);
while (query.executeStep()) {
    int id         = query.getColumn(0).getInt();
    std::string name = query.getColumn(1).getString();
    std::println("{}: {}", id, name);
}

// Transaction
SQLite::Transaction txn{db};
for (const auto& name : names) {
    SQLite::Statement ins{db, "INSERT INTO users (name) VALUES (?)"};
    ins.bind(1, name);
    ins.exec();
}
txn.commit();

PostgreSQL with pqxx

cpp

Godbolt

#include <pqxx/pqxx>

pqxx::connection conn{
    "host=localhost dbname=mydb user=postgres password=secret"
};

// Non-transactional query (read-only, auto-commit)
pqxx::nontransaction ntx{conn};
pqxx::result r = ntx.exec("SELECT version()");
std::println("{}", r[0][0].c_str());

// Transaction
pqxx::work txn{conn};

// Prepared statement — register once, call many times
conn.prepare("insert_user",
    "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id");

auto result = txn.exec_prepared("insert_user", "Alice", "alice@example.com");
int new_id = result[0][0].as<int>();

// Parameterized query (one-shot, no need to prepare)
auto rows = txn.exec_params(
    "SELECT id, name FROM users WHERE age > $1 AND active = $2",
    18, true);

for (const auto& row : rows) {
    auto id   = row["id"].as<int>();
    auto name = row["name"].as<std::string>();
    std::println("{}: {}", id, name);
}

txn.commit();

Connection Pooling

cpp

Godbolt

#include <pqxx/pqxx>
#include <queue>
#include <mutex>

class ConnectionPool {
    std::queue<std::unique_ptr<pqxx::connection>> pool_;
    std::mutex mtx_;
    std::string conn_str_;
    size_t max_size_;

public:
    ConnectionPool(std::string cs, size_t size)
        : conn_str_{std::move(cs)}, max_size_{size}
    {
        for (size_t i = 0; i < size; ++i)
            pool_.push(std::make_unique<pqxx::connection>(conn_str_));
    }

    // RAII connection handle
    class Handle {
        ConnectionPool* pool_;
        std::unique_ptr<pqxx::connection> conn_;
    public:
        Handle(ConnectionPool* p, std::unique_ptr<pqxx::connection> c)
            : pool_{p}, conn_{std::move(c)} {}
        ~Handle() { pool_->release(std::move(conn_)); }
        pqxx::connection& get() { return *conn_; }
        pqxx::connection* operator->() { return conn_.get(); }
    };

    Handle acquire() {
        std::lock_guard lock{mtx_};
        if (pool_.empty())
            return {this, std::make_unique<pqxx::connection>(conn_str_)};
        auto conn = std::move(pool_.front());
        pool_.pop();
        return {this, std::move(conn)};
    }

    void release(std::unique_ptr<pqxx::connection> conn) {
        std::lock_guard lock{mtx_};
        if (pool_.size() < max_size_)
            pool_.push(std::move(conn));
    }
};

// Usage
ConnectionPool pool{"dbname=mydb", 10};
{
    auto conn = pool.acquire();
    pqxx::work txn{conn.get()};
    txn.exec("UPDATE counters SET val = val + 1");
    txn.commit();
}  // connection returned to pool

Simple ORM Pattern

cpp

Godbolt

struct User {
    int id = 0;
    std::string name;
    std::string email;
    bool active = true;
};

class UserRepository {
    pqxx::connection& conn_;

public:
    explicit UserRepository(pqxx::connection& c) : conn_{c} {}

    std::optional<User> find_by_id(int id) {
        pqxx::nontransaction ntx{conn_};
        auto r = ntx.exec_params(
            "SELECT id, name, email, active FROM users WHERE id = $1", id);
        if (r.empty()) return std::nullopt;
        return User{r[0]["id"].as<int>(),
                    r[0]["name"].as<std::string>(),
                    r[0]["email"].as<std::string>(),
                    r[0]["active"].as<bool>()};
    }

    User save(User user) {
        pqxx::work txn{conn_};
        if (user.id == 0) {
            auto r = txn.exec_params(
                "INSERT INTO users (name, email, active) "
                "VALUES ($1, $2, $3) RETURNING id",
                user.name, user.email, user.active);
            user.id = r[0][0].as<int>();
        } else {
            txn.exec_params(
                "UPDATE users SET name=$1, email=$2, active=$3 WHERE id=$4",
                user.name, user.email, user.active, user.id);
        }
        txn.commit();
        return user;
    }

    std::vector<User> find_all_active() {
        pqxx::nontransaction ntx{conn_};
        auto r = ntx.exec("SELECT id, name, email, active FROM users WHERE active");
        std::vector<User> users;
        for (const auto& row : r)
            users.push_back({row[0].as<int>(), row[1].as<std::string>(),
                             row[2].as<std::string>(), row[3].as<bool>()});
        return users;
    }
};

Error Handling

cpp

Godbolt

try {
    pqxx::work txn{conn};
    txn.exec("INSERT INTO users VALUES (1, 'Alice')");
    txn.commit();
} catch (const pqxx::unique_violation& e) {
    std::println("duplicate key: {}", e.what());
} catch (const pqxx::foreign_key_violation& e) {
    std::println("FK violation: {}", e.what());
} catch (const pqxx::sql_error& e) {
    std::println("SQL error: {}, query: {}", e.what(), e.query());
} catch (const pqxx::broken_connection& e) {
    std::println("connection lost: {}", e.what());
    // reconnect and retry
}

CMake Setup

cmake

# SQLite
find_package(SQLite3 REQUIRED)
target_link_libraries(myapp SQLite::SQLite3)

# SQLiteCpp (via FetchContent)
FetchContent_Declare(SQLiteCpp
    GIT_REPOSITORY https://github.com/SRombauts/SQLiteCpp.git
    GIT_TAG 3.3.1)
FetchContent_MakeAvailable(SQLiteCpp)
target_link_libraries(myapp SQLiteCpp)

# pqxx (PostgreSQL)
find_package(PostgreSQL REQUIRED)
find_package(pqxx REQUIRED)
target_link_libraries(myapp pqxx::pqxx)

A fixed-size connection pool eliminates per-request handshake overhead and bounds the number of simultaneous backend connections. The pattern below adds lazy initialization and health-check reconnection that the earlier minimal example omits.

cpp

Godbolt

#include <pqxx/pqxx>
#include <condition_variable>
#include <mutex>
#include <vector>
#include <chrono>

class PgPool {
public:
    struct Config {
        std::string dsn;
        size_t      min_size  = 2;   // connections created at startup
        size_t      max_size  = 10;  // hard ceiling
        std::chrono::seconds acquire_timeout{5};
    };

    explicit PgPool(Config cfg) : cfg_{std::move(cfg)} {
        // Lazy: only open min_size connections eagerly
        for (size_t i = 0; i < cfg_.min_size; ++i)
            idle_.push_back(make_conn());
    }

    // RAII handle — returns connection on destruction
    class Lease {
        PgPool*                             pool_;
        std::unique_ptr<pqxx::connection>   conn_;
    public:
        Lease(PgPool* p, std::unique_ptr<pqxx::connection> c)
            : pool_{p}, conn_{std::move(c)} {}
        ~Lease() { if (pool_) pool_->checkin(std::move(conn_)); }
        Lease(Lease&&) = default;
        Lease& operator=(Lease&&) = default;
        pqxx::connection& get() { return *conn_; }
        pqxx::connection* operator->() { return conn_.get(); }
    };

    Lease checkout() {
        std::unique_lock lock{mu_};
        bool ok = cv_.wait_for(lock, cfg_.acquire_timeout, [&] {
            return !idle_.empty() || active_ < cfg_.max_size;
        });
        if (!ok) throw std::runtime_error("connection pool exhausted");

        std::unique_ptr<pqxx::connection> conn;
        if (!idle_.empty()) {
            conn = std::move(idle_.back());
            idle_.pop_back();
            // Health check: try a trivial query
            if (!is_alive(*conn))
                conn = make_conn();
        } else {
            conn = make_conn();
        }
        ++active_;
        return {this, std::move(conn)};
    }

private:
    Config                                          cfg_;
    std::mutex                                      mu_;
    std::condition_variable                         cv_;
    std::vector<std::unique_ptr<pqxx::connection>>  idle_;
    size_t                                          active_ = 0;

    std::unique_ptr<pqxx::connection> make_conn() {
        return std::make_unique<pqxx::connection>(cfg_.dsn);
    }

    static bool is_alive(pqxx::connection& c) {
        try {
            pqxx::nontransaction n{c};
            n.exec("SELECT 1");
            return true;
        } catch (...) { return false; }
    }

    void checkin(std::unique_ptr<pqxx::connection> conn) {
        std::lock_guard lock{mu_};
        --active_;
        if (idle_.size() < cfg_.max_size)
            idle_.push_back(std::move(conn));
        cv_.notify_one();
    }
};

// Usage
PgPool pool{{"dbname=mydb", 4, 20}};
{
    auto lease = pool.checkout();
    pqxx::work txn{lease.get()};
    txn.exec("UPDATE hits SET count = count + 1 WHERE page = 'home'");
    txn.commit();
}  // connection returns to pool automatically

Prepared Statements and Parameterized Queries

Prepared statements are parsed once and executed many times — they eliminate repeated parse overhead and, more importantly, make SQL injection structurally impossible. Both SQLite's C API and libpq expose distinct prepare/bind/execute phases.

cpp

Godbolt

// --- SQLite3 C API: persistent prepared statement ---
sqlite3_stmt* insert_stmt = nullptr;
sqlite3_prepare_v2(db,
    "INSERT INTO events (user_id, action, ts) VALUES (?, ?, ?)",
    -1, &insert_stmt, nullptr);

// Reuse the same stmt thousands of times
auto log_event = [&](int uid, std::string_view action, int64_t ts) {
    sqlite3_reset(insert_stmt);
    sqlite3_clear_bindings(insert_stmt);
    sqlite3_bind_int(insert_stmt,  1, uid);
    sqlite3_bind_text(insert_stmt, 2, action.data(),
                      static_cast<int>(action.size()), SQLITE_TRANSIENT);
    sqlite3_bind_int64(insert_stmt, 3, ts);
    if (sqlite3_step(insert_stmt) != SQLITE_DONE)
        throw std::runtime_error(sqlite3_errmsg(db));
};

// Clean up once
sqlite3_finalize(insert_stmt);

// --- libpq C API: server-side prepared statement ---
#include <libpq-fe.h>

PGconn* pg = PQconnectdb("dbname=mydb");

// Prepare on the server — name must be unique per connection
PGresult* res = PQprepare(pg, "ins_event",
    "INSERT INTO events (user_id, action) VALUES ($1, $2)",
    0, nullptr);
PQclear(res);

auto pg_log = [&](int uid, const std::string& action) {
    std::string uid_s = std::to_string(uid);
    const char* params[] = { uid_s.c_str(), action.c_str() };
    PGresult* r = PQexecPrepared(pg, "ins_event", 2, params, nullptr, nullptr, 0);
    if (PQresultStatus(r) != PGRES_COMMAND_OK)
        throw std::runtime_error(PQerrorMessage(pg));
    PQclear(r);
};

ORM Libraries Compared

SOCI, sqlpp11, and ODB each offer C++ type safety but differ fundamentally in approach — choose based on whether you need reflection-free mapping, compile-time SQL, or schema generation. None of them requires raw string concatenation.

Library	Query style	Code gen?	DB support	Header-only?
SOCI	Stream-like `<<`	No	SQLite, PG, MySQL, Oracle	No
sqlpp11	C++ operator overloading	Yes (tables)	SQLite, MySQL, PG	Yes (core)
ODB	Pragma-annotated classes	Yes (ORM)	All major	No

cpp

Godbolt

// --- SOCI ---
#include <soci/soci.h>
#include <soci/sqlite3/soci-sqlite3.h>

soci::session sql{soci::sqlite3, "app.db"};

int count;
sql << "SELECT COUNT(*) FROM users WHERE active = :a", soci::use(true), soci::into(count);

struct User { int id; std::string name; };
User u;
soci::statement st = (sql.prepare
    << "SELECT id, name FROM users WHERE id = :id",
       soci::use(42), soci::into(u.id), soci::into(u.name));
st.execute(true);

// --- sqlpp11 (generated table header from DDL) ---
// Auto-generated: sqlpp11-ddl2cpp schema.sql schema.h
#include "schema.h"
#include <sqlpp11/sqlite3/sqlite3.h>

auto db = sqlpp::sqlite3::connection{cfg};
Users users;
for (const auto& row : db(select(users.id, users.name)
                               .from(users)
                               .where(users.active == true))) {
    std::println("{}: {}", row.id, row.name.value());
}
// SQL injection is impossible — query is a C++ expression tree

Async Database Access

Blocking database calls stall a coroutine scheduler; use non-blocking drivers or thread-pool offloading so the event loop stays responsive. libpqxx 7+ adds async pipeline support; for coroutines the simplest approach is co_await-able wrappers around thread-pool submissions.

cpp

Godbolt

#include <pqxx/pqxx>
#include <coroutine>
#include <future>
#include <thread>

// Simple awaitable that runs a DB call on a thread pool
template<typename R>
struct DbTask {
    std::future<R> fut_;

    bool await_ready() { return false; }

    void await_suspend(std::coroutine_handle<> h) {
        std::thread([this, h]() mutable {
            // future already set by the factory below
            h.resume();
        }).detach();
    }

    R await_resume() { return fut_.get(); }
};

// Factory: submit lambda to pool, return awaitable
template<typename F>
auto db_async(F&& f) {
    using R = std::invoke_result_t<F>;
    auto task = std::async(std::launch::async, std::forward<F>(f));
    return DbTask<R>{std::move(task)};
}

// libpqxx pipeline (batches multiple statements, overlaps server-side execution)
void pipeline_example(pqxx::connection& conn) {
    pqxx::work txn{conn};
    pqxx::pipeline pipe{txn};

    auto id1 = pipe.insert("SELECT name FROM users WHERE id = 1");
    auto id2 = pipe.insert("SELECT name FROM users WHERE id = 2");
    auto id3 = pipe.insert("SELECT COUNT(*) FROM events");

    pipe.complete();   // send all, receive all

    std::println("{}", pipe.retrieve(id1).front()["name"].c_str());
    std::println("{}", pipe.retrieve(id2).front()["name"].c_str());
    std::println("{}", pipe.retrieve(id3).front()[0].as<int>());

    txn.commit();
}

SQLite as Embedded Database

SQLite's WAL mode, page-size tuning, and in-memory variant make it a first-class embedded store — not just a toy. A few PRAGMA settings applied at open time can double throughput for write-heavy workloads.

cpp

Godbolt

sqlite3* db;
sqlite3_open("app.db", &db);

// Recommended PRAGMA set for performance
const char* setup_sql = R"(
    PRAGMA journal_mode = WAL;        -- concurrent reads during write
    PRAGMA synchronous   = NORMAL;    -- safe with WAL, much faster than FULL
    PRAGMA cache_size    = -32000;    -- 32 MB page cache (negative = kibibytes)
    PRAGMA temp_store    = MEMORY;    -- temp tables in RAM
    PRAGMA mmap_size     = 134217728; -- 128 MB memory-mapped I/O
    PRAGMA foreign_keys  = ON;
)";
sqlite3_exec(db, setup_sql, nullptr, nullptr, nullptr);

// In-memory database (lost on close, great for tests)
sqlite3* mem_db;
sqlite3_open(":memory:", &mem_db);

// Shared-cache in-memory (multiple connections see same data in one process)
sqlite3_open("file:sharedmem?mode=memory&cache=shared", &mem_db);

// WAL checkpoint (force WAL frames back to main DB file)
sqlite3_wal_checkpoint_v2(db, nullptr, SQLITE_CHECKPOINT_TRUNCATE,
                           nullptr, nullptr);

PRAGMA	Default	Recommended	Effect
`journal_mode`	DELETE	WAL	Concurrent reads; writer doesn't block readers
`synchronous`	FULL	NORMAL	Fsync only at WAL checkpoint
`cache_size`	-2000 (2 MB)	-32000 (32 MB)	Reduces disk I/O
`temp_store`	FILE	MEMORY	Sort/group temp tables in RAM
`mmap_size`	0	128 MB+	OS-level read caching via mmap

Transactions and Isolation Levels

Wrapping multiple statements in a single transaction is the single biggest performance lever for SQLite and PostgreSQL — it collapses N fsync calls to one. Retry logic on serialization failures is mandatory for PostgreSQL with REPEATABLE READ or SERIALIZABLE.

cpp

Godbolt

// SQLite: explicit transaction with retry on SQLITE_BUSY
bool run_transaction(sqlite3* db, auto body) {
    for (int attempt = 0; attempt < 5; ++attempt) {
        sqlite3_exec(db, "BEGIN IMMEDIATE", nullptr, nullptr, nullptr);
        try {
            body();
            if (sqlite3_exec(db, "COMMIT", nullptr, nullptr, nullptr) == SQLITE_OK)
                return true;
        } catch (...) {}
        sqlite3_exec(db, "ROLLBACK", nullptr, nullptr, nullptr);
        // Exponential back-off
        std::this_thread::sleep_for(std::chrono::milliseconds(10 << attempt));
    }
    return false;
}

// SQLite savepoint (nested transaction)
sqlite3_exec(db, "SAVEPOINT sp1", nullptr, nullptr, nullptr);
// ... inner work ...
sqlite3_exec(db, "RELEASE SAVEPOINT sp1",   nullptr, nullptr, nullptr); // commit inner
// or:
sqlite3_exec(db, "ROLLBACK TO SAVEPOINT sp1", nullptr, nullptr, nullptr); // undo inner

// PostgreSQL isolation levels via pqxx
pqxx::work txn{conn};
txn.exec("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ");
// ... queries ...
try {
    txn.commit();
} catch (const pqxx::serialization_failure&) {
    // Retry the entire transaction — state was inconsistent
}

Isolation level	Dirty read	Non-repeatable read	Phantom read	Serialization anomaly
READ COMMITTED (default PG)	No	Yes	Yes	Yes
REPEATABLE READ	No	No	No (PG)	Yes
SERIALIZABLE	No	No	No	No

Schema Migrations in C++

A version table records the highest applied migration number; the application compares it to its compiled-in list and runs only the delta on startup. This avoids external migration tools while keeping the schema evolution in source control alongside the C++ code.

cpp

Godbolt

#include <sqlite3.h>
#include <string_view>
#include <vector>
#include <stdexcept>

struct Migration {
    int         version;
    std::string_view sql;
};

constexpr Migration migrations[] = {
    {1, "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL)"},
    {2, "ALTER TABLE users ADD COLUMN email TEXT"},
    {3, "CREATE INDEX idx_users_email ON users(email)"},
};

void run_migrations(sqlite3* db) {
    // Ensure the schema_version table exists
    sqlite3_exec(db,
        "CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL)",
        nullptr, nullptr, nullptr);

    // Read current version (0 if table is empty)
    sqlite3_stmt* s;
    sqlite3_prepare_v2(db,
        "SELECT COALESCE(MAX(version), 0) FROM schema_version",
        -1, &s, nullptr);
    sqlite3_step(s);
    int current = sqlite3_column_int(s, 0);
    sqlite3_finalize(s);

    for (const auto& m : migrations) {
        if (m.version <= current) continue;

        sqlite3_exec(db, "BEGIN", nullptr, nullptr, nullptr);

        char* err = nullptr;
        if (sqlite3_exec(db, m.sql.data(), nullptr, nullptr, &err) != SQLITE_OK) {
            std::string msg = err;
            sqlite3_free(err);
            sqlite3_exec(db, "ROLLBACK", nullptr, nullptr, nullptr);
            throw std::runtime_error("Migration " + std::to_string(m.version)
                                     + " failed: " + msg);
        }

        sqlite3_stmt* upd;
        sqlite3_prepare_v2(db,
            "INSERT INTO schema_version (version) VALUES (?)", -1, &upd, nullptr);
        sqlite3_bind_int(upd, 1, m.version);
        sqlite3_step(upd);
        sqlite3_finalize(upd);

        sqlite3_exec(db, "COMMIT", nullptr, nullptr, nullptr);
    }
}

BLOB Storage

SQLite and PostgreSQL both store binary data directly in the database, avoiding file-system coherency problems — useful for thumbnails, certificates, or any asset tightly coupled to a row. For objects larger than a few megabytes, stream them in chunks rather than loading the whole blob into memory.

cpp

Godbolt

// --- SQLite: write a BLOB ---
sqlite3_stmt* ins;
sqlite3_prepare_v2(db,
    "INSERT INTO files (name, data) VALUES (?, ?)", -1, &ins, nullptr);

std::vector<uint8_t> file_bytes = read_file("cert.pem");
sqlite3_bind_text(ins, 1, "cert.pem", -1, SQLITE_STATIC);
sqlite3_bind_blob(ins, 2, file_bytes.data(),
                  static_cast<int>(file_bytes.size()), SQLITE_TRANSIENT);
sqlite3_step(ins);
sqlite3_finalize(ins);

// --- SQLite: stream-read a large BLOB (zero-copy) ---
sqlite3_blob* blob;
sqlite3_blob_open(db, "main", "files", "data",
                  /*rowid=*/1, /*write=*/0, &blob);

int blob_size = sqlite3_blob_bytes(blob);
constexpr int CHUNK = 65536;
std::vector<uint8_t> chunk(CHUNK);

for (int offset = 0; offset < blob_size; offset += CHUNK) {
    int to_read = std::min(CHUNK, blob_size - offset);
    sqlite3_blob_read(blob, chunk.data(), to_read, offset);
    process_chunk(chunk.data(), to_read);
}
sqlite3_blob_close(blob);

// --- PostgreSQL: large object (lo_*) API via libpq ---
#include <libpq-fe.h>

PGconn* pg = PQconnectdb("dbname=mydb");
PQexec(pg, "BEGIN");

// Create a new large object, returns OID
Oid loid = lo_creat(pg, INV_READ | INV_WRITE);
int lfd  = lo_open(pg, loid, INV_WRITE);

uint8_t buf[4096];
size_t  n;
FILE*   f = fopen("big_file.bin", "rb");
while ((n = fread(buf, 1, sizeof(buf), f)) > 0)
    lo_write(pg, lfd, reinterpret_cast<char*>(buf), n);
fclose(f);

lo_close(pg, lfd);
PQexec(pg, "COMMIT");

// Store loid in a row for later retrieval
std::string q = "UPDATE assets SET data_oid = " + std::to_string(loid)
                + " WHERE id = 42";
PQexec(pg, q.c_str());