feat: implement StreamCreateForecasts

README.md

@@ -617,6 +617,13 @@ Forecaster represents a generative source of predicted values.
 | Field | Type | Label | Description |
 | ----- | ---- | ----- | ----------- |
 | observer_uuid | [string](#string) |  |  || observer_name | [string](#string) |  |  |
+</details><a name="ocf-dp-StreamCreateForecastsResponse"></a>
+<details><summary>StreamCreateForecastsResponse</summary>
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| forecast_uuids | [string](#string) | repeated | A list of the UUIDs generated for the successfully created forecasts. |
 </details><a name="ocf-dp-StreamForecastDataRequest"></a>
 <details><summary>StreamForecastDataRequest</summary>
 
@@ -886,6 +893,15 @@ Useful for analytics and performance monitoring.
 
 _[StreamForecastDataRequest](#ocf-dp-StreamForecastDataRequest) / [StreamForecastDataResponse](#ocf-dp-StreamForecastDataResponse) stream_
 
+<a name="StreamCreateForecasts"></a>
+
+#### StreamCreateForecasts
+
+StreamCreateForecasts allows for efficient batch creation of multiple forecasts and their values.
+Note: This method is executed in a single transaction. To prevent resource exhaustion, a maximum of 5000 forecasts can be sent per stream. Exceeding this limit will abort the stream and roll back all inserts.
+
+_[CreateForecastRequest](#ocf-dp-CreateForecastRequest) stream / [StreamCreateForecastsResponse](#ocf-dp-StreamCreateForecastsResponse)_
+
 
 
 

examples/python-notebook/example.py

@@ -20,7 +20,6 @@
 from ocf.dp.dp import common_pb2
 from ocf.dp.dp_data import messages_pb2, service_pb2_grpc
 import pandas as pd
-import xarray as xr
 import datetime as dt
 
 
@@ -77,7 +76,7 @@ async def main() -> None:
         end_time = gfreq_response.values[-1].target_timestamp_utc.ToDatetime(tzinfo=dt.UTC)
         print(f"\tReceived {len(gfreq_response.values)} forecast points from {start_time} to {end_time}")
 
-        print(f":: -> Converting response to a dataframe")
+        print(":: -> Converting response to a dataframe")
         # preserving_proto_field_name prevents conversion to lowerCamelCase.
         # always_print_fields_with_no_presence ensures all fields are present in the dict, even if they have no value in the protobuf.
         df = pd.DataFrame.from_dict([
@@ -95,14 +94,14 @@ async def main() -> None:
         ).drop(["p50_value_fraction", "p10", "p90"], axis=1)
         print(df.head())
 
-        print(f":: Getting 'ground truths' for the same location and time period")
+        print(":: Getting 'ground truths' for the same location and time period")
 
-        print(f":: -> Getting an observer")
+        print(":: -> Getting an observer")
         loresp = await dpc.ListObservers(messages_pb2.ListObserversRequest())
         observer = next(o for o in loresp.observers if "pvlive" in o.observer_name)
         print(f"\t{observer.observer_name=}")
 
-        print(f":: -> Getting the ground truth for the UK national location")
+        print(":: -> Getting the ground truth for the UK national location")
         gtreq = messages_pb2.GetObservationsAsTimeseriesRequest(
             location_uuid=uk_location.location_uuid,
             energy_source=common_pb2.EnergySource.ENERGY_SOURCE_SOLAR,

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/rs/zerolog v1.34.0
 	github.com/stretchr/testify v1.11.1
 	github.com/testcontainers/testcontainers-go v0.40.0
+	golang.org/x/sync v0.20.0
 	google.golang.org/grpc v1.79.2
 	google.golang.org/protobuf v1.36.11
 )
@@ -80,7 +81,6 @@ require (
 	golang.org/x/crypto v0.48.0 // indirect
 	golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa // indirect
 	golang.org/x/net v0.51.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/text v0.34.0 // indirect
 	golang.org/x/time v0.12.0 // indirect

internal/server/dummy/dataserverimpl.go

@@ -18,6 +18,8 @@ import (
 
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/structpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
@@ -723,5 +725,12 @@ func (d *DataPlatformDataServiceServerImpl) UpdateForecaster(
 	}, nil
 }
 
+// StreamCreateForecasts implements dp.DataPlatformDataServiceServer.
+func (s *DataPlatformDataServiceServerImpl) StreamCreateForecasts(
+	stream grpc.ClientStreamingServer[pb.CreateForecastRequest, pb.StreamCreateForecastsResponse],
+) error {
+	return status.Errorf(codes.Unimplemented, "method StreamCreateForecasts not implemented")
+}
+
 // Compile-time check to ensure the interface is implemented fully.
 var _ pb.DataPlatformDataServiceServer = (*DataPlatformDataServiceServerImpl)(nil)