Before you start

Make sure that you can compile and run core tests and read a basic example Writing your first HTTP server.

Also, it is expected that you are familiar with basic YDB notions (tables, topics, changefeeds, consumers, etc.).

Step by step guide

In this tutorial we will write a service that demonstrates the YDB driver. The sample covers table queries, transactions, BSON storage, and topic consumption from changefeeds.

The service would have the following REST API:

HTTP POST by path /ydb/upsert-row inserts a single row into the events table;
HTTP POST by path /ydb/upsert-rows inserts multiple rows in one query;
HTTP POST by path /ydb/upsert-2rows inserts two rows inside a transaction;
HTTP POST by path /ydb/select-rows selects rows by secondary index;
HTTP POST by path /ydb/bson-upserting stores a BSON document in the orders table;
HTTP POST by path /ydb/bson-reading reads a BSON document from the orders table;

Additionally, the service contains two background components that consume changefeed messages from YDB topics.

HTTP handler base class

Most handlers in the sample inherit from a common base class that obtains a ydb::TableClient from ydb::YdbComponent. That client is thread safe, you can use it concurrently from different threads and tasks.

class BaseHandler : public server::handlers::HttpHandlerJsonBase {
public:
    BaseHandler(const components::ComponentConfig& config, const components::ComponentContext& context)
        : HttpHandlerJsonBase(config, context),
          ydb_client_(context.FindComponent<ydb::YdbComponent>().GetTableClient("sampledb"))
    {}
 
protected:
    ydb::TableClient& Ydb() const { return *ydb_client_; }
 
private:
    std::shared_ptr<ydb::TableClient> ydb_client_;
};

UpsertRowHandler

The simplest way to execute a YQL query is to call ydb::TableClient::ExecuteDataQuery() with a ydb::Query object and bound parameters:

formats::json::Value UpsertRowHandler::
    HandleRequestJsonThrow(const server::http::HttpRequest& http_request, const formats::json::Value& request, server::request::RequestContext&)
        const {
    http_request.GetHttpResponse().SetContentType(http::content_type::kApplicationJson);
    static const ydb::Query kUpsertQuery{
        R"(
--!syntax_v1
DECLARE $id_key AS String;
DECLARE $name_key AS Utf8;
DECLARE $service_key AS String;
DECLARE $channel_key AS Int64;
DECLARE $state_key AS Json?;
 
UPSERT INTO events (id, name, service, channel, created, state)
VALUES ($id_key, $name_key, $service_key, $channel_key, CurrentUtcTimestamp(), $state_key);
      )",
        ydb::Query::NameLiteral{"upsert-row"},
        ydb::Query::LogMode::kNameOnly,
    };
 
    auto response = Ydb().ExecuteDataQuery(
        kUpsertQuery,  //
        "$id_key",
        request["id"].As<std::string>(),  //
        "$name_key",
        request["name"].As<ydb::Utf8>(),  //
        "$service_key",
        request["service"].As<std::string>(),  //
        "$channel_key",
        request["channel"].As<std::int64_t>(),  //
        "$state_key",
        request["state"].As<std::optional<formats::json::Value>>()  //
    );
 
    if (response.GetCursorCount()) {
        throw std::runtime_error("Unexpected response data");
    }
 
    return formats::json::MakeObject();
}

Note that you can pass queries directly as string literals, or store them in external YQL files. See Embedding SQL/YQL files via userver_add_sql_library for more information.

UpsertRowsHandler

For bulk inserts you can bind a list of C++ structs to a YQL parameter. The struct must declare ydb::StructMemberNames and provide a Parse() function for JSON deserialization:

struct UpsertRowsRequest {
    static constexpr ydb::StructMemberNames kYdbMemberNames{};
 
    std::string id;
    ydb::Utf8 name;
    std::string service;
    std::optional<int64_t> channel;
};
 
UpsertRowsRequest Parse(const formats::json::Value& json, formats::parse::To<UpsertRowsRequest>) {
    UpsertRowsRequest request;
 
    request.id = json["id"].As<std::string>();
    request.name = ydb::Utf8{json["name"].As<std::string>()};
    request.service = json["service"].As<std::string>();
    request.channel = json["channel"].As<std::optional<int64_t>>();
 
    return request;
}

Then pass the vector of structs to ExecuteDataQuery:

formats::json::Value UpsertRowsHandler::
    HandleRequestJsonThrow(const server::http::HttpRequest& request, const formats::json::Value& request_json, server::request::RequestContext&)
        const {
    request.GetHttpResponse().SetContentType(http::content_type::kApplicationJson);
    static const ydb::Query kUpsertQuery{
        R"(
--!syntax_v1
DECLARE $items AS List<Struct<'id': String, 'name': Utf8, 'service':
                       String, 'channel': Int64?>>;
UPSERT INTO events (id, name, service, channel, created)
SELECT id, name, service, channel, CurrentUtcTimestamp() FROM AS_TABLE($items);
      )",
        ydb::Query::NameLiteral{"upsert-rows"},
        ydb::Query::LogMode::kNameOnly,
    };
 
    const auto upsert_rows = request_json["items"].As<std::vector<UpsertRowsRequest>>();
 
    const auto response = Ydb().ExecuteDataQuery(kUpsertQuery, "$items", upsert_rows);
 
    if (response.GetCursorCount()) {
        throw std::runtime_error("Unexpected response data");
    }
 
    return formats::json::MakeObject();
}

Upsert2RowsHandler

To run several queries atomically, use ydb::TableClient::RetryTx(). The lambda receives a ydb::TxActor and should return ydb::TxAction::kCommit on success:

formats::json::
    Value
    Upsert2RowsHandler::HandleRequestJsonThrow(const server::http::HttpRequest& http_request, const formats::json::Value& request, server::request::RequestContext&) const {
    http_request.GetHttpResponse().SetContentType(http::content_type::kApplicationJson);
    static const ydb::Query kUpsertQuery{
        R"(
--!syntax_v1
DECLARE $id_key AS String;
DECLARE $name_key AS Utf8;
DECLARE $service_key AS String;
DECLARE $channel_key AS Int64;
DECLARE $state_key AS Json?;
 
UPSERT INTO events (id, name, service, channel, created, state)
VALUES ($id_key, $name_key, $service_key, $channel_key, CurrentUtcTimestamp(), $state_key);
      )",
        ydb::Query::NameLiteral{"upsert-2rows"},
        ydb::Query::LogMode::kNameOnly,
    };
 
    Ydb().RetryTx("trx", {.tx_mode = ydb::TransactionMode::kSerializableRW}, [&](ydb::TxActor& tx) {
        for (auto i : {1, 2}) {
            auto response = tx.Execute(
                kUpsertQuery,  //
                "$id_key",
                request["id"].As<std::string>() + std::to_string(i),  //
                "$name_key",
                ydb::Utf8{request["name"].As<std::string>() + std::to_string(i)},  //
                "$service_key",
                request["service"].As<std::string>(),  //
                "$channel_key",
                request["channel"].As<int64_t>(),  //
                "$state_key",
                request["state"].As<std::optional<formats::json::Value>>()  //
            );
 
            if (response.GetCursorCount() != 0) {
                throw std::runtime_error("Unexpected response data");
            }
        }
 
        return ydb::TxAction::kCommit;
    });
 
    return formats::json::MakeObject();
}

SelectRowsHandler

Reads can specify per-query ydb::OperationSettings (retries, timeout, transaction mode) and iterate over the result cursor:

formats::json::Value SelectRowsHandler::
    HandleRequestJsonThrow(const server::http::HttpRequest& request, const formats::json::Value& request_json, server::request::RequestContext&)
        const {
    request.GetHttpResponse().SetContentType(http::content_type::kApplicationJson);
    const ydb::OperationSettings query_params = {
        3,                                // retries
        std::chrono::milliseconds(1100),  // client_timeout
        ydb::TransactionMode::kStaleRO
    };
    static const ydb::Query kSelectQuery = {
        "--!syntax_v1\n"
        "DECLARE $service_key AS String;"
        "DECLARE $channel_keys AS List<Int64>;"
        "DECLARE $created_key AS Timestamp;"
        "SELECT id, name, service, channel, created, state "
        "FROM events VIEW sample_index "
        "WHERE service = $service_key AND channel IN $channel_keys "
        "AND created > $created_key;",
        ydb::Query::NameLiteral("select"),
        ydb::Query::LogMode::kNameOnly,
    };
 
    auto response = Ydb().ExecuteDataQuery(
        query_params,
        kSelectQuery,  //
        "$service_key",
        request_json["service"].As<std::string>(),  //
        "$channel_keys",
        request_json["channels"].As<std::vector<int64_t>>(),  //
        "$created_key",
        request_json["created"].As<std::chrono::system_clock::time_point>()
    );
 
    if (response.GetCursorCount() != 1) {
        throw std::runtime_error("Unexpected response data");
    }
 
    formats::json::ValueBuilder items_builder(formats::json::Type::kArray);
    for (auto row : response.GetSingleCursor()) {
        formats::json::ValueBuilder item;
        item["id"] = row.Get<std::string>("id");
        item["name"] = row.Get<ydb::Utf8>("name").GetUnderlying();
        item["service"] = row.Get<std::string>("service");
        item["channel"] = row.Get<std::int64_t>("channel");
        item["created"] = row.Get<std::chrono::system_clock::time_point>("created");
 
        auto state = row.Get<std::optional<formats::json::Value>>("state");
        if (state) {
            item["state"] = std::move(*state);
        }
 
        items_builder.PushBack(item.ExtractValue());
    }
 
    return formats::json::MakeObject("items", items_builder.ExtractValue());
}

BSON handlers

YDB can store opaque binary data. The sample stores BSON documents in a String column and converts them using formats::bson helpers.

The reading handler obtains the table client directly from the component:

class BsonReadingHandler final : public server::handlers::HttpHandlerBase {
public:
    static constexpr std::string_view kName = "handler-bson-reading";
 
    BsonReadingHandler(const components::ComponentConfig& config, const components::ComponentContext& context)
        : HttpHandlerBase(config, context),
          ydb_client_(context.FindComponent<ydb::YdbComponent>().GetTableClient("sampledb"))
    {}
 
    std::string HandleRequest(server::http::HttpRequest& request, server::request::RequestContext& context)
        const override;
 
private:
    std::shared_ptr<ydb::TableClient> ydb_client_;
};

std::string BsonReadingHandler::HandleRequest(server::http::HttpRequest& request, server::request::RequestContext&)
    const {
    static const std::string kSelectQuery = R"(
--!syntax_v1
DECLARE $id AS String;
SELECT doc FROM orders WHERE id = $id;
  )";
    auto response = ydb_client_->ExecuteDataQuery(kSelectQuery, "$id", request.GetArg("id"));
 
    if (response.GetCursorCount() != 1) {
        throw std::runtime_error("Unexpected response data");
    }
    auto cursor = response.GetSingleCursor();
 
    if (!cursor.empty()) {
        auto row = cursor.GetFirstRow();
 
        return formats::bson::ToBinaryString(
                   formats::bson::MakeDoc("doc", formats::bson::FromBinaryString(row.Get<std::string>("doc")))
        )
            .ToString();
    }
 
    request.SetResponseStatus(server::http::HttpStatus::kNotFound);
    return {};
}

The upserting handler writes the raw request body into the table:

class BsonUpsertingHandler final : public server::handlers::HttpHandlerBase {
public:
    static constexpr std::string_view kName = "handler-bson-upserting";
 
    BsonUpsertingHandler(const components::ComponentConfig& config, const components::ComponentContext& context)
        : HttpHandlerBase(config, context),
          ydb_client_(context.FindComponent<ydb::YdbComponent>().GetTableClient("sampledb"))
    {}
 
    std::string HandleRequest(server::http::HttpRequest& request, server::request::RequestContext& context)
        const override;
 
private:
    std::shared_ptr<ydb::TableClient> ydb_client_;
};

std::string BsonUpsertingHandler::HandleRequest(server::http::HttpRequest& request, server::request::RequestContext&)
    const {
    const auto& id = request.GetArg("id");
    const auto& body = request.RequestBody();
 
    ydb_client_->ExecuteDataQuery(kInsertQuery, "$id", id, "$doc", body);
    return {};
}

Topic reader components

The sample demonstrates two ways to consume YDB topics:

ydb::TopicClient — regular topic reader;
ydb::FederatedTopicClient — federated topic reader.

Both components start a background task on construction that creates a read session, processes events in a loop, and restarts the session on failure:

TopicReaderComponent::TopicReaderComponent(
    const components::ComponentConfig& config,
    const components::ComponentContext& context
)
    : components::ComponentBase(config, context) {
    const auto consumer_name = config["consumer-name"].As<std::string>();
    const auto topics = config["topics"].As<std::vector<std::string>>();
    const auto restart_session_delay = config["restart-session-delay"].As<std::chrono::milliseconds>();
 
    auto topic_reader = std::make_unique<TopicReader>(
        context.FindComponent<ydb::YdbComponent>().GetTopicClient("sampledb"),
        ConstructReadSessionSettings(consumer_name, topics),
        restart_session_delay
    );
 
    read_task_ = utils::CriticalAsync(config.Name() + "-read-task", [topic_reader = std::move(topic_reader)] {
        topic_reader->Run();
    });
}

The federated variant uses GetFederatedTopicClient() instead:

FederatedTopicReaderComponent::FederatedTopicReaderComponent(
    const components::ComponentConfig& config,
    const components::ComponentContext& context
)
    : components::ComponentBase(config, context) {
    const auto consumer_name = config["consumer-name"].As<std::string>();
    const auto topics = config["topics"].As<std::vector<std::string>>();
    const auto restart_session_delay = config["restart-session-delay"].As<std::chrono::milliseconds>();
 
    auto topic_reader = std::make_unique<FederatedTopicReader>(
        context.FindComponent<ydb::YdbComponent>().GetFederatedTopicClient("sampledb"),
        ConstructReadSessionSettings(consumer_name, topics),
        restart_session_delay
    );
 
    read_task_ = utils::CriticalAsync(config.Name() + "-read-task", [topic_reader = std::move(topic_reader)] {
        topic_reader->Run();
    });
}

The table records has a changefeed configured in samples/ydb_service/ydb/migrations/0002_records_changefeed.sql. When rows are inserted or deleted, the topic reader components receive JSON messages with the changed data.

Static config

Static configuration of the service is quite close to the configuration from Writing your first HTTP server except for the YDB component, handlers and topic readers:

        # yaml
        ydb:
            operation-settings:
                client-timeout: 1100ms
                retries: 3
            databases:
                sampledb:
                    database: sampledb
                    max_pool_size: 10
                    min_pool_size: 5

HTTP handlers:

        # yaml
        handler-bson-reading:
            method: POST
            path: /ydb/bson-reading
            task_processor: main-task-processor
        handler-bson-upserting:
            method: POST
            path: /ydb/bson-upserting
            task_processor: main-task-processor
        handler-select-rows:
            method: POST
            path: /ydb/select-rows
            task_processor: main-task-processor
        handler-upsert-row:
            method: POST
            path: /ydb/upsert-row
            task_processor: main-task-processor
        handler-upsert-rows:
            method: POST
            path: /ydb/upsert-rows
            task_processor: main-task-processor
        handler-upsert-2rows:
            method: POST
            path: /ydb/upsert-2rows
            task_processor: main-task-processor

Topic reader components:

        # yaml
        sample-topic-reader:
            consumer-name: sample-topic-consumer
            topics:
              - records/changefeed
            restart-session-delay: 200ms
 
        sample-federated-topic-reader:
            consumer-name: sample-federated-topic-consumer
            topics:
              - records/changefeed
            restart-session-delay: 200ms

There are more static options for the YDB component configuration, all of them are described at ydb::YdbComponent.

See YDB for YDB hints and more usage samples.

int main()

Finally, we add our components to the components::MinimalServerComponentList(), and start the server with static configuration.

int main(int argc, char* argv[]) {
    auto component_list =
        ::components::MinimalServerComponentList()
            .AppendComponentList(USERVER_NAMESPACE::dynamic_config::updater::ComponentList())
            .Append<components::TestsuiteSupport>()
            .Append<server::handlers::TestsControl>()
            .AppendComponentList(clients::http::ComponentList())
            .Append<clients::dns::Component>()
            .Append<components::DefaultSecdistProvider>()
            .Append<components::Secdist>()
            .Append<sample::BsonReadingHandler>()
            .Append<sample::BsonUpsertingHandler>()
            .Append<sample::SelectRowsHandler>()
            .Append<sample::Upsert2RowsHandler>()
            .Append<sample::UpsertRowHandler>()
            .Append<sample::UpsertRowsHandler>()
            .Append<sample::TopicReaderComponent>()
            .Append<sample::FederatedTopicReaderComponent>()
            .Append<ydb::YdbComponent>();
 
    return ::utils::DaemonMain(argc, argv, component_list);
}

Build and Run

To build the sample, execute the following build steps at the userver root directory:

mkdir build_release
cd build_release
cmake -DCMAKE_BUILD_TYPE=Release ..
make userver-samples-ydb_service

The sample could be started by running make start-userver-samples-ydb_service. The command would invoke testsuite start target that sets proper paths in the configuration files, prepares and starts YDB, and starts the service.

To start the service manually start the YDB server and run ./samples/ydb_service/userver-samples-ydb_service -c </path/to/static_config.yaml>.

Now you can send a request to your service from another terminal:

bash
$ curl -X POST 'http://localhost:8080/ydb/upsert-row' \
    -H 'Content-Type: application/json' \
    -d '{"id":"hello","name":"world","service":"demo","channel":1}'
{}

Functional testing

Functional tests for the service could be implemented using the testsuite. To do that you have to:

Turn on the pytest_userver.plugins.ydb plugin:
# Import YDB plugins

pytest_plugins = ['pytest_userver.plugins.ydb']
Write the tests:
async def test_upsert_row(service_client, ydb):

response = await service_client.post(

'ydb/upsert-row',

json={

'id': 'id-upsert',

'name': 'name-upsert',

'service': 'srv',

'channel': 123,

},

)

assert response.status_code == 200

assert 'application/json' in response.headers['Content-Type']

assert response.json() == {}

cursor = ydb.execute('SELECT * FROM events WHERE id = "id-upsert"')

assert len(cursor) == 1

assert len(cursor[0].rows) == 1

row = cursor[0].rows[0]

assert row.pop('created') > 0

assert row == {

'id': b'id-upsert',

'name': 'name-upsert',

'service': b'srv',

'channel': 123,

'state': None,

}

@pytest.mark.ydb(files=['fill_events.sql'])

async def test_select_rows(service_client):

response = await service_client.post(

'ydb/select-rows',

json={

'service': 'srv',

'channels': [1, 2, 3],

'created': '2019-10-30T11:20:00+00:00',

},

)

assert response.status_code == 200

assert 'application/json' in response.headers['Content-Type']

assert response.json() == {

'items': [

{

'id': 'id-1',

'name': 'name-1',

'service': 'srv',

'channel': 1,

'created': '2019-10-31T11:20:00+00:00',

},

],

}

@pytest.mark.ydb(files=['fill_orders.sql'])

async def test_bson_reading(service_client):

response = await service_client.post(

'ydb/bson-reading',

params={'id': 'id1'},

)

assert response.status_code == 200

raw_bson = bson.BSON(response.content).decode()

res = raw_bson['doc']

assert res == {

'h': {'e': {'l': {'l': '0'}}},

'w': {'o': {'r': {'l': 'd'}}},

}

async def test_ok(service_client, ydb):

# validate YDB state

cursor = ydb.execute(SQL_REQUEST)

assert len(cursor) == 1

assert not cursor[0].rows

# perform request

response = await service_client.post(

'ydb/bson-upserting',

params={'id': 'id1'},

data=bson.BSON.encode(DATA),

headers={'Content-Type': 'application/bson'},

)

# validate response

assert response.status_code == 200

# validate YDB state

cursor = ydb.execute(SQL_REQUEST)

assert len(cursor) == 1

assert len(cursor[0].rows) == 1

assert DATA == bson.BSON.decode(cursor[0].rows[0]['doc'])

@pytest.mark.parametrize(

'testpoint_name',

[

pytest.param('topic-handle-message', id='topic'),

pytest.param('federated-topic-handle-message', id='federated-topic'),

],

)

async def test_topic(testpoint_name, service_client, ydb, testpoint):

@testpoint(testpoint_name)

async def handle_message_testpoint(data):

pass

await service_client.enable_testpoints()

ydb.execute(

"""

UPSERT INTO `records` (id, name, str, num)

VALUES ("test-topic-id", "test-topic-name", "test-topic-str", 321);

""",

)

upsert_record_args = await handle_message_testpoint.wait_call()

upsert_record_message = upsert_record_args['data']

assert upsert_record_message['key'] == ['test-topic-id', 'test-topic-name']

assert base64.b64decode(upsert_record_message['newImage']['str']) == b'test-topic-str'

assert upsert_record_message['newImage']['num'] == 321

ydb.execute(

"""

DELETE FROM `records`

WHERE id = "test-topic-id" AND name = "test-topic-name";

""",

)

delete_record_args = await handle_message_testpoint.wait_call()

delete_record_message = delete_record_args['data']

assert delete_record_message['key'] == ['test-topic-id', 'test-topic-name']

assert 'newImage' not in delete_record_message

Full sources

See the full example:

⇦ Kafka service | YDB topic writer service ⇨