Pipes

A pipe runs between extraction and the handler: the transport pulls a value out of the request (a path segment, a query param, a JSON body, a GraphQL argument, a WS payload, a queued job), hands it to the pipe, and the pipe either returns the transformed value or rejects with a PipeError the transport renders as its native error.

The trait lives in nest-rs-pipes, transport-agnostic, and binds per argument on every transport. Two forms, one surface:

• HTTP wraps an extractor: nest_rs_http::Piped<P, E> / Valid<E> run the pipe over what E extracts.
• GraphQL, WebSockets, and queues deserialize a single typed value, so there is no extractor to wrap: nest_rs_pipes::Piped<P, T> / Valid<T> expose the wire type T and hand the handler the transformed value. The transport macro (#[resolver], #[messages], #[processor]) does the stripping and the call.

pub trait Pipe {
    type In;
    type Out;
    fn transform(input: Self::In) -> Result<Self::Out, PipeError>;
}

Pipes are stateless

A pipe is a zero-sized marker named at a call site — never instantiated, never injected. transform is an associated function; there’s no &self and no DI container in scope. That’s deliberate: a pipe is a pure transform, the same input always produces the same output, and the type name is enough to pick which one runs.

use nest_rs_pipes::{Pipe, PipeError};

pub struct Trim;

impl Pipe for Trim {
    type In = String;
    type Out = String;
    fn transform(input: String) -> Result<String, PipeError> {
        Ok(input.trim().to_string())
    }
}

If you need DI-injected logic at the request boundary, that’s a service or an interceptor — not a pipe.

The built-in set

nest-rs-pipes ships the common cases. Reach for them before writing your own:

Pipe	In → Out	Use it for
Parse<T>	String → T (any FromStr)	Generic conversion: ParseInt, ParseFloat, ParseBool
ParseUuid	String → Uuid	Any-version UUID
ParseUuidV4 / ParseUuidV7 / …	String → Uuid	Version-pinned UUID
ParseArray<P>	String → Vec<P::Out>	Comma-separated list, each item piped through P
Trim, Lowercase, Uppercase	String → String	Edge normalization
ValidationPipe<T>	T → T (where T: Validate)	Run validator attribute rules

A PipeError rejection surfaces as the transport’s native error — the handler body never runs:

Transport	A rejection becomes
HTTP	RFC 9457 problem+json 400 ({ "type": …, "title": "Bad Request", "status": 400, "detail": "must be a UUID v7" })
GraphQL	An error in the response’s errors array
WebSockets	An error frame on the socket
Queue	A job error — the queue’s retry/failure policy applies

ValidationPipe<T> attaches the structured field-level errors as an errors extension member.

Validation at the HTTP edge

Valid<E> is the ergonomic form of Piped<ValidationPipe<T>, E>: extract E, then validate.

use nest_rs_http::Valid;
use poem::web::Json;
use serde::Deserialize;
use validator::Validate;

#[derive(Deserialize, Validate)]
pub struct CreateUser {
    #[validate(email)]
    email: String,
    #[validate(length(min = 8))]
    password: String,
}

#[post("/")]
async fn create(&self, Valid(Json(input)): Valid<Json<CreateUser>>)
    -> Result<Json<User>>
{
    Ok(Json(self.svc.create(input).await?))
}

A malformed JSON body is 400 before the pipe runs. A well-formed body that fails validation is an RFC 9457 problem+json 400 with:

{
  "type": "https://www.rfc-editor.org/rfc/rfc9110#status.400",
  "title": "Bad Request",
  "status": 400,
  "detail": "validation failed",
  "errors": {
    "email": [{ "code": "email", "message": null, "params": { /* ... */ } }],
    "password": [{ "code": "length", "params": { "min": 8 } }]
  }
}

No manual checks in the handler.

Custom transforms with Piped<P, E>

When the input needs a transform other than Validate, name the pipe at the call site:

use nest_rs_http::Piped;
use nest_rs_pipes::ParseUuidV7;
use poem::web::Path;

#[get("/:id")]
async fn get(&self, Piped(id): Piped<ParseUuidV7, Path<String>>) -> Json<User> {
    Json(self.svc.find(id).await?)
}

Path<String> extracts the raw segment; ParseUuidV7::transform validates it as a v7 UUID. A non-UUID string is 400 with "must be a UUID v7" — not a stringly-typed handler doing its own parse::<Uuid>().

The same pipe beyond HTTP

On GraphQL, WebSockets, and queues the argument is the wire value, so the binding is nest_rs_pipes::Piped<P, T> / Valid<T> — T is what appears on the wire (the SDL argument type, the WS payload, the job payload); the handler body sees P::Out through into_inner() / Deref, exactly as on HTTP.

// GraphQL — the SDL argument is `String`; the body sees it trimmed.
#[query]
#[public]
async fn trimmed(&self, raw: Piped<Trim, String>) -> async_graphql::Result<String> {
    Ok(raw.into_inner())
}

// GraphQL — validates the input object; the SDL argument is `NameInput`.
#[query]
#[public]
async fn named(&self, input: Valid<NameInput>) -> async_graphql::Result<String> {
    Ok(input.into_inner().name)
}

// WebSockets — the message payload is a `String`, piped before the handler.
#[subscribe_message("trim")]
async fn trim_handler(&self, name: Piped<Trim, String>) -> String {
    name.into_inner()
}

// Queue — the job payload is a `String`, piped after deserialization.
#[process(queue = "signups", concurrency = 1, retries = 0)]
async fn handle(&self, name: Piped<Trim, String>) -> anyhow::Result<()> {
    /* ... */
}

Pipes vs Bind

A pipe converts a primitive input — an ID string into a Uuid, a JSON blob into a validated DTO. A Bind<S, A> extractor goes further: it parses the id, loads the row through a service, and authorizes it — returning 404 if the row doesn’t exist within the caller’s scope, 403 if denied.

// Pipe — pure conversion, no DB
#[get("/_validate/:id")]
async fn validate(&self, Piped(id): Piped<ParseUuidV7, Path<String>>) -> String {
    format!("{id} is valid")
}

// Bind — convert + load + authorize
#[get("/:id")]
async fn get(&self, user: Bind<UsersService, Read>) -> Json<User> {
    Json(User::from(&*user))
}

Bind depends on the data layer (its GraphQL analog is bind — see By-id binding). Pipe is transport-agnostic — the same ParseUuidV7 binds identically on HTTP, GraphQL, WebSockets, and queues.

Writing a pipe

Three rules:

1. Stateless. transform is an associated function; no &self, no container.
2. One file per pipe under crates/nest-rs-pipes/src/pipes/ (one role, one file). Reusable pipes belong to the framework crate — never to an app.
3. Use the error envelope. Return PipeError::new(msg) for a simple message, PipeError::with_details(msg, details) to carry structured field-level errors.

use nest_rs_pipes::{Pipe, PipeError};

pub struct NonEmpty;

impl Pipe for NonEmpty {
    type In = String;
    type Out = String;
    fn transform(input: String) -> Result<String, PipeError> {
        if input.trim().is_empty() {
            Err(PipeError::new("must not be empty"))
        } else {
            Ok(input)
        }
    }
}

Then bind it at the call site:

#[post("/")]
async fn create(&self, Piped(name): Piped<NonEmpty, Json<String>>) -> &'static str {
    "ok"
}

Note

Reusable pipes live in the framework, never in an app. A single-feature transform is a service. A pattern that recurs in two apps belongs in crates/nest-rs-pipes/.

Going further

• HTTP / controllers — every extractor that composes with a pipe.
• Guards — gate access; pipes don’t.
• Database / Bind — pipes are for primitives, Bind loads and authorizes a row.
• oauth/http/controller.rs in the demo — binds Piped<Lowercase, Path<String>> so the OAuth provider segment is case-insensitive.