Skip to content

Pyo3: Type signatures for all exposed functions #167

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ReinierMaas wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Pyo3: Type signatures for all exposed functions #167

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion libs/opsqueue_python/src/async_util.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
14 changes: 8 additions & 6 deletions libs/opsqueue_python/src/common.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)?;
Expand Down Expand Up @@ -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 })
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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
Expand All @@ -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| {
Expand Down Expand Up @@ -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()
Expand 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)?;
Expand Down
10 changes: 6 additions & 4 deletions libs/opsqueue_python/src/consumer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<'_>,
Expand All @@ -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<'_>,
Expand All @@ -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<'_>,
Expand All @@ -165,6 +166,7 @@ impl ConsumerClient {
}

#[allow(clippy::type_complexity)]
#[pyo3(signature = (strategy, fun))]
pub fn run_per_chunk(
&self,
strategy: &Strategy,
Expand Down Expand Up @@ -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>,
Expand Down
24 changes: 15 additions & 9 deletions libs/opsqueue_python/src/producer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<'_>,
Expand Down Expand Up @@ -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<'_>,
Expand All @@ -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<'_>,
Expand All @@ -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<'_>,
Expand Down Expand Up @@ -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
Expand All @@ -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<'_>,
Expand Down Expand Up @@ -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<'_>,
Expand Down Expand Up @@ -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<'_>,
Expand Down Expand Up @@ -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<'_>,
Expand All @@ -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>,
Expand Down
Loading