You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Add a Python client interface to the MLDP Ingestion Service (DpIngestionService) — the full ingestion surface, not just the unary call. The project currently has no real ingestion client outside the dp-desktop-app JavaFX demo, so this fills a genuine product gap: a facility standing up MLDP needs a straightforward Python path to get data in, not only to query it back out.
This is a peer sub-issue to the query interface (#7) and follows the same one-clean-sub-issue-per-PR pattern established by #5 (PV metadata) and #9 (machine config): params classes → _build_*_request() → _send_*() with three-tier error handling → *ApiResult, exposed via MldpClient (client.ingestion / the existing ingestion_client).
Why full-surface (not unary-only)
The bulk of the design work is the shared IngestDataRequest payload, which is identical across every ingestion RPC:
common.DataFrame = dataTimestamps (a DataTimestamps oneof: explicit timestampListor a samplingClock = startTime + periodNanos + count) + a large set of typed column arms (dataColumns generic DataValue, plus doubleColumns/int64Columns/stringColumns/imageColumns/structColumns/the array-column variants, etc.).
Response: IngestDataResponse with exceptionalResult | ackResult { numRows, numColumns }.
Because unary and streaming wrap the same payload + params model, unary-only would still require building 100% of that substrate and would risk a breaking reshape of the public params API when streaming lands later (the same "renaming later is breaking" argument that drove QueryParams/client.query in #7). Building the full surface once validates the shared types against every consumer up front.
Scope (phased internally; one PR)
Phase 1 — shared payload/params model + unary ingestData + unit tests
User-friendly params for a DataFrame: Python-native columns + a timestamps spec that accepts either an explicit timestamp list or a sampling clock (start + period + count), reusing to_timestamp().
ingestData() unary wrapper + IngestDataApiResult (surfacing ackResult.numRows/numColumns and the three-tier error handling).
This is the foundation, fully exercised by the unary path.
Phase 2 — streaming
ingestDataStream() (client-streaming / stream_unary) and ingestDataBidiStream() (stream_stream), built on the Phase 1 payload model.
Iterator-based ergonomics symmetric with the query client's streaming.
Phase 3 — closed-loop integration test + docs
Live ingest→query-back round-trip against the local ecosystem, asserting exact value round-trip, trimmed half-open range, dense column/timestamp alignment, and paging.
This also retroactively completes the query integration test in interface to v2 time-series data query API #7, which is intentionally deferred pending this client (a closed-loop query test needs a way to put known data in; assuming a test DB is pre-populated was explicitly rejected).
README + CLAUDE.md quick-start ("getting data in") section.
Notes / out of scope for now
queryRequestStatus, subscribeData — separate concerns; not part of the core ingest path. Can be filed separately if wanted.
Summary
Add a Python client interface to the MLDP Ingestion Service (
DpIngestionService) — the full ingestion surface, not just the unary call. The project currently has no real ingestion client outside thedp-desktop-appJavaFX demo, so this fills a genuine product gap: a facility standing up MLDP needs a straightforward Python path to get data in, not only to query it back out.This is a peer sub-issue to the query interface (#7) and follows the same one-clean-sub-issue-per-PR pattern established by #5 (PV metadata) and #9 (machine config): params classes →
_build_*_request()→_send_*()with three-tier error handling →*ApiResult, exposed viaMldpClient(client.ingestion/ the existingingestion_client).Why full-surface (not unary-only)
The bulk of the design work is the shared
IngestDataRequestpayload, which is identical across every ingestion RPC:IngestDataRequest { providerId, clientRequestId, ingestionDataFrame: common.DataFrame }common.DataFrame=dataTimestamps(aDataTimestampsoneof: explicittimestampListor asamplingClock= startTime + periodNanos + count) + a large set of typed column arms (dataColumnsgenericDataValue, plusdoubleColumns/int64Columns/stringColumns/imageColumns/structColumns/the array-column variants, etc.).IngestDataResponsewithexceptionalResult|ackResult { numRows, numColumns }.Because unary and streaming wrap the same payload + params model, unary-only would still require building 100% of that substrate and would risk a breaking reshape of the public params API when streaming lands later (the same "renaming later is breaking" argument that drove
QueryParams/client.queryin #7). Building the full surface once validates the shared types against every consumer up front.Scope (phased internally; one PR)
Phase 1 — shared payload/params model + unary
ingestData+ unit testsDataFrame: Python-native columns + a timestamps spec that accepts either an explicit timestamp list or a sampling clock (start + period + count), reusingto_timestamp().ingestData()unary wrapper +IngestDataApiResult(surfacingackResult.numRows/numColumnsand the three-tier error handling).Phase 2 — streaming
ingestDataStream()(client-streaming /stream_unary) andingestDataBidiStream()(stream_stream), built on the Phase 1 payload model.Phase 3 — closed-loop integration test + docs
Notes / out of scope for now
queryRequestStatus,subscribeData— separate concerns; not part of the core ingest path. Can be filed separately if wanted.serializedDataColumnson the ingest side mirror the query-side deferral (interface to v2 time-series data query API #7 Q3) — default to the dense typed columns.Dependency
Blocks the query integration test in #7 (#7). Part of epic #10.