aws-ssm-bridge 0.4.0

//! Python bindings for aws-ssm-bridge
//!
//! This module provides Python bindings using PyO3 for the aws-ssm-bridge library.

use pyo3::exceptions::{PyRuntimeError, PyValueError};
use pyo3::prelude::*;
use tracing_subscriber;

mod interactive;
mod session;

use session::{PyOutputStream, PySession, PySessionConfig, PySessionManager, PySessionType};

// ---------------------------------------------------------------------------
// Python exception hierarchy
// ---------------------------------------------------------------------------
//
// AwsSsmBridgeError
//   ├── SsmSessionError      (Error::Session)
//   ├── SsmProtocolError     (Error::Protocol)
//   ├── SsmTransportError    (Error::Transport)
//   ├── SsmAwsSdkError       (Error::AwsSdk)
//   ├── SsmTimeoutError      (Error::Timeout)
//   └── SsmCancelledError    (Error::Cancelled)
//
// Python callers import these via `from aws_ssm_bridge import AwsSsmBridgeError, ...`
//
// The `#[allow(missing_docs)]` on the inner module suppresses the lint that
// fires on structs generated by `pyo3::create_exception!`, which does not
// accept doc-comment syntax itself.
#[allow(missing_docs)]
mod exceptions {
    pyo3::create_exception!(_internal, AwsSsmBridgeError, pyo3::exceptions::PyException);
    pyo3::create_exception!(_internal, SsmSessionError, AwsSsmBridgeError);
    pyo3::create_exception!(_internal, SsmProtocolError, AwsSsmBridgeError);
    pyo3::create_exception!(_internal, SsmTransportError, AwsSsmBridgeError);
    pyo3::create_exception!(_internal, SsmAwsSdkError, AwsSsmBridgeError);
    pyo3::create_exception!(_internal, SsmTimeoutError, AwsSsmBridgeError);
    pyo3::create_exception!(_internal, SsmCancelledError, AwsSsmBridgeError);
}
use exceptions::*;

/// Initialize the Python module
#[pymodule]
fn _internal(m: &Bound<'_, PyModule>) -> PyResult<()> {
    // Register classes
    m.add_class::<PySessionManager>()?;
    m.add_class::<PySession>()?;
    m.add_class::<PySessionConfig>()?;
    m.add_class::<PySessionType>()?;
    m.add_class::<PyOutputStream>()?;

    // Register interactive shell classes
    interactive::register(m)?;

    // Register utility functions
    m.add_function(wrap_pyfunction!(configure_logging, m)?)?;

    // Register exception hierarchy — must come after classes so Python
    // callers can import `from aws_ssm_bridge import AwsSsmBridgeError`.
    m.add("AwsSsmBridgeError", m.py().get_type::<AwsSsmBridgeError>())?;
    m.add("SsmSessionError", m.py().get_type::<SsmSessionError>())?;
    m.add("SsmProtocolError", m.py().get_type::<SsmProtocolError>())?;
    m.add("SsmTransportError", m.py().get_type::<SsmTransportError>())?;
    m.add("SsmAwsSdkError", m.py().get_type::<SsmAwsSdkError>())?;
    m.add("SsmTimeoutError", m.py().get_type::<SsmTimeoutError>())?;
    m.add("SsmCancelledError", m.py().get_type::<SsmCancelledError>())?;

    // Add module version
    m.add("__version__", env!("CARGO_PKG_VERSION"))?;

    Ok(())
}

/// Configure logging verbosity.
///
/// Call this before creating a SessionManager to control log output.
/// Levels: "error", "warn", "info", "debug", "trace"
///
/// Alternatively, set the RUST_LOG environment variable (e.g. RUST_LOG=debug).
/// Note: logging can only be configured once per process; subsequent calls are
/// silently ignored (the global tracing subscriber is set on the first call).
#[pyfunction]
#[pyo3(signature = (level="warn"))]
fn configure_logging(level: &str) -> PyResult<()> {
    let directive: tracing::level_filters::LevelFilter = match level.to_lowercase().as_str() {
        "error" => tracing::Level::ERROR.into(),
        "warn" => tracing::Level::WARN.into(),
        "info" => tracing::Level::INFO.into(),
        "debug" => tracing::Level::DEBUG.into(),
        "trace" => tracing::Level::TRACE.into(),
        "off" => tracing::level_filters::LevelFilter::OFF,
        _ => {
            return Err(PyValueError::new_err(format!(
                "Invalid log level '{}'. Use: off, error, warn, info, debug, trace",
                level
            )));
        }
    };

    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::from_default_env().add_directive(directive.into()),
        )
        .try_init();

    Ok(())
}

/// Convert a Rust [`crate::Error`] to the most specific Python exception subclass.
///
/// Python callers can catch the base `AwsSsmBridgeError` for all library errors,
/// or a specific subclass (e.g. `SsmTransportError`) for fine-grained handling.
pub(crate) fn to_py_err(err: crate::Error) -> PyErr {
    use crate::Error;

    match err {
        Error::Config(msg) => PyValueError::new_err(msg),
        Error::InvalidState(msg) => PyRuntimeError::new_err(msg),
        Error::Session(e) => SsmSessionError::new_err(e.to_string()),
        Error::Protocol(e) => SsmProtocolError::new_err(e.to_string()),
        Error::Transport(e) => SsmTransportError::new_err(e.to_string()),
        Error::AwsSdk { message, .. } => SsmAwsSdkError::new_err(message),
        Error::Timeout => SsmTimeoutError::new_err("Operation timed out"),
        Error::Cancelled => SsmCancelledError::new_err("Operation was cancelled"),
        other => AwsSsmBridgeError::new_err(other.to_string()),
    }
}