From 2ae254785013e70b557eaabecf077cdb3e03da84 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Thu, 13 Aug 2026 18:19:35 +0200 Subject: [PATCH] Pyo3: Type signatures --- libs/opsqueue_python/src/async_util.rs | 2 +- libs/opsqueue_python/src/common.rs | 14 ++++++++------ libs/opsqueue_python/src/consumer.rs | 10 ++++++---- libs/opsqueue_python/src/producer.rs | 24 +++++++++++++++--------- 4 files changed, 30 insertions(+), 20 deletions(-) diff --git a/libs/opsqueue_python/src/async_util.rs b/libs/opsqueue_python/src/async_util.rs index bdf78264..f6203d68 100644 --- a/libs/opsqueue_python/src/async_util.rs +++ b/libs/opsqueue_python/src/async_util.rs @@ -36,7 +36,7 @@ where } /// Version of `future_into_py` that uses the `TokioRuntimeThatIsInScope` -pub fn future_into_py(py: Python<'_>, fut: F) -> PyResult> +pub(crate) fn future_into_py(py: Python<'_>, fut: F) -> PyResult> where F: Future> + Send + 'static, T: for<'py> IntoPyObject<'py> + Send + 'static, diff --git a/libs/opsqueue_python/src/common.rs b/libs/opsqueue_python/src/common.rs index b0fbc0b3..0402c85a 100644 --- a/libs/opsqueue_python/src/common.rs +++ b/libs/opsqueue_python/src/common.rs @@ -33,6 +33,7 @@ pub struct SubmissionId { #[pymethods] impl SubmissionId { #[new] + #[pyo3(signature = (id))] fn new(id: u64) -> CPyResult { let _is_inner_valid = opsqueue::common::submission::SubmissionId::try_from(id).map_err(CError)?; @@ -69,6 +70,7 @@ pub struct ChunkIndex { #[pymethods] impl ChunkIndex { #[new] + #[pyo3(signature = (id))] fn new(id: u64) -> CPyResult { let _is_inner_valid = opsqueue::common::chunk::ChunkIndex::new(id).map_err(CError)?; Ok(ChunkIndex { id }) @@ -230,7 +232,7 @@ impl Chunk { /// # Errors /// /// Returns an error if fetching chunk bytes from object storage fails. - pub async fn from_internal( + pub(crate) async fn from_internal( c: chunk::Chunk, s: submission::Submission, object_store_client: &ObjectStoreClient, @@ -285,7 +287,7 @@ pub struct ChunkFailed { impl ChunkFailed { #[must_use] - pub fn from_internal(c: chunk::ChunkFailed, _s: &submission::SubmissionFailed) -> Self { + pub(crate) fn from_internal(c: chunk::ChunkFailed, _s: &submission::SubmissionFailed) -> Self { ChunkFailed { submission_id: c.submission_id.into(), chunk_index: c.chunk_index.into(), @@ -561,7 +563,7 @@ impl SubmissionNotCancellable { /// /// Returns the underlying future error, or a fatal Python exception when /// an interrupt signal is detected. -pub async fn run_unless_interrupted( +pub(crate) async fn run_unless_interrupted( future: impl IntoFuture>, ) -> Result where @@ -573,7 +575,7 @@ where } } -pub async fn check_signals_in_background() -> FatalPythonException { +async fn check_signals_in_background() -> FatalPythonException { loop { tokio::time::sleep(SIGNAL_CHECK_INTERVAL).await; let res = Python::attach(|py| { @@ -615,7 +617,7 @@ pub async fn check_signals_in_background() -> FatalPythonException { /// /// Panics if creating the Tokio runtime fails. #[must_use] -pub fn start_runtime() -> Arc { +pub(crate) fn start_runtime() -> Arc { let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(1) .enable_all() @@ -635,7 +637,7 @@ pub fn start_runtime() -> Arc { /// # Panics /// /// Panics if formatting a Python traceback fails. -pub fn format_pyerr(err: &PyErr) -> String { +pub(crate) fn format_pyerr(err: &PyErr) -> String { Python::attach(|py| { let msg: Option = (|| { let traceback = err.traceback(py)?; diff --git a/libs/opsqueue_python/src/consumer.rs b/libs/opsqueue_python/src/consumer.rs index b31dd9fd..182533f6 100644 --- a/libs/opsqueue_python/src/consumer.rs +++ b/libs/opsqueue_python/src/consumer.rs @@ -91,12 +91,13 @@ impl ConsumerClient { ) } - #[allow(clippy::type_complexity)] /// Reserve up to `max` chunks from the queue. /// /// # Errors /// /// Returns an error if reservation fails or if chunk bytes cannot be retrieved. + #[allow(clippy::type_complexity)] + #[pyo3(signature = (max, strategy))] pub fn reserve_chunks( &self, py: Python<'_>, @@ -114,12 +115,12 @@ impl ConsumerClient { py.detach(|| self.reserve_chunks_gilless(max, strategy.into())) } - #[pyo3(signature = (submission_id, submission_prefix, chunk_index, output_content))] /// Complete a chunk and optionally upload output content to object storage. /// /// # Errors /// /// Returns an error if upload or completion request fails. + #[pyo3(signature = (submission_id, submission_prefix, chunk_index, output_content))] pub fn complete_chunk( &self, py: Python<'_>, @@ -145,12 +146,12 @@ impl ConsumerClient { }) } - #[pyo3(signature = (submission_id, submission_prefix, chunk_index, failure))] /// Mark a chunk as failed. /// /// # Errors /// /// Returns an error if the failure report cannot be submitted. + #[pyo3(signature = (submission_id, submission_prefix, chunk_index, failure))] pub fn fail_chunk( &self, py: Python<'_>, @@ -165,6 +166,7 @@ impl ConsumerClient { } #[allow(clippy::type_complexity)] + #[pyo3(signature = (strategy, fun))] pub fn run_per_chunk( &self, strategy: &Strategy, @@ -316,7 +318,7 @@ impl ConsumerClient { /// # Errors /// /// Returns an error if the failure report cannot be submitted. - pub fn fail_chunk_gilless( + fn fail_chunk_gilless( &self, submission_id: SubmissionId, _submission_prefix: Option, diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs index e3167a9b..3b917ada 100644 --- a/libs/opsqueue_python/src/producer.rs +++ b/libs/opsqueue_python/src/producer.rs @@ -135,6 +135,7 @@ impl ProducerClient { /// /// Returns an error if the submission cannot be cancelled or if the request fails. #[allow(clippy::result_large_err, clippy::type_complexity)] + #[pyo3(signature = (id))] pub fn cancel_submission( &self, py: Python<'_>, @@ -168,6 +169,7 @@ impl ProducerClient { /// # Errors /// /// Returns an error if contacting the server fails. + #[pyo3(signature = (id))] pub fn get_submission_status( &self, py: Python<'_>, @@ -193,6 +195,7 @@ impl ProducerClient { /// # Errors /// /// Returns an error if contacting the server fails. + #[pyo3(signature = (prefix))] pub fn lookup_submission_id_by_prefix( &self, py: Python<'_>, @@ -216,6 +219,7 @@ impl ProducerClient { /// /// Returns an error if too many submissions match or if the request fails. #[allow(clippy::needless_pass_by_value)] + #[pyo3(signature = (strategic_metadata))] pub fn lookup_submission_ids_by_strategic_metadata( &self, py: Python<'_>, @@ -246,17 +250,16 @@ impl ProducerClient { /// # Errors /// /// Returns an error if submission insertion fails. - #[pyo3(signature = (chunk_contents, metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] + #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] pub fn insert_submission_direct( &self, py: Python<'_>, chunk_contents: Vec, metadata: Option, + strategic_metadata: Option, chunk_size: Option, otel_trace_carrier: CarrierMap, ) -> CPyResult> { - let strategic_metadata = std::collections::HashMap::default(); - py.detach(|| { let submission = opsqueue::producer::InsertSubmission { chunk_size: chunk_size.map(|n| chunk::ChunkSize(n.cast_signed())), @@ -264,7 +267,7 @@ impl ProducerClient { contents: chunk_contents, }, metadata, - strategic_metadata, + strategic_metadata: strategic_metadata.unwrap_or_default(), }; self.block_unless_interrupted(async move { self.client @@ -276,13 +279,13 @@ impl ProducerClient { }) } - #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] - #[allow(clippy::type_complexity)] /// Insert submission chunks via object storage and enqueue the submission. /// /// # Errors /// /// Returns an error if chunk upload or submission insertion fails. + #[allow(clippy::type_complexity)] + #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] pub fn insert_submission_chunks( &self, py: Python<'_>, @@ -343,12 +346,13 @@ impl ProducerClient { }) } - #[allow(clippy::result_large_err, clippy::type_complexity)] /// Try streaming completed submission chunks without waiting. /// /// # Errors /// /// Returns an error if the submission is incomplete/failed or if the request fails. + #[allow(clippy::result_large_err, clippy::type_complexity)] + #[pyo3(signature = (id))] pub fn try_stream_completed_submission_chunks( &self, py: Python<'_>, @@ -376,13 +380,13 @@ impl ProducerClient { }) } - #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] - #[allow(clippy::result_large_err, clippy::type_complexity)] /// Submit chunks and then stream the completed output chunks. /// /// # Errors /// /// Returns an error if upload, submission creation, or streaming fails. + #[allow(clippy::result_large_err, clippy::type_complexity)] + #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] pub fn run_submission_chunks( &self, py: Python<'_>, @@ -438,6 +442,7 @@ impl ProducerClient { /// /// Returns an error if polling or output streaming fails. #[allow(clippy::result_large_err, clippy::type_complexity)] + #[pyo3(signature = (submission_id))] pub fn blocking_stream_completed_submission_chunks( &self, py: Python<'_>, @@ -462,6 +467,7 @@ impl ProducerClient { /// # Errors /// /// Returns a Python error if creating the awaitable fails. + #[pyo3(signature = (submission_id))] pub fn async_stream_completed_submission_chunks<'p>( &self, py: Python<'p>,