Upsert data

Update a single existing entry matched by a predicate — replace its vector, its metadata, or both.

Every operation works on both Vector DB stores (you supply raw vectors) and AI stores (you supply text or images and Ahnlich AI generates the embeddings for you). Use the Vector DB / AI switch in the sample below.

Parameters

Parameter	Type	Description
store	string	Name of the target store.
condition	predicate	A predicate that must match exactly one entry.
new_key	vector · optional	New vector to replace the matched entry's key.
new_value	metadata · optional	Metadata to update on the matched entry.
merge_metadata	bool · optional	true merges into existing metadata; false (default) replaces it entirely.
schema	string · optional	Schema to target. Defaults to public.

All requests accept an optional schema field. When omitted, the server uses the public schema.

Behavior

• Errors if zero or multiple entries match the predicate — it targets a single entry.
• Update the vector only, the metadata only, or both in one call.
• With merge_metadata: true, unchanged fields are preserved.

Sample query

Database engine

Python

Python

import asyncio
from grpclib.client import Channel
from ahnlich_client_py.grpc import keyval, metadata, predicates
from ahnlich_client_py.grpc.services.db_service import DbServiceStub
from ahnlich_client_py.grpc.db import query as db_query


async def upsert():
  async with Channel(host="127.0.0.1", port=1369) as channel:
    client = DbServiceStub(channel)

    condition = predicates.PredicateCondition(
      value=predicates.Predicate(
        equals=predicates.Equals(
          key="id",
          value=metadata.MetadataValue(raw_string="123")
        )
      )
    )

    new_value = keyval.StoreValue(
      value={"status": metadata.MetadataValue(raw_string="published")}
    )

    response = await client.upsert(
      db_query.Upsert(
        store="my_store",
        schema="analytics",
        condition=condition,
        new_value=new_value,
        merge_metadata=True
      )
    )

if __name__ == "__main__":
  asyncio.run(upsert())

Node.js

TypeScript

import { createDbClient } from "ahnlich-client-node";
import { Upsert } from "ahnlich-client-node/grpc/db/query_pb";
import { StoreValue } from "ahnlich-client-node/grpc/keyval_pb";
import { MetadataValue } from "ahnlich-client-node/grpc/metadata_pb";
import { PredicateCondition, Predicate, Equals } from "ahnlich-client-node/grpc/predicates_pb";

async function upsertEntry() {
  const client = createDbClient("127.0.0.1:1369");

  const condition = new PredicateCondition({
    kind: {
      case: "value",
      value: new Predicate({
        kind: {
          case: "equals",
          value: new Equals({
            key: "id",
            value: new MetadataValue({ value: { case: "rawString", value: "123" } }),
          }),
        },
      }),
    },
  });

  const newValue = new StoreValue({
    value: {
      status: new MetadataValue({ value: { case: "rawString", value: "published" } }),
    },
  });

  const response = await client.upsert(
    new Upsert({
      store: "my_store",
      schema: "analytics",
      condition,
      newValue,
      mergeMetadata: true,
    })
  );

  console.log("Updated:", response.upsert?.updated);
}

upsertEntry();

Go

Go

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"

    dbsvc   "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/services/db_service"
    dbquery "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/db/query"
    keyval  "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/keyval"
    metadata "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/metadata"
    predicates "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/predicates"
)

const ServerAddr = "127.0.0.1:1369"

func stringPtr(value string) *string { return &value }

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    conn, err := grpc.NewClient(ServerAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatalf("failed to dial: %v", err)
    }
    defer conn.Close()

    client := dbsvc.NewDBServiceClient(conn)

    condition := &predicates.PredicateCondition{
        Kind: &predicates.PredicateCondition_Value{
            Value: &predicates.Predicate{
                Kind: &predicates.Predicate_Equals{
                    Equals: &predicates.Equals{
                        Key: "id",
                        Value: &metadata.MetadataValue{
                            Value: &metadata.MetadataValue_RawString{RawString: "123"},
                        },
                    },
                },
            },
        },
    }

    newValue := &keyval.StoreValue{
        Value: map[string]*metadata.MetadataValue{
            "status": {Value: &metadata.MetadataValue_RawString{RawString: "published"}},
        },
    }

    resp, err := client.Upsert(ctx, &dbquery.Upsert{
        Store:         "my_store",
        Schema:        stringPtr("analytics"),
        Condition:     condition,
        NewValue:      newValue,
        MergeMetadata: true,
    })
    if err != nil {
        log.Fatalf("Upsert failed: %v", err)
    }
    fmt.Printf("Updated: %d\n", resp.Upsert.Updated)
}

Rust

Rust

use ahnlich_client_rs::db::DbClient;
use ahnlich_types::{
    db::query::Upsert,
    keyval::{StoreKey, StoreValue},
    metadata::{MetadataValue, metadata_value::Value},
    predicates::{Predicate, PredicateCondition, predicate_condition::Kind, predicate::Kind as PredKind},
};
use std::collections::HashMap;


#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connect to DB server
    let db_client = DbClient::new("127.0.0.1:1369".to_string()).await?;

    let tracing_id: Option<String> = None;

    // Construct predicate condition
    let condition = PredicateCondition {
        kind: Some(Kind::Value(Predicate {
            kind: Some(PredKind::Equals(ahnlich_types::predicates::Equals {
                key: "id".to_string(),
                value: Some(MetadataValue {
                    value: Some(Value::RawString("123".to_string())),
                }),
            })),
        })),
    };

    // New metadata to merge/replace
    let new_value = Some(StoreValue {
        value: HashMap::from_iter([(
            "status".into(),
            MetadataValue {
                value: Some(Value::RawString("published".into())),
            },
        )]),
    });

    let params = Upsert {
        store: "my_store".to_string(),
        schema: Some("analytics".to_string()),
        condition: Some(condition),
        new_key: None, // Optional: new vector
        new_value,
        merge_metadata: true, // Merge instead of replace
    };

    // Call upsert
    match db_client.upsert(params, tracing_id).await {
        Ok(result) => {
            println!("Upsert result: {:?}", result);
        }
        Err(err) => {
            eprintln!("Error in upsert: {:?}", err);
        }
    }

    Ok(())
}

CLI

$UPSERT VALUE {status: published} IN my_store WHERE (id = 123)

AI proxy

Python

Python

import asyncio
from grpclib.client import Channel
from ahnlich_client_py.grpc.services.ai_service import AiServiceStub
from ahnlich_client_py.grpc.ai import query as ai_query
from ahnlich_client_py.grpc import keyval, metadata, predicates
from ahnlich_client_py.grpc.ai import preprocess


async def upsert():
  async with Channel(host="127.0.0.1", port=1370) as channel:
      client = AiServiceStub(channel)
      
      condition = predicates.PredicateCondition(
        value=predicates.Predicate(
          equals=predicates.Equals(
            key="filename",
            value=metadata.MetadataValue(raw_string="photo.jpg")
          )
        )
      )
      
      new_value = keyval.StoreValue(
        value={"tags": metadata.MetadataValue(raw_string="cat,outdoors")}
      )
      
      response = await client.upsert(
          ai_query.Upsert(
              store="images",
              schema="media",
              condition=condition,
              new_input=None,  # Optional: new image/text to re-embed
              new_value=new_value,
              preprocess_action=preprocess.PreprocessAction.NoPreprocessing,
              execution_provider=None,
              model_params={}
          )
      )
      print(response) #Set(upsert=StoreUpsert(updated=1, inserted=0))


if __name__ == "__main__":
  asyncio.run(upsert())

Node.js

TypeScript

import { createAiClient } from "ahnlich-client-node";
import { Upsert } from "ahnlich-client-node/grpc/ai/query_pb";
import { StoreValue } from "ahnlich-client-node/grpc/keyval_pb";
import { MetadataValue } from "ahnlich-client-node/grpc/metadata_pb";
import { PredicateCondition, Predicate, Equals } from "ahnlich-client-node/grpc/predicates_pb";
import { PreprocessAction } from "ahnlich-client-node/grpc/ai/preprocess_pb";

async function upsertEntry() {
  const client = createAiClient("127.0.0.1:1370");

  const condition = new PredicateCondition({
    kind: {
      case: "value",
      value: new Predicate({
        kind: {
          case: "equals",
          value: new Equals({
            key: "filename",
            value: new MetadataValue({ value: { case: "rawString", value: "photo.jpg" } }),
          }),
        },
      }),
    },
  });

  const newValue = new StoreValue({
    value: {
      tags: new MetadataValue({ value: { case: "rawString", value: "cat,outdoors" } }),
    },
  });

  const response = await client.upsert(
    new Upsert({
      store: "images",
      schema: "media",
      condition,
      newValue,
      preprocessAction: PreprocessAction.NO_PREPROCESSING,
    })
  );

  console.log("Updated:", response.upsert?.updated);
}

upsertEntry();

Go

Go

package main

import (
  "context"
  "fmt"
  "log"
  "time"

  "google.golang.org/grpc"
  "google.golang.org/grpc/credentials/insecure"

  aiquery "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/ai/query"
  aisvc "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/services/ai_service"
  keyval "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/keyval"
  metadata "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/metadata"
  predicates "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/predicates"
  preprocess "github.com/deven96/ahnlich/sdk/ahnlich-client-go/grpc/ai/preprocess"
)

const AIAddr = "127.0.0.1:1370"

func stringPtr(value string) *string { return &value }

func main() {
  ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  defer cancel()

  conn, err := grpc.DialContext(ctx, AIAddr,
      grpc.WithTransportCredentials(insecure.NewCredentials()),
      grpc.WithBlock(),
  )
  if err != nil {
      log.Fatalf("Failed to connect: %v", err)
  }
  defer conn.Close()

  client := aisvc.NewAIServiceClient(conn)

  condition := &predicates.PredicateCondition{
      Kind: &predicates.PredicateCondition_Value{
          Value: &predicates.Predicate{
              Kind: &predicates.Predicate_Equals{
                  Equals: &predicates.Equals{
                      Key: "filename",
                      Value: &metadata.MetadataValue{
                          Value: &metadata.MetadataValue_RawString{RawString: "photo.jpg"},
                      },
                  },
              },
          },
      },
  }

  newValue := &keyval.StoreValue{
      Value: map[string]*metadata.MetadataValue{
          "tags": {Value: &metadata.MetadataValue_RawString{RawString: "cat,outdoors"}},
      },
  }

  resp, err := client.Upsert(ctx, &aiquery.Upsert{
      Store:             "images",
      Schema:            stringPtr("media"),
      Condition:         condition,
      NewInput:          nil, // Optional: new image/text to re-embed
      NewValue:          newValue,
      PreprocessAction:  preprocess.PreprocessAction_NoPreprocessing,
      ExecutionProvider: nil,
      ModelParams:       map[string]string{},
  })
  if err != nil {
      log.Fatalf("Upsert failed: %v", err)
  }
  fmt.Printf("Updated: %d\n", resp.Upsert.Updated)
}

Rust

Rust

use ahnlich_client_rs::ai::AiClient;
use ahnlich_client_rs::error::AhnlichError;
use ahnlich_types::ai::preprocess::PreprocessAction;
use ahnlich_types::ai::query::Upsert;
use ahnlich_types::keyval::{StoreInput, StoreValue};
use ahnlich_types::metadata::{MetadataValue, metadata_value::Value};
use ahnlich_types::predicates::{Predicate, PredicateCondition, predicate_condition::Kind, predicate::Kind as PredKind};
use std::collections::HashMap;


#[tokio::main]
async fn main() -> Result<(), AhnlichError> {
    // Connect to AI server
    let addr = "127.0.0.1:1370";
    let client = AiClient::new(addr.to_string()).await?;

    // Construct predicate condition
    let condition = PredicateCondition {
        kind: Some(Kind::Value(Predicate {
            kind: Some(PredKind::Equals(ahnlich_types::predicates::Equals {
                key: "filename".to_string(),
                value: Some(MetadataValue {
                    value: Some(Value::RawString("photo.jpg".to_string())),
                }),
            })),
        })),
    };

    // New metadata to merge
    let new_value = Some(StoreValue {
        value: HashMap::from_iter([(
            "tags".into(),
            MetadataValue {
                value: Some(Value::RawString("cat,outdoors".into())),
            },
        )]),
    });

    let params = Upsert {
        store: "images".to_string(),
        schema: Some("media".to_string()),
        condition: Some(condition),
        new_input: None, // Optional: new image/text to re-embed
        new_value,
        preprocess_action: PreprocessAction::NoPreprocessing as i32,
        execution_provider: None,
        model_params: HashMap::new(),
    };

    // Run the upsert command
    let res = client.upsert(params, None).await?;
    println!("Upsert result: {:?}", res.upsert);

    Ok(())
}

CLI

$UPSERT VALUE {status: published} IN my_store WHERE (id = 123)

Response

A Set response with upsert counts (e.g. inserted: 0, updated: 1).

Related

• Insert data
• Get by predicate