Skip to content

feat: add bulk import DTO request/response and Status-returning BulkImport overloads #613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yhmo wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat: add bulk import DTO request/response and Status-returning BulkImport overloads #613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
126 changes: 109 additions & 17 deletions src/impl/BulkImport.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,23 +21,100 @@
namespace milvus {
namespace {

nlohmann::json
PostImportRequest(const std::string& url, const std::string& request_path, const std::string& api_key,
const nlohmann::json& request_payload) {
httplib::Client client(url);
httplib::Headers headers = {
{"Authorization", "Bearer " + api_key},
};
Status
PostImport(const std::string& url, const std::string& request_path, const std::string& api_key,
const nlohmann::json& request_payload, const std::string& db_name, const BulkImportConfig& config,
BulkImportResponse& response) {
response.SetRawJson(nlohmann::json::object());
try {
httplib::Client client(url);
client.enable_server_certificate_verification(config.VerifyServerCert());
if (!config.CaCertPath().empty()) {
client.set_ca_cert_path(config.CaCertPath());
}
if (config.Timeout() > 0) {
client.set_connection_timeout(config.Timeout());
client.set_read_timeout(config.Timeout(), 0);
client.set_write_timeout(config.Timeout(), 0);
}
httplib::Headers headers = {
Comment thread
yhmo marked this conversation as resolved.
{"Authorization", "Bearer " + api_key},
};
if (!db_name.empty()) {
headers.emplace("DB-Name", db_name);
}

auto result = client.Post(request_path, headers, request_payload.dump(), "application/json");
if (!result) {
Comment thread
yhmo marked this conversation as resolved.
return {StatusCode::RPC_FAILED, "failed to post import request: " + httplib::to_string(result.error())};
}
if (result->status != 200) {
return {StatusCode::SERVER_FAILED,
"import request failed with HTTP status " + std::to_string(result->status) + ": " + result->body};
}
try {
response.SetRawJson(nlohmann::json::parse(result->body));
} catch (const std::exception& e) {
return {StatusCode::JSON_PARSE_ERROR, std::string("failed to parse import response: ") + e.what()};
}
} catch (const std::exception& e) {
return {StatusCode::UNKNOWN_ERROR, std::string("failed to post import request: ") + e.what()};
}
return Status::OK();
}

auto response = client.Post(request_path, headers, request_payload.dump(), "application/json");
if (response && response->status == 200) {
return nlohmann::json::parse(response->body);
Status
CheckImportResponse(const BulkImportResponse& response) {
if (response.Code() != 0) {
return {StatusCode::SERVER_FAILED,
"import request failed with code " + std::to_string(response.Code()) + ": " + response.Message()};
}
return nullptr;
return Status::OK();
}

} // namespace

Status
BulkImport::CreateImportJobsImpl(const std::string& url, const std::string& api_key,
const nlohmann::json& request_payload, const BulkImportConfig& config,
BulkImportResponse& response) {
auto status = PostImport(url, "/v2/vectordb/jobs/import/create", api_key, request_payload, "", config, response);
return status.IsOk() ? CheckImportResponse(response) : status;
}

Status
BulkImport::ListImportJobsImpl(const std::string& url, const std::string& api_key,
const nlohmann::json& request_payload, const BulkImportConfig& config,
BulkImportResponse& response) {
auto status = PostImport(url, "/v2/vectordb/jobs/import/list", api_key, request_payload, "", config, response);
return status.IsOk() ? CheckImportResponse(response) : status;
}

Status
BulkImport::GetImportJobProgressImpl(const std::string& url, const std::string& api_key,
const nlohmann::json& request_payload, const std::string& db_name,
const BulkImportConfig& config, BulkImportResponse& response) {
auto status =
PostImport(url, "/v2/vectordb/jobs/import/describe", api_key, request_payload, db_name, config, response);
return status.IsOk() ? CheckImportResponse(response) : status;
}

Status
BulkImport::CommitImportImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload,
const std::string& db_name, const BulkImportConfig& config, BulkImportResponse& response) {
auto status =
PostImport(url, "/v2/vectordb/jobs/import/commit", api_key, request_payload, db_name, config, response);
return status.IsOk() ? CheckImportResponse(response) : status;
}

Status
BulkImport::AbortImportImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload,
const std::string& db_name, const BulkImportConfig& config, BulkImportResponse& response) {
auto status =
PostImport(url, "/v2/vectordb/jobs/import/abort", api_key, request_payload, db_name, config, response);
return status.IsOk() ? CheckImportResponse(response) : status;
}

nlohmann::json
BulkImport::CreateImportJobs(const std::string& url, const std::string& collection_name,
const std::vector<std::string>& files, const std::string& db_name,
Expand All @@ -56,7 +133,10 @@ BulkImport::CreateImportJobs(const std::string& url, const std::string& collecti
if (!options.empty()) {
request_payload["options"] = options;
}
return PostImportRequest(url, "/v2/vectordb/jobs/import/create", api_key, request_payload);
BulkImportResponse response;
auto status =
PostImport(url, "/v2/vectordb/jobs/import/create", api_key, request_payload, "", BulkImportConfig{}, response);
return status.IsOk() ? response.RawJson() : nlohmann::json{};
}

nlohmann::json
Expand All @@ -66,28 +146,40 @@ BulkImport::ListImportJobs(const std::string& url, const std::string& collection
{"collectionName", collection_name},
{"dbName", db_name},
};
return PostImportRequest(url, "/v2/vectordb/jobs/import/list", api_key, request_payload);
BulkImportResponse response;
auto status =
PostImport(url, "/v2/vectordb/jobs/import/list", api_key, request_payload, "", BulkImportConfig{}, response);
return status.IsOk() ? response.RawJson() : nlohmann::json{};
}

nlohmann::json
BulkImport::GetImportJobProgress(const std::string& url, const std::string& job_id, const std::string& db_name,
const std::string& api_key) {
nlohmann::json payload = {{"dbName", db_name}, {"jobID", job_id}};
return PostImportRequest(url, "/v2/vectordb/jobs/import/get_progress", api_key, payload);
nlohmann::json payload = {{"dbName", db_name}, {"jobId", job_id}};
Comment thread
yhmo marked this conversation as resolved.
BulkImportResponse response;
auto status = PostImport(url, "/v2/vectordb/jobs/import/get_progress", api_key, payload, db_name,
Comment thread
yhmo marked this conversation as resolved.
BulkImportConfig{}, response);
return status.IsOk() ? response.RawJson() : nlohmann::json{};
}

nlohmann::json
BulkImport::CommitImport(const std::string& url, const std::string& job_id, const std::string& db_name,
const std::string& api_key) {
nlohmann::json payload = {{"dbName", db_name}, {"jobId", job_id}};
return PostImportRequest(url, "/v2/vectordb/jobs/import/commit", api_key, payload);
BulkImportResponse response;
auto status =
PostImport(url, "/v2/vectordb/jobs/import/commit", api_key, payload, db_name, BulkImportConfig{}, response);
return status.IsOk() ? response.RawJson() : nlohmann::json{};
}

nlohmann::json
BulkImport::AbortImport(const std::string& url, const std::string& job_id, const std::string& db_name,
const std::string& api_key) {
nlohmann::json payload = {{"dbName", db_name}, {"jobId", job_id}};
return PostImportRequest(url, "/v2/vectordb/jobs/import/abort", api_key, payload);
BulkImportResponse response;
auto status =
PostImport(url, "/v2/vectordb/jobs/import/abort", api_key, payload, db_name, BulkImportConfig{}, response);
return status.IsOk() ? response.RawJson() : nlohmann::json{};
}

} // namespace milvus
101 changes: 101 additions & 0 deletions src/impl/request/import/CloudDescribeImportRequest.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "milvus/request/import/CloudDescribeImportRequest.h"

namespace milvus {

const std::string&
CloudDescribeImportRequest::ClusterId() const {
return cluster_id_;
}

void
CloudDescribeImportRequest::SetClusterId(const std::string& cluster_id) {
cluster_id_ = cluster_id;
}

CloudDescribeImportRequest&
CloudDescribeImportRequest::WithClusterId(const std::string& cluster_id) {
SetClusterId(cluster_id);
return *this;
}

const std::string&
CloudDescribeImportRequest::ProjectId() const {
return project_id_;
}

void
CloudDescribeImportRequest::SetProjectId(const std::string& project_id) {
project_id_ = project_id;
}

CloudDescribeImportRequest&
CloudDescribeImportRequest::WithProjectId(const std::string& project_id) {
SetProjectId(project_id);
return *this;
}

const std::string&
CloudDescribeImportRequest::RegionId() const {
return region_id_;
}

void
CloudDescribeImportRequest::SetRegionId(const std::string& region_id) {
region_id_ = region_id;
}

CloudDescribeImportRequest&
CloudDescribeImportRequest::WithRegionId(const std::string& region_id) {
SetRegionId(region_id);
return *this;
}

const std::string&
CloudDescribeImportRequest::JobId() const {
return job_id_;
}

void
CloudDescribeImportRequest::SetJobId(const std::string& job_id) {
job_id_ = job_id;
}

CloudDescribeImportRequest&
CloudDescribeImportRequest::WithJobId(const std::string& job_id) {
SetJobId(job_id);
return *this;
}

nlohmann::json
CloudDescribeImportRequest::ToJson() const {
nlohmann::json payload = nlohmann::json::object();
if (!cluster_id_.empty()) {
payload["clusterId"] = cluster_id_;
}
if (!project_id_.empty()) {
payload["projectId"] = project_id_;
}
if (!region_id_.empty()) {
payload["regionId"] = region_id_;
}
payload["jobId"] = job_id_;
return payload;
}

} // namespace milvus
Loading
Loading