Ingest records into a stream
Ingests a batch of records into a stream. Requires streamRecordsAcl:WRITE and dataModelsAcl:READ. See Ingest records.
POST
/
streams
/
{streamId}
/
records
Python SDK
from cognite.client.data_classes.data_modeling.records import RecordWrite, RecordContainerId, RecordSource
# 1712000000000 = 2024-04-01T19:33:20Z
client.data_modeling.records.ingest(
RecordWrite(
space="my-space",
external_id="rec-1",
sources=[
RecordSource(
source=RecordContainerId(space="my-space", external_id="my-container"),
properties={
"equipment_id": "PUMP-001",
"location": "Building-A",
"temperature": 22.5,
"recorded_at": 1712000000000,
},
)
],
),
stream_id="my-stream",
)curl --request POST \
--url https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--data '
{
"items": [
{
"space": "my-space",
"externalId": "rec-1",
"sources": [
{
"source": {
"type": "container",
"space": "my-space",
"externalId": "my-container"
},
"properties": {
"equipment_id": "PUMP-001",
"location": "Building-A",
"temperature": 22.5,
"recorded_at": 1712000000000
}
}
]
},
{
"space": "my-space",
"externalId": "rec-2",
"sources": [
{
"source": {
"type": "view",
"space": "my-space",
"externalId": "my-view",
"version": "v1"
},
"properties": {
"equipmentId": "PUMP-002",
"temperature": 23.1
}
}
]
}
]
}
'const options = {
method: 'POST',
headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
body: JSON.stringify({
items: [
{
space: 'my-space',
externalId: 'rec-1',
sources: [
{
source: {type: 'container', space: 'my-space', externalId: 'my-container'},
properties: {
equipment_id: 'PUMP-001',
location: 'Building-A',
temperature: 22.5,
recorded_at: 1712000000000
}
}
]
},
{
space: 'my-space',
externalId: 'rec-2',
sources: [
{
source: {type: 'view', space: 'my-space', externalId: 'my-view', version: 'v1'},
properties: {equipmentId: 'PUMP-002', temperature: 23.1}
}
]
}
]
})
};
fetch('https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records', 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",
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' => 'my-space',
'externalId' => 'rec-1',
'sources' => [
[
'source' => [
'type' => 'container',
'space' => 'my-space',
'externalId' => 'my-container'
],
'properties' => [
'equipment_id' => 'PUMP-001',
'location' => 'Building-A',
'temperature' => 22.5,
'recorded_at' => 1712000000000
]
]
]
],
[
'space' => 'my-space',
'externalId' => 'rec-2',
'sources' => [
[
'source' => [
'type' => 'view',
'space' => 'my-space',
'externalId' => 'my-view',
'version' => 'v1'
],
'properties' => [
'equipmentId' => 'PUMP-002',
'temperature' => 23.1
]
]
]
]
]
]),
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"
payload := strings.NewReader("{\n \"items\": [\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-1\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container\",\n \"space\": \"my-space\",\n \"externalId\": \"my-container\"\n },\n \"properties\": {\n \"equipment_id\": \"PUMP-001\",\n \"location\": \"Building-A\",\n \"temperature\": 22.5,\n \"recorded_at\": 1712000000000\n }\n }\n ]\n },\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-2\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"view\",\n \"space\": \"my-space\",\n \"externalId\": \"my-view\",\n \"version\": \"v1\"\n },\n \"properties\": {\n \"equipmentId\": \"PUMP-002\",\n \"temperature\": 23.1\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")
.header("Authorization", "Bearer <token>")
.header("Content-Type", "application/json")
.body("{\n \"items\": [\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-1\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container\",\n \"space\": \"my-space\",\n \"externalId\": \"my-container\"\n },\n \"properties\": {\n \"equipment_id\": \"PUMP-001\",\n \"location\": \"Building-A\",\n \"temperature\": 22.5,\n \"recorded_at\": 1712000000000\n }\n }\n ]\n },\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-2\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"view\",\n \"space\": \"my-space\",\n \"externalId\": \"my-view\",\n \"version\": \"v1\"\n },\n \"properties\": {\n \"equipmentId\": \"PUMP-002\",\n \"temperature\": 23.1\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")
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\": \"my-space\",\n \"externalId\": \"rec-1\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container\",\n \"space\": \"my-space\",\n \"externalId\": \"my-container\"\n },\n \"properties\": {\n \"equipment_id\": \"PUMP-001\",\n \"location\": \"Building-A\",\n \"temperature\": 22.5,\n \"recorded_at\": 1712000000000\n }\n }\n ]\n },\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-2\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"view\",\n \"space\": \"my-space\",\n \"externalId\": \"my-view\",\n \"version\": \"v1\"\n },\n \"properties\": {\n \"equipmentId\": \"PUMP-002\",\n \"temperature\": 23.1\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.
Scopes
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 ingest 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.
⌘I
Python SDK
from cognite.client.data_classes.data_modeling.records import RecordWrite, RecordContainerId, RecordSource
# 1712000000000 = 2024-04-01T19:33:20Z
client.data_modeling.records.ingest(
RecordWrite(
space="my-space",
external_id="rec-1",
sources=[
RecordSource(
source=RecordContainerId(space="my-space", external_id="my-container"),
properties={
"equipment_id": "PUMP-001",
"location": "Building-A",
"temperature": 22.5,
"recorded_at": 1712000000000,
},
)
],
),
stream_id="my-stream",
)curl --request POST \
--url https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--data '
{
"items": [
{
"space": "my-space",
"externalId": "rec-1",
"sources": [
{
"source": {
"type": "container",
"space": "my-space",
"externalId": "my-container"
},
"properties": {
"equipment_id": "PUMP-001",
"location": "Building-A",
"temperature": 22.5,
"recorded_at": 1712000000000
}
}
]
},
{
"space": "my-space",
"externalId": "rec-2",
"sources": [
{
"source": {
"type": "view",
"space": "my-space",
"externalId": "my-view",
"version": "v1"
},
"properties": {
"equipmentId": "PUMP-002",
"temperature": 23.1
}
}
]
}
]
}
'const options = {
method: 'POST',
headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
body: JSON.stringify({
items: [
{
space: 'my-space',
externalId: 'rec-1',
sources: [
{
source: {type: 'container', space: 'my-space', externalId: 'my-container'},
properties: {
equipment_id: 'PUMP-001',
location: 'Building-A',
temperature: 22.5,
recorded_at: 1712000000000
}
}
]
},
{
space: 'my-space',
externalId: 'rec-2',
sources: [
{
source: {type: 'view', space: 'my-space', externalId: 'my-view', version: 'v1'},
properties: {equipmentId: 'PUMP-002', temperature: 23.1}
}
]
}
]
})
};
fetch('https://{cluster}.cognitedata.com/api/v1/projects/{project}/streams/{streamId}/records', 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",
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' => 'my-space',
'externalId' => 'rec-1',
'sources' => [
[
'source' => [
'type' => 'container',
'space' => 'my-space',
'externalId' => 'my-container'
],
'properties' => [
'equipment_id' => 'PUMP-001',
'location' => 'Building-A',
'temperature' => 22.5,
'recorded_at' => 1712000000000
]
]
]
],
[
'space' => 'my-space',
'externalId' => 'rec-2',
'sources' => [
[
'source' => [
'type' => 'view',
'space' => 'my-space',
'externalId' => 'my-view',
'version' => 'v1'
],
'properties' => [
'equipmentId' => 'PUMP-002',
'temperature' => 23.1
]
]
]
]
]
]),
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"
payload := strings.NewReader("{\n \"items\": [\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-1\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container\",\n \"space\": \"my-space\",\n \"externalId\": \"my-container\"\n },\n \"properties\": {\n \"equipment_id\": \"PUMP-001\",\n \"location\": \"Building-A\",\n \"temperature\": 22.5,\n \"recorded_at\": 1712000000000\n }\n }\n ]\n },\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-2\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"view\",\n \"space\": \"my-space\",\n \"externalId\": \"my-view\",\n \"version\": \"v1\"\n },\n \"properties\": {\n \"equipmentId\": \"PUMP-002\",\n \"temperature\": 23.1\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")
.header("Authorization", "Bearer <token>")
.header("Content-Type", "application/json")
.body("{\n \"items\": [\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-1\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container\",\n \"space\": \"my-space\",\n \"externalId\": \"my-container\"\n },\n \"properties\": {\n \"equipment_id\": \"PUMP-001\",\n \"location\": \"Building-A\",\n \"temperature\": 22.5,\n \"recorded_at\": 1712000000000\n }\n }\n ]\n },\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-2\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"view\",\n \"space\": \"my-space\",\n \"externalId\": \"my-view\",\n \"version\": \"v1\"\n },\n \"properties\": {\n \"equipmentId\": \"PUMP-002\",\n \"temperature\": 23.1\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")
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\": \"my-space\",\n \"externalId\": \"rec-1\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"container\",\n \"space\": \"my-space\",\n \"externalId\": \"my-container\"\n },\n \"properties\": {\n \"equipment_id\": \"PUMP-001\",\n \"location\": \"Building-A\",\n \"temperature\": 22.5,\n \"recorded_at\": 1712000000000\n }\n }\n ]\n },\n {\n \"space\": \"my-space\",\n \"externalId\": \"rec-2\",\n \"sources\": [\n {\n \"source\": {\n \"type\": \"view\",\n \"space\": \"my-space\",\n \"externalId\": \"my-view\",\n \"version\": \"v1\"\n },\n \"properties\": {\n \"equipmentId\": \"PUMP-002\",\n \"temperature\": 23.1\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>"
}
]
}
}