Upsert (create or update) records into a stream
curl --request POST \
--url https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--data '
{
"items": [
{
"space": "mySpace",
"externalId": "some_id",
"sources": [
{
"source": {
"type": "container,",
"space": "mySpace,",
"externalId": "myContainer"
},
"properties": {
"someStringProperty": "someStringValue",
"someDirectRelation": {
"space": "mySpace",
"externalId": "someNode"
},
"someIntArrayProperty": [
1,
2,
3,
4
]
}
}
]
}
]
}
'import requests
url = "https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert"
payload = { "items": [
{
"space": "mySpace",
"externalId": "some_id",
"sources": [
{
"source": {
"type": "container,",
"space": "mySpace,",
"externalId": "myContainer"
},
"properties": {
"someStringProperty": "someStringValue",
"someDirectRelation": {
"space": "mySpace",
"externalId": "someNode"
},
"someIntArrayProperty": [1, 2, 3, 4]
}
}
]
}
] }
headers = {
"Authorization": "Bearer <token>",
"Content-Type": "application/json"
}
response = requests.post(url, json=payload, headers=headers)
print(response.text)const options = {
method: 'POST',
headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
body: JSON.stringify({
items: [
{
space: 'mySpace',
externalId: 'some_id',
sources: [
{
source: {type: 'container,', space: 'mySpace,', externalId: 'myContainer'},
properties: {
someStringProperty: 'someStringValue',
someDirectRelation: {space: 'mySpace', externalId: 'someNode'},
someIntArrayProperty: [1, 2, 3, 4]
}
}
]
}
]
})
};
fetch('https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert', options)
.then(res => res.json())
.then(res => console.log(res))
.catch(err => console.error(err));<?php
$curl = curl_init();
curl_setopt_array($curl, [
CURLOPT_URL => "https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert",
CURLOPT_RETURNTRANSFER => true,
CURLOPT_ENCODING => "",
CURLOPT_MAXREDIRS => 10,
CURLOPT_TIMEOUT => 30,
CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
CURLOPT_CUSTOMREQUEST => "POST",
CURLOPT_POSTFIELDS => json_encode([
'items' => [
[
'space' => 'mySpace',
'externalId' => 'some_id',
'sources' => [
[
'source' => [
'type' => 'container,',
'space' => 'mySpace,',
'externalId' => 'myContainer'
],
'properties' => [
'someStringProperty' => 'someStringValue',
'someDirectRelation' => [
'space' => 'mySpace',
'externalId' => 'someNode'
],
'someIntArrayProperty' => [
1,
2,
3,
4
]
]
]
]
]
]
]),
CURLOPT_HTTPHEADER => [
"Authorization: Bearer <token>",
"Content-Type: application/json"
],
]);
$response = curl_exec($curl);
$err = curl_error($curl);
curl_close($curl);
if ($err) {
echo "cURL Error #:" . $err;
} else {
echo $response;
}package main
import (
"fmt"
"strings"
"net/http"
"io"
)
func main() {
url := "https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert"
payload := strings.NewReader("{\n \"items\": [\n {\n \"space\": \"mySpace\",\n \"externalId\": \"some_id\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container,\",\n \"space\": \"mySpace,\",\n \"externalId\": \"myContainer\"\n },\n \"properties\": {\n \"someStringProperty\": \"someStringValue\",\n \"someDirectRelation\": {\n \"space\": \"mySpace\",\n \"externalId\": \"someNode\"\n },\n \"someIntArrayProperty\": [\n 1,\n 2,\n 3,\n 4\n ]\n }\n }\n ]\n }\n ]\n}")
req, _ := http.NewRequest("POST", url, payload)
req.Header.Add("Authorization", "Bearer <token>")
req.Header.Add("Content-Type", "application/json")
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
body, _ := io.ReadAll(res.Body)
fmt.Println(string(body))
}HttpResponse<String> response = Unirest.post("https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert")
.header("Authorization", "Bearer <token>")
.header("Content-Type", "application/json")
.body("{\n \"items\": [\n {\n \"space\": \"mySpace\",\n \"externalId\": \"some_id\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container,\",\n \"space\": \"mySpace,\",\n \"externalId\": \"myContainer\"\n },\n \"properties\": {\n \"someStringProperty\": \"someStringValue\",\n \"someDirectRelation\": {\n \"space\": \"mySpace\",\n \"externalId\": \"someNode\"\n },\n \"someIntArrayProperty\": [\n 1,\n 2,\n 3,\n 4\n ]\n }\n }\n ]\n }\n ]\n}")
.asString();require 'uri'
require 'net/http'
url = URI("https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert")
http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true
request = Net::HTTP::Post.new(url)
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n \"items\": [\n {\n \"space\": \"mySpace\",\n \"externalId\": \"some_id\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container,\",\n \"space\": \"mySpace,\",\n \"externalId\": \"myContainer\"\n },\n \"properties\": {\n \"someStringProperty\": \"someStringValue\",\n \"someDirectRelation\": {\n \"space\": \"mySpace\",\n \"externalId\": \"someNode\"\n },\n \"someIntArrayProperty\": [\n 1,\n 2,\n 3,\n 4\n ]\n }\n }\n ]\n }\n ]\n}"
response = http.request(request)
puts response.read_body{}{
"error": {
"code": 401,
"message": "Could not authenticate.",
"missing": [
{}
],
"duplicated": [
{}
]
}
}{
"error": {
"code": 409,
"message": "<string>",
"partial": [
{
"space": "mySpace",
"externalId": "some_id",
"code": 409,
"message": "<string>"
}
]
}
}{
"error": {
"code": 422,
"message": "<string>",
"duplicated": [
{
"externalId": "mySpace.myRecord1"
}
]
}
}{
"error": {
"code": 409,
"message": "<string>",
"partial": [
{
"space": "mySpace",
"externalId": "some_id",
"code": 409,
"message": "<string>"
}
]
}
}{
"error": {
"code": 409,
"message": "<string>",
"partial": [
{
"space": "mySpace",
"externalId": "some_id",
"code": 409,
"message": "<string>"
}
]
}
}Upsert (create or update) records into a stream
Required capabilities:
streamRecordsAcl:WRITEdataModelsAcl:READ
Note: This endpoint is only available for mutable streams.
Before a user can upsert records, all referenced schema elements (Spaces, Containers, Properties) must be defined/created in Data Modeling.
To upsert records, the user must have capabilities to access (read) the referenced containers as well as capabilities to access (write) the referenced record space.
Batches of records are ingested or updated into a stream.
Under normal upsert operation, the record will be created or updated, and the properties defined in each of the containers in the sources array will be populated.
If record identifiers (space + externalId) are already in the stream, records are fully updated (no partial updates for this endpoint).
Otherwise, records are created.
Note: The maximum total request size is 10MB.
POST
/
streams
/
{streamId}
/
records
/
upsert
Upsert (create or update) records into a stream
curl --request POST \
--url https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--data '
{
"items": [
{
"space": "mySpace",
"externalId": "some_id",
"sources": [
{
"source": {
"type": "container,",
"space": "mySpace,",
"externalId": "myContainer"
},
"properties": {
"someStringProperty": "someStringValue",
"someDirectRelation": {
"space": "mySpace",
"externalId": "someNode"
},
"someIntArrayProperty": [
1,
2,
3,
4
]
}
}
]
}
]
}
'import requests
url = "https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert"
payload = { "items": [
{
"space": "mySpace",
"externalId": "some_id",
"sources": [
{
"source": {
"type": "container,",
"space": "mySpace,",
"externalId": "myContainer"
},
"properties": {
"someStringProperty": "someStringValue",
"someDirectRelation": {
"space": "mySpace",
"externalId": "someNode"
},
"someIntArrayProperty": [1, 2, 3, 4]
}
}
]
}
] }
headers = {
"Authorization": "Bearer <token>",
"Content-Type": "application/json"
}
response = requests.post(url, json=payload, headers=headers)
print(response.text)const options = {
method: 'POST',
headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
body: JSON.stringify({
items: [
{
space: 'mySpace',
externalId: 'some_id',
sources: [
{
source: {type: 'container,', space: 'mySpace,', externalId: 'myContainer'},
properties: {
someStringProperty: 'someStringValue',
someDirectRelation: {space: 'mySpace', externalId: 'someNode'},
someIntArrayProperty: [1, 2, 3, 4]
}
}
]
}
]
})
};
fetch('https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert', options)
.then(res => res.json())
.then(res => console.log(res))
.catch(err => console.error(err));<?php
$curl = curl_init();
curl_setopt_array($curl, [
CURLOPT_URL => "https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert",
CURLOPT_RETURNTRANSFER => true,
CURLOPT_ENCODING => "",
CURLOPT_MAXREDIRS => 10,
CURLOPT_TIMEOUT => 30,
CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
CURLOPT_CUSTOMREQUEST => "POST",
CURLOPT_POSTFIELDS => json_encode([
'items' => [
[
'space' => 'mySpace',
'externalId' => 'some_id',
'sources' => [
[
'source' => [
'type' => 'container,',
'space' => 'mySpace,',
'externalId' => 'myContainer'
],
'properties' => [
'someStringProperty' => 'someStringValue',
'someDirectRelation' => [
'space' => 'mySpace',
'externalId' => 'someNode'
],
'someIntArrayProperty' => [
1,
2,
3,
4
]
]
]
]
]
]
]),
CURLOPT_HTTPHEADER => [
"Authorization: Bearer <token>",
"Content-Type: application/json"
],
]);
$response = curl_exec($curl);
$err = curl_error($curl);
curl_close($curl);
if ($err) {
echo "cURL Error #:" . $err;
} else {
echo $response;
}package main
import (
"fmt"
"strings"
"net/http"
"io"
)
func main() {
url := "https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert"
payload := strings.NewReader("{\n \"items\": [\n {\n \"space\": \"mySpace\",\n \"externalId\": \"some_id\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container,\",\n \"space\": \"mySpace,\",\n \"externalId\": \"myContainer\"\n },\n \"properties\": {\n \"someStringProperty\": \"someStringValue\",\n \"someDirectRelation\": {\n \"space\": \"mySpace\",\n \"externalId\": \"someNode\"\n },\n \"someIntArrayProperty\": [\n 1,\n 2,\n 3,\n 4\n ]\n }\n }\n ]\n }\n ]\n}")
req, _ := http.NewRequest("POST", url, payload)
req.Header.Add("Authorization", "Bearer <token>")
req.Header.Add("Content-Type", "application/json")
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
body, _ := io.ReadAll(res.Body)
fmt.Println(string(body))
}HttpResponse<String> response = Unirest.post("https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert")
.header("Authorization", "Bearer <token>")
.header("Content-Type", "application/json")
.body("{\n \"items\": [\n {\n \"space\": \"mySpace\",\n \"externalId\": \"some_id\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container,\",\n \"space\": \"mySpace,\",\n \"externalId\": \"myContainer\"\n },\n \"properties\": {\n \"someStringProperty\": \"someStringValue\",\n \"someDirectRelation\": {\n \"space\": \"mySpace\",\n \"externalId\": \"someNode\"\n },\n \"someIntArrayProperty\": [\n 1,\n 2,\n 3,\n 4\n ]\n }\n }\n ]\n }\n ]\n}")
.asString();require 'uri'
require 'net/http'
url = URI("https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records/upsert")
http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true
request = Net::HTTP::Post.new(url)
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n \"items\": [\n {\n \"space\": \"mySpace\",\n \"externalId\": \"some_id\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container,\",\n \"space\": \"mySpace,\",\n \"externalId\": \"myContainer\"\n },\n \"properties\": {\n \"someStringProperty\": \"someStringValue\",\n \"someDirectRelation\": {\n \"space\": \"mySpace\",\n \"externalId\": \"someNode\"\n },\n \"someIntArrayProperty\": [\n 1,\n 2,\n 3,\n 4\n ]\n }\n }\n ]\n }\n ]\n}"
response = http.request(request)
puts response.read_body{}{
"error": {
"code": 401,
"message": "Could not authenticate.",
"missing": [
{}
],
"duplicated": [
{}
]
}
}{
"error": {
"code": 409,
"message": "<string>",
"partial": [
{
"space": "mySpace",
"externalId": "some_id",
"code": 409,
"message": "<string>"
}
]
}
}{
"error": {
"code": 422,
"message": "<string>",
"duplicated": [
{
"externalId": "mySpace.myRecord1"
}
]
}
}{
"error": {
"code": 409,
"message": "<string>",
"partial": [
{
"space": "mySpace",
"externalId": "some_id",
"code": 409,
"message": "<string>"
}
]
}
}{
"error": {
"code": 409,
"message": "<string>",
"partial": [
{
"space": "mySpace",
"externalId": "some_id",
"code": 409,
"message": "<string>"
}
]
}
}Authorizations
oidc-tokenoauth2-client-credentialsoauth2-open-industrial-dataoauth2-auth-code
Access token issued by the CDF project's configured identity provider. Access token must be an OpenID Connect token, and the project must be configured to accept OpenID Connect tokens. Use a header key of 'Authorization' with a value of 'Bearer $accesstoken'. The token can be obtained through any flow supported by the identity provider.
Path Parameters
An identifier of the stream where the records are stored.
Required string length:
1 - 100Pattern:
^[a-z]([a-z0-9_-]{0,98}[a-z0-9])?$Example:
"test1"
Body
application/json
Records to upsert into a stream.
List of records to write.
Required array length:
1 - 1000 elementsShow child attributes
Show child attributes
Response
An empty response is returned if all records from the request are successfully accepted to be created or updated.
Empty JSON object indicates all records are successfully accepted.
Last modified on July 10, 2026
Was this page helpful?
⌘I