Pyo3: Type signatures

Author: ReinierMaas

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<T, F>(py: Python<'_>, fut: F) -> PyResult<Bound<'_, PyAny>>
+pub(crate) fn future_into_py<T, F>(py: Python<'_>, fut: F) -> PyResult<Bound<'_, PyAny>>
 where
     F: Future<Output = PyResult<T>> + Send + 'static,
     T: for<'py> IntoPyObject<'py> + Send + 'static,

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<Self, TryFromIntError> {
         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<Self, TryFromIntError> {
         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<T, E>(
+pub(crate) async fn run_unless_interrupted<T, E>(
     future: impl IntoFuture<Output = Result<T, E>>,
 ) -> Result<T, E>
 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<tokio::runtime::Runtime> {
+pub(crate) fn start_runtime() -> Arc<tokio::runtime::Runtime> {
     let runtime = tokio::runtime::Builder::new_multi_thread()
         .worker_threads(1)
         .enable_all()
@@ -635,7 +637,7 @@ pub fn start_runtime() -> Arc<tokio::runtime::Runtime> {
 /// # 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<String> = (|| {
             let traceback = err.traceback(py)?;

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<String>,

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,25 +250,24 @@ 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<chunk::Content>,
         metadata: Option<submission::Metadata>,
+        strategic_metadata: Option<StrategicMetadataMap>,
         chunk_size: Option<u64>,
         otel_trace_carrier: CarrierMap,
     ) -> CPyResult<SubmissionId, E<FatalPythonException, InternalProducerClientError>> {
-        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())),
                 chunk_contents: ChunkContents::Direct {
                     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>,