common: implement remaining builtin text codecs (#8687)

* common: implement remaining builtin text codecs Move utf-16/32/7, unicode-escape, raw-unicode-escape and the bytes escape transform into rustpython-common and call them from _codecs instead of _pycodecs. Assisted-by: Claude:grok-4.6 * host_env: share locale grouping, uuid and pipe state Add localeconv_numeric so grouping keeps a CHAR_MAX terminator, UuidCreateSequential for _uuid, and GetNamedPipeHandleStateW. Assisted-by: Claude:grok-4.6 * Fix utf-7 clippy and the utf-16-be snippet length Assisted-by: Claude:grok-4.6 * codecs: treat escape decode as final and accept str unicode_escape_decode and raw_unicode_escape_decode default final to true and accept a str as UTF-8. Fix clippy on wide_order, escape encode, and encode closures. Remove expectedFailure markers that now pass. Assisted-by: Grok:grok-4.6
2026-10-28 18:59:04 +08:00 · 2026-09-12 15:51:28 +09:00
1 parent 43ed0a80b6
commit 66a3d9daa7
@@ -828,8 +828,6 @@ class CodecCallbackTest(unittest.TestCase):
                self.assertEqual(exc.end, 2)
                self.assertEqual(exc.object, input)

-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
    def test_encode_unencodable_replacement(self):
        def unencrepl(exc):
            if isinstance(exc, UnicodeEncodeError):
@@ -1063,7 +1061,6 @@ class CodecCallbackTest(unittest.TestCase):
                decoded = input.decode(enc, "test.bug36819")
                self.assertEqual(decoded, 'abcdx' * 51)

-    @unittest.expectedFailureIf(sys.platform != "win32", "TODO: RUSTPYTHON")
    def test_encodehelper_bug36819(self):
        handler = RepeatedPosReturn()
        codecs.register_error("test.bug36819", handler.handle)
@@ -1133,8 +1130,6 @@ class CodecCallbackTest(unittest.TestCase):
            text = 'abc<def>ghi'*n
            text.translate(charmap)

-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
    def test_mutatingdecodehandler(self):
        baddata = [
            ("ascii", b"\xff"),
@@ -1171,8 +1166,6 @@ class CodecCallbackTest(unittest.TestCase):
            self.assertEqual(data.decode(encoding, "test.mutating"), "\u4242")

    # issue32583
-    # TODO: RUSTPYTHON
-    @unittest.expectedFailure
    def test_crashing_decode_handler(self):
        # better generating one more character to fill the extra space slot
        # so in debug build it can steadily fail
@@ -1199,7 +1199,6 @@ class EscapeDecodeTest(unittest.TestCase):
        check(br"[\x41]", b"[A]")
        check(br"[\x410]", b"[A0]")

-    @unittest.expectedFailure  # TODO: RUSTPYTHON; DeprecationWarning not triggered
    def test_warnings(self):
        decode = codecs.escape_decode
        check = coding_checker(self, decode)
@@ -2582,7 +2581,6 @@ class WithStmtTest(unittest.TestCase):


 class TypesTest(unittest.TestCase):
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; AttributeError: module 'codecs' has no attribute 'utf_32_ex_decode'. Did you mean: 'utf_16_ex_decode'?
    def test_decode_unicode(self):
        # Most decoders don't accept unicode input
        decoders = [
@@ -2684,7 +2682,6 @@ class UnicodeEscapeTest(ReadTest, unittest.TestCase):
        check(br"\u20ac", "\u20ac")
        check(br"\U0001d120", "\U0001d120")

-    @unittest.expectedFailure  # TODO: RUSTPYTHON; DeprecationWarning not triggered
    def test_decode_warnings(self):
        decode = codecs.unicode_escape_decode
        check = coding_checker(self, decode)
@@ -7,6 +7,14 @@ use crate::wtf8::{CodePoint, Wtf8, Wtf8Buf};

 #[cfg(feature = "cjk-codecs")]
 pub mod cjk;
+mod wide;
+pub use wide::ByteOrder;
+pub mod escape;
+pub mod raw_unicode_escape;
+pub mod unicode_escape;
+pub mod utf16;
+pub mod utf32;
+pub mod utf7;

 pub trait StrBuffer: AsRef<Wtf8> {
    fn is_compatible_with(&self, kind: StrKind) -> bool {
@@ -0,0 +1,153 @@
+//! Bytes-to-bytes Python string-literal escape transform (`escape_encode` /
+//! `escape_decode`).
+
+use alloc::string::String;
+use alloc::vec::Vec;
+
+/// Encode `data` the way `escape_encode` / `string_escape_encode` does.
+#[must_use]
+pub fn encode(data: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(data.len());
+    for &byte in data {
+        match byte {
+            b'\t' => out.extend_from_slice(b"\\t"),
+            b'\n' => out.extend_from_slice(b"\\n"),
+            b'\r' => out.extend_from_slice(b"\\r"),
+            b'\\' => out.extend_from_slice(b"\\\\"),
+            b'\'' => out.extend_from_slice(b"\\'"),
+            0x20..=0x7e => out.push(byte),
+            value => out.extend_from_slice(format!("\\x{value:02x}").as_bytes()),
+        }
+    }
+    out
+}
+
+/// How a malformed `\x` escape is reported.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum EscapeErrorMode {
+    Strict,
+    Replace,
+    Ignore,
+}
+
+impl EscapeErrorMode {
+    #[must_use]
+    pub fn from_name(name: &str) -> Option<Self> {
+        match name {
+            "strict" => Some(Self::Strict),
+            "replace" => Some(Self::Replace),
+            "ignore" => Some(Self::Ignore),
+            _ => None,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum EscapeDecodeError {
+    TrailingBackslash,
+    InvalidHex { position: usize },
+    UnknownHandler { name: String },
+}
+
+/// Decode a Python bytes-literal escape sequence.
+///
+/// `warning` is the first invalid-escape deprecation text, if any.
+pub fn decode(
+    data: &[u8],
+    errors: EscapeErrorMode,
+) -> Result<(Vec<u8>, Option<String>), EscapeDecodeError> {
+    let mut out = Vec::with_capacity(data.len());
+    let mut pos = 0usize;
+    let mut warning = None;
+    while pos < data.len() {
+        if data[pos] != b'\\' {
+            out.push(data[pos]);
+            pos += 1;
+            continue;
+        }
+        let escape_start = pos;
+        pos += 1;
+        if pos == data.len() {
+            return Err(EscapeDecodeError::TrailingBackslash);
+        }
+        let ch = data[pos];
+        pos += 1;
+        match ch {
+            b'\n' => {}
+            b'\\' => out.push(b'\\'),
+            b'\'' => out.push(b'\''),
+            b'"' => out.push(b'"'),
+            b'b' => out.push(0x08),
+            b'f' => out.push(0x0c),
+            b't' => out.push(b'\t'),
+            b'n' => out.push(b'\n'),
+            b'r' => out.push(b'\r'),
+            b'v' => out.push(0x0b),
+            b'a' => out.push(0x07),
+            b'0'..=b'7' => {
+                let octal_start = pos - 1;
+                while pos < data.len()
+                    && pos < octal_start + 3
+                    && (b'0'..=b'7').contains(&data[pos])
+                {
+                    pos += 1;
+                }
+                let raw = data[octal_start..pos]
+                    .iter()
+                    .fold(0u16, |value, digit| value * 8 + u16::from(digit - b'0'));
+                if raw >= 256 && warning.is_none() {
+                    warning = Some(invalid_escape_warning(
+                        "b",
+                        &String::from_utf8_lossy(&data[octal_start..pos]),
+                        true,
+                    ));
+                }
+                out.push(raw as u8);
+            }
+            b'x' => {
+                let hi = data.get(pos).and_then(|byte| (*byte as char).to_digit(16));
+                let lo = data
+                    .get(pos + 1)
+                    .and_then(|byte| (*byte as char).to_digit(16));
+                if let (Some(hi), Some(lo)) = (hi, lo) {
+                    out.push((hi * 16 + lo) as u8);
+                    pos += 2;
+                } else {
+                    match errors {
+                        EscapeErrorMode::Strict => {
+                            return Err(EscapeDecodeError::InvalidHex {
+                                position: escape_start,
+                            });
+                        }
+                        EscapeErrorMode::Replace => out.push(b'?'),
+                        EscapeErrorMode::Ignore => {}
+                    }
+                    if data.get(pos).is_some_and(u8::is_ascii_hexdigit) {
+                        pos += 1;
+                    }
+                }
+            }
+            other => {
+                out.push(b'\\');
+                pos -= 1;
+                if warning.is_none() {
+                    warning = Some(invalid_escape_warning(
+                        "b",
+                        &char::from(other).to_string(),
+                        false,
+                    ));
+                }
+            }
+        }
+    }
+    Ok((out, warning))
+}
+
+fn invalid_escape_warning(prefix: &str, sequence: &str, octal: bool) -> String {
+    let kind = if octal {
+        "an invalid octal escape sequence"
+    } else {
+        "an invalid escape sequence"
+    };
+    format!("{prefix}\"\\{sequence}\" is {kind}. Such sequences will not work in the future. ")
+}
@@ -0,0 +1,112 @@
+//! `raw-unicode-escape` encode / incremental decode.
+
+use super::*;
+use crate::wtf8::{CodePoint, Wtf8, Wtf8Buf};
+
+pub const ENCODING_NAME: &str = "rawunicodeescape";
+
+/// Encode `s`. Code points below 0x100 stay as a Latin-1 byte; the rest
+/// become `\uXXXX` / `\UXXXXXXXX`.
+#[must_use]
+pub fn encode_bytes(s: &Wtf8) -> Vec<u8> {
+    let mut out = Vec::new();
+    for cp in s.code_points() {
+        let v = cp.to_u32();
+        if v < 0x100 {
+            out.push(v as u8);
+        } else if v < 0x10000 {
+            out.extend_from_slice(format!("\\u{v:04x}").as_bytes());
+        } else {
+            out.extend_from_slice(format!("\\U{v:08x}").as_bytes());
+        }
+    }
+    out
+}
+
+pub fn encode<Ctx, E>(ctx: Ctx, _errors: &E) -> Result<Vec<u8>, Ctx::Error>
+where
+    Ctx: EncodeContext,
+    E: EncodeErrorHandler<Ctx>,
+{
+    Ok(encode_bytes(ctx.remaining_data()))
+}
+
+pub fn decode<Ctx, E>(
+    mut ctx: Ctx,
+    errors: &E,
+    final_decode: bool,
+) -> Result<(Wtf8Buf, usize), Ctx::Error>
+where
+    Ctx: DecodeContext,
+    E: DecodeErrorHandler<Ctx>,
+{
+    let mut out = Wtf8Buf::new();
+    loop {
+        let rest = ctx.remaining_data();
+        if rest.is_empty() {
+            break;
+        }
+        let b = rest[0];
+        if b != b'\\' {
+            out.push_char(b as char);
+            ctx.advance(1);
+            continue;
+        }
+        let kind = rest.get(1).copied();
+        let want = match kind {
+            Some(b'u') => 4usize,
+            Some(b'U') => 8usize,
+            _ => 0,
+        };
+        if want != 0 {
+            let escape_start = ctx.position();
+            let digits_start = 2;
+            if !final_decode && rest.len() < digits_start + want {
+                break;
+            }
+            let available = (digits_start + want).min(rest.len());
+            let mut hex_end = digits_start;
+            while hex_end < available && rest[hex_end].is_ascii_hexdigit() {
+                hex_end += 1;
+            }
+            let numeric = if available == digits_start + want && hex_end == available {
+                core::str::from_utf8(&rest[digits_start..available])
+                    .ok()
+                    .and_then(|s| u32::from_str_radix(s, 16).ok())
+            } else {
+                None
+            };
+            if let Some(c) = numeric.and_then(CodePoint::from_u32) {
+                out.push(c);
+                ctx.advance(available);
+                continue;
+            }
+            let error_end = ctx.position()
+                + if numeric.is_some() {
+                    available
+                } else {
+                    hex_end
+                };
+            let reason = if numeric.is_some() {
+                "illegal Unicode character"
+            } else if want == 4 {
+                "truncated \\uXXXX escape"
+            } else {
+                "truncated \\UXXXXXXXX escape"
+            };
+            let replace = ctx.handle_error(errors, escape_start..error_end, Some(reason))?;
+            out.push_wtf8(replace.as_ref());
+            continue;
+        }
+        if kind.is_none() && !final_decode {
+            break;
+        }
+        out.push_char(b as char);
+        ctx.advance(1);
+        if let Some(next) = kind {
+            out.push_char(next as char);
+            ctx.advance(1);
+        }
+    }
+    Ok((out, ctx.position()))
+}
@@ -0,0 +1,272 @@
+//! `unicode-escape` encode / incremental decode.
+
+use super::*;
+use crate::wtf8::{CodePoint, Wtf8, Wtf8Buf};
+
+pub const ENCODING_NAME: &str = "unicodeescape";
+
+fn push_hex(out: &mut Vec<u8>, prefix: u8, cp: u32, digits: usize) {
+    out.push(b'\\');
+    out.push(prefix);
+    for shift in (0..digits).rev() {
+        out.push(b"0123456789abcdef"[((cp >> (shift * 4)) & 0xf) as usize]);
+    }
+}
+
+/// Encode `s` as a Python unicode-escape byte string. Every code point is
+/// representable, so this cannot fail.
+#[must_use]
+pub fn encode_bytes(s: &Wtf8) -> Vec<u8> {
+    let mut out = Vec::new();
+    for cp in s.code_points() {
+        match cp.to_u32() {
+            0x5c => out.extend_from_slice(br"\\"),
+            0x09 => out.extend_from_slice(br"\t"),
+            0x0a => out.extend_from_slice(br"\n"),
+            0x0d => out.extend_from_slice(br"\r"),
+            0x20..=0x7e => out.push(cp.to_u32() as u8),
+            c @ 0x00..=0xff => push_hex(&mut out, b'x', c, 2),
+            c @ 0x100..=0xffff => push_hex(&mut out, b'u', c, 4),
+            c => push_hex(&mut out, b'U', c, 8),
+        }
+    }
+    out
+}
+
+pub fn encode<Ctx, E>(ctx: Ctx, _errors: &E) -> Result<Vec<u8>, Ctx::Error>
+where
+    Ctx: EncodeContext,
+    E: EncodeErrorHandler<Ctx>,
+{
+    Ok(encode_bytes(ctx.remaining_data()))
+}
+
+/// Optional first invalid-escape deprecation text. The caller issues it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct EscapeNote {
+    pub message: String,
+}
+
+fn invalid_escape_warning(prefix: &str, sequence: &str, octal: bool) -> String {
+    let kind = if octal {
+        "an invalid octal escape sequence"
+    } else {
+        "an invalid escape sequence"
+    };
+    format!("{prefix}\"\\{sequence}\" is {kind}. Such sequences will not work in the future. ")
+}
+
+/// Incremental unicode-escape decode.
+pub fn decode<Ctx, E>(
+    mut ctx: Ctx,
+    errors: &E,
+    final_decode: bool,
+) -> Result<(Wtf8Buf, usize, Option<EscapeNote>), Ctx::Error>
+where
+    Ctx: DecodeContext,
+    E: DecodeErrorHandler<Ctx>,
+{
+    let mut out = Wtf8Buf::new();
+    let mut warning = None;
+    loop {
+        let rest = ctx.remaining_data();
+        if rest.is_empty() {
+            break;
+        }
+        let ch = rest[0];
+        if ch != b'\\' {
+            if let Some(cp) = CodePoint::from_u32(ch as u32) {
+                out.push(cp);
+            }
+            ctx.advance(1);
+            continue;
+        }
+        let escape_start = ctx.position();
+        if rest.len() == 1 {
+            if !final_decode {
+                break;
+            }
+            let end = ctx.full_data().len();
+            let replace =
+                ctx.handle_error(errors, escape_start..end, Some("\\ at end of string"))?;
+            out.push_wtf8(replace.as_ref());
+            continue;
+        }
+        let intro = rest[1];
+        match intro {
+            b'\n' => ctx.advance(2),
+            b'\\' => {
+                out.push_char('\\');
+                ctx.advance(2);
+            }
+            b'\'' => {
+                out.push_char('\'');
+                ctx.advance(2);
+            }
+            b'"' => {
+                out.push_char('"');
+                ctx.advance(2);
+            }
+            b'b' => {
+                out.push_char('\x08');
+                ctx.advance(2);
+            }
+            b'f' => {
+                out.push_char('\x0c');
+                ctx.advance(2);
+            }
+            b't' => {
+                out.push_char('\t');
+                ctx.advance(2);
+            }
+            b'n' => {
+                out.push_char('\n');
+                ctx.advance(2);
+            }
+            b'r' => {
+                out.push_char('\r');
+                ctx.advance(2);
+            }
+            b'v' => {
+                out.push_char('\x0b');
+                ctx.advance(2);
+            }
+            b'a' => {
+                out.push_char('\x07');
+                ctx.advance(2);
+            }
+            b'0'..=b'7' => {
+                ctx.advance(2);
+                let mut value = (intro - b'0') as u32;
+                let octal_start = escape_start + 1;
+                for _ in 0..2 {
+                    let rest = ctx.remaining_data();
+                    if rest.first().is_some_and(|b| matches!(b, b'0'..=b'7')) {
+                        value = (value << 3) + (rest[0] - b'0') as u32;
+                        ctx.advance(1);
+                    }
+                }
+                if value > 0o377 && warning.is_none() {
+                    let seq = &ctx.full_data()[octal_start..ctx.position()];
+                    warning = Some(EscapeNote {
+                        message: invalid_escape_warning("", &String::from_utf8_lossy(seq), true),
+                    });
+                }
+                if let Some(cp) = CodePoint::from_u32(value) {
+                    out.push(cp);
+                }
+            }
+            b'x' | b'u' | b'U' => {
+                let (digits, message) = match intro {
+                    b'x' => (2usize, "truncated \\xXX escape"),
+                    b'u' => (4usize, "truncated \\uXXXX escape"),
+                    _ => (8usize, "truncated \\UXXXXXXXX escape"),
+                };
+                if !final_decode && rest.len() < 2 + digits {
+                    break;
+                }
+                ctx.advance(2);
+                let rest = ctx.remaining_data();
+                if rest.len() >= digits && rest[..digits].iter().all(u8::is_ascii_hexdigit) {
+                    let value =
+                        u32::from_str_radix(core::str::from_utf8(&rest[..digits]).unwrap(), 16)
+                            .unwrap();
+                    if let Some(cp) = CodePoint::from_u32(value) {
+                        out.push(cp);
+                        ctx.advance(digits);
+                        continue;
+                    }
+                    let start = escape_start;
+                    let end = ctx.position() + digits;
+                    let replace =
+                        ctx.handle_error(errors, start..end, Some("illegal Unicode character"))?;
+                    out.push_wtf8(replace.as_ref());
+                    continue;
+                }
+                let mut hex_end = 0;
+                while hex_end < rest.len() && rest[hex_end].is_ascii_hexdigit() {
+                    hex_end += 1;
+                }
+                let start = escape_start;
+                let end = ctx.position() + hex_end;
+                let replace = ctx.handle_error(errors, start..end, Some(message))?;
+                out.push_wtf8(replace.as_ref());
+            }
+            b'N' => {
+                if rest.len() == 2 && !final_decode {
+                    break;
+                }
+                if rest.len() >= 3 && rest[2] == b'{' {
+                    let name_start = 3;
+                    let mut look = name_start;
+                    while look < rest.len() && rest[look] != b'}' {
+                        look += 1;
+                    }
+                    if look >= rest.len() {
+                        if !final_decode {
+                            break;
+                        }
+                        let start = escape_start;
+                        let end = ctx.full_data().len();
+                        let replace = ctx.handle_error(
+                            errors,
+                            start..end,
+                            Some("malformed \\N character escape"),
+                        )?;
+                        out.push_wtf8(replace.as_ref());
+                        continue;
+                    }
+                    if look == name_start {
+                        let start = escape_start;
+                        let end = ctx.position() + name_start;
+                        let replace = ctx.handle_error(
+                            errors,
+                            start..end,
+                            Some("malformed \\N character escape"),
+                        )?;
+                        out.push_wtf8(replace.as_ref());
+                        continue;
+                    }
+                    let name = core::str::from_utf8(&rest[name_start..look]).ok();
+                    match name.and_then(rustpython_unicode::lookup_character) {
+                        Some(ch) => {
+                            out.push_char(ch);
+                            ctx.advance(look + 1);
+                            continue;
+                        }
+                        None => {
+                            let start = escape_start;
+                            let end = ctx.position() + look + 1;
+                            let replace = ctx.handle_error(
+                                errors,
+                                start..end,
+                                Some("unknown Unicode character name"),
+                            )?;
+                            out.push_wtf8(replace.as_ref());
+                            continue;
+                        }
+                    }
+                }
+                ctx.advance(2);
+                let start = escape_start;
+                let end = ctx.position();
+                let replace =
+                    ctx.handle_error(errors, start..end, Some("malformed \\N character escape"))?;
+                out.push_wtf8(replace.as_ref());
+            }
+            other => {
+                if warning.is_none() {
+                    warning = Some(EscapeNote {
+                        message: invalid_escape_warning("", &char::from(other).to_string(), false),
+                    });
+                }
+                out.push_char('\\');
+                if let Some(cp) = CodePoint::from_u32(other as u32) {
+                    out.push(cp);
+                }
+                ctx.advance(2);
+            }
+        }
+    }
+    Ok((out, ctx.position(), warning))
+}
@@ -0,0 +1,97 @@
+//! UTF-16 encode / incremental decode.
+
+use super::wide::{self, emit_utf16, is_surrogate, push_codepoint, read_u16, resolve_bom};
+use super::*;
+use crate::wtf8::Wtf8Buf;
+
+pub use super::wide::ByteOrder;
+
+pub const ENCODING_NAME: &str = "utf-16";
+pub const ENCODING_NAME_LE: &str = "utf-16-le";
+pub const ENCODING_NAME_BE: &str = "utf-16-be";
+
+const ERR_REASON: &str = "surrogates not allowed";
+
+pub fn encode<Ctx, E>(
+    ctx: Ctx,
+    errors: &E,
+    order: ByteOrder,
+    bom: bool,
+) -> Result<Vec<u8>, Ctx::Error>
+where
+    Ctx: EncodeContext,
+    E: EncodeErrorHandler<Ctx>,
+{
+    wide::encode_wide(ctx, errors, order.is_big_endian(), bom, 2, ERR_REASON)
+}
+
+/// Decode one incremental chunk.
+///
+/// Returns `(text, consumed, byteorder)` where `byteorder` is CPython's
+/// `-1` / `0` / `1`.
+pub fn decode<Ctx, E>(
+    mut ctx: Ctx,
+    errors: &E,
+    order: ByteOrder,
+    final_decode: bool,
+) -> Result<(Wtf8Buf, usize, i32), Ctx::Error>
+where
+    Ctx: DecodeContext,
+    E: DecodeErrorHandler<Ctx>,
+{
+    let (big_endian, skip, byteorder) = resolve_bom(ctx.remaining_data(), order, 2);
+    ctx.advance(skip);
+    let mut out = Wtf8Buf::new();
+    loop {
+        let rest = ctx.remaining_data();
+        if rest.len() < 2 {
+            if rest.is_empty() || !final_decode {
+                break;
+            }
+            let start = ctx.position();
+            let end = ctx.full_data().len();
+            let replace = ctx.handle_error(errors, start..end, Some("truncated data"))?;
+            out.push_wtf8(replace.as_ref());
+            continue;
+        }
+        let ch = read_u16(rest, big_endian);
+        if !is_surrogate(ch as u32) {
+            push_codepoint(&mut out, ch as u32);
+            ctx.advance(2);
+            continue;
+        }
+        if ch >= 0xdc00 {
+            let start = ctx.position();
+            let replace = ctx.handle_error(errors, start..start + 2, Some("illegal encoding"))?;
+            out.push_wtf8(replace.as_ref());
+            continue;
+        }
+        if rest.len() < 4 {
+            if !final_decode {
+                break;
+            }
+            let start = ctx.position();
+            let end = ctx.full_data().len();
+            let replace = ctx.handle_error(errors, start..end, Some("unexpected end of data"))?;
+            out.push_wtf8(replace.as_ref());
+            continue;
+        }
+        let ch2 = read_u16(&rest[2..], big_endian);
+        if (0xdc00..=0xdfff).contains(&ch2) {
+            let c = (((ch as u32 & 0x3ff) << 10) | (ch2 as u32 & 0x3ff)) + 0x10000;
+            push_codepoint(&mut out, c);
+            ctx.advance(4);
+        } else {
+            let start = ctx.position();
+            let replace =
+                ctx.handle_error(errors, start..start + 2, Some("illegal UTF-16 surrogate"))?;
+            out.push_wtf8(replace.as_ref());
+        }
+    }
+    Ok((out, ctx.position(), byteorder))
+}
+
+/// Encode a scalar (including a surrogate) as one or two UTF-16 units.
+pub fn encode_codepoint(out: &mut Vec<u8>, cp: u32, order: ByteOrder) {
+    emit_utf16(out, cp, order.is_big_endian());
+}
@@ -0,0 +1,86 @@
+//! UTF-32 encode / incremental decode.
+
+use super::wide::{self, emit_utf32, is_surrogate, push_codepoint, read_u32, resolve_bom};
+use super::*;
+use crate::wtf8::Wtf8Buf;
+
+pub use super::wide::ByteOrder;
+
+pub const ENCODING_NAME: &str = "utf-32";
+pub const ENCODING_NAME_LE: &str = "utf-32-le";
+pub const ENCODING_NAME_BE: &str = "utf-32-be";
+
+const ERR_REASON: &str = "surrogates not allowed";
+
+pub fn encode<Ctx, E>(
+    ctx: Ctx,
+    errors: &E,
+    order: ByteOrder,
+    bom: bool,
+) -> Result<Vec<u8>, Ctx::Error>
+where
+    Ctx: EncodeContext,
+    E: EncodeErrorHandler<Ctx>,
+{
+    wide::encode_wide(ctx, errors, order.is_big_endian(), bom, 4, ERR_REASON)
+}
+
+/// Decode one incremental chunk.
+///
+/// Returns `(text, consumed, byteorder)` where `byteorder` is CPython's
+/// `-1` / `0` / `1`.
+pub fn decode<Ctx, E>(
+    mut ctx: Ctx,
+    errors: &E,
+    order: ByteOrder,
+    final_decode: bool,
+) -> Result<(Wtf8Buf, usize, i32), Ctx::Error>
+where
+    Ctx: DecodeContext,
+    E: DecodeErrorHandler<Ctx>,
+{
+    let (big_endian, skip, byteorder) = resolve_bom(ctx.remaining_data(), order, 4);
+    ctx.advance(skip);
+    let mut out = Wtf8Buf::new();
+    loop {
+        let rest = ctx.remaining_data();
+        if rest.len() < 4 {
+            if rest.is_empty() || !final_decode {
+                break;
+            }
+            let start = ctx.position();
+            let end = ctx.full_data().len();
+            let replace = ctx.handle_error(errors, start..end, Some("truncated data"))?;
+            out.push_wtf8(replace.as_ref());
+            continue;
+        }
+        let ch = read_u32(rest, big_endian);
+        if is_surrogate(ch) {
+            let start = ctx.position();
+            let replace = ctx.handle_error(
+                errors,
+                start..start + 4,
+                Some("code point in surrogate code point range(0xd800, 0xe000)"),
+            )?;
+            out.push_wtf8(replace.as_ref());
+            continue;
+        }
+        if ch >= 0x110000 {
+            let start = ctx.position();
+            let replace = ctx.handle_error(
+                errors,
+                start..start + 4,
+                Some("code point not in range(0x110000)"),
+            )?;
+            out.push_wtf8(replace.as_ref());
+            continue;
+        }
+        push_codepoint(&mut out, ch);
+        ctx.advance(4);
+    }
+    Ok((out, ctx.position(), byteorder))
+}
+
+pub fn encode_codepoint(out: &mut Vec<u8>, cp: u32, order: ByteOrder) {
+    emit_utf32(out, cp, order.is_big_endian());
+}
@@ -0,0 +1,254 @@
+//! UTF-7 encode / incremental decode.
+
+use super::*;
+use crate::wtf8::{CodePoint, Wtf8, Wtf8Buf};
+
+pub const ENCODING_NAME: &str = "utf-7";
+
+fn is_base64(b: u8) -> bool {
+    b.is_ascii_alphanumeric() || b == b'+' || b == b'/'
+}
+
+fn to_base64(n: u32) -> u8 {
+    b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"[(n & 0x3f) as usize]
+}
+
+fn from_base64(b: u8) -> u32 {
+    match b {
+        b'a'..=b'z' => (b - 71) as u32,
+        b'A'..=b'Z' => (b - 65) as u32,
+        b'0'..=b'9' => (b + 4) as u32,
+        b'+' => 62,
+        _ => 63,
+    }
+}
+
+fn decode_direct(b: u8) -> bool {
+    b <= 127 && b != b'+'
+}
+
+fn category(oc: u32) -> u8 {
+    if oc > 127 {
+        return 3;
+    }
+    let b = oc as u8;
+    if matches!(b, b'\t' | b'\n' | b'\r' | b' ') {
+        2
+    } else if b.is_ascii_alphanumeric() || b"'(),-./:?".contains(&b) {
+        0
+    } else if b"!\"#$%&*;<=>@[]^_`{|}".contains(&b) {
+        1
+    } else {
+        3
+    }
+}
+
+fn encode_direct(oc: u32) -> bool {
+    oc < 128 && oc > 0 && category(oc) != 3
+}
+
+fn encode_unit(out: &mut Vec<u8>, unit: u32, bits: &mut u32, buffer: &mut u32) {
+    *bits += 16;
+    *buffer = (*buffer << 16) | unit;
+    while *bits >= 6 {
+        out.push(to_base64(*buffer >> (*bits - 6)));
+        *bits -= 6;
+    }
+    *buffer &= (1 << *bits) - 1;
+}
+
+/// Encode `s` as UTF-7. Every code point is representable, so this cannot fail.
+#[must_use]
+pub fn encode_bytes(s: &Wtf8) -> Vec<u8> {
+    let mut out = Vec::new();
+    let mut in_shift = false;
+    let mut bits = 0;
+    let mut buffer = 0;
+    for cp in s.code_points() {
+        let oc = cp.to_u32();
+        if !in_shift {
+            if oc == b'+' as u32 {
+                out.extend_from_slice(b"+-");
+            } else if encode_direct(oc) {
+                out.push(oc as u8);
+            } else {
+                out.push(b'+');
+                in_shift = true;
+                emit_code(&mut out, oc, &mut bits, &mut buffer);
+            }
+        } else if encode_direct(oc) {
+            if bits != 0 {
+                out.push(to_base64(buffer << (6 - bits)));
+                buffer = 0;
+                bits = 0;
+            }
+            in_shift = false;
+            if is_base64(oc as u8) || oc == b'-' as u32 {
+                out.push(b'-');
+            }
+            out.push(oc as u8);
+        } else {
+            emit_code(&mut out, oc, &mut bits, &mut buffer);
+        }
+    }
+    if bits != 0 {
+        out.push(to_base64(buffer << (6 - bits)));
+    }
+    if in_shift {
+        out.push(b'-');
+    }
+    out
+}
+
+fn emit_code(out: &mut Vec<u8>, oc: u32, bits: &mut u32, buffer: &mut u32) {
+    if oc >= 0x10000 {
+        encode_unit(out, 0xd800 | ((oc - 0x10000) >> 10), bits, buffer);
+        encode_unit(out, 0xdc00 | ((oc - 0x10000) & 0x3ff), bits, buffer);
+    } else {
+        encode_unit(out, oc, bits, buffer);
+    }
+}
+
+pub fn encode<Ctx, E>(ctx: Ctx, _errors: &E) -> Result<Vec<u8>, Ctx::Error>
+where
+    Ctx: EncodeContext,
+    E: EncodeErrorHandler<Ctx>,
+{
+    Ok(encode_bytes(ctx.remaining_data()))
+}
+
+/// Incremental UTF-7 decode.
+pub fn decode<Ctx, E>(
+    mut ctx: Ctx,
+    errors: &E,
+    final_decode: bool,
+) -> Result<(Wtf8Buf, usize), Ctx::Error>
+where
+    Ctx: DecodeContext,
+    E: DecodeErrorHandler<Ctx>,
+{
+    let mut out = Wtf8Buf::new();
+    let mut in_shift = false;
+    let mut bits = 0u32;
+    let mut buffer = 0u32;
+    let mut surrogate = 0u32;
+    let mut shift_out_start = 0usize;
+    let mut startinpos = 0usize;
+    loop {
+        let rest = ctx.remaining_data();
+        if rest.is_empty() {
+            break;
+        }
+        let ch = rest[0];
+        if in_shift {
+            if is_base64(ch) {
+                buffer = (buffer << 6) | from_base64(ch);
+                bits += 6;
+                ctx.advance(1);
+                if bits >= 16 {
+                    let out_ch = buffer >> (bits - 16);
+                    bits -= 16;
+                    buffer &= (1 << bits) - 1;
+                    if surrogate != 0 {
+                        if (0xdc00..=0xdfff).contains(&out_ch) {
+                            let code = (((surrogate & 0x3ff) << 10) | (out_ch & 0x3ff)) + 0x10000;
+                            if let Some(cp) = CodePoint::from_u32(code) {
+                                out.push(cp);
+                            }
+                            surrogate = 0;
+                            continue;
+                        }
+                        if let Some(cp) = CodePoint::from_u32(surrogate) {
+                            out.push(cp);
+                        }
+                        surrogate = 0;
+                    }
+                    if (0xd800..=0xdbff).contains(&out_ch) {
+                        surrogate = out_ch;
+                    } else if let Some(cp) = CodePoint::from_u32(out_ch) {
+                        out.push(cp);
+                    }
+                }
+            } else {
+                in_shift = false;
+                if bits >= 6 {
+                    ctx.advance(1);
+                    let start = startinpos;
+                    let end = ctx.position();
+                    let replace = ctx.handle_error(
+                        errors,
+                        start..end,
+                        Some("partial character in shift sequence"),
+                    )?;
+                    out.push_wtf8(replace.as_ref());
+                    continue;
+                } else if bits > 0 && buffer != 0 {
+                    ctx.advance(1);
+                    let start = startinpos;
+                    let end = ctx.position();
+                    let replace = ctx.handle_error(
+                        errors,
+                        start..end,
+                        Some("non-zero padding bits in shift sequence"),
+                    )?;
+                    out.push_wtf8(replace.as_ref());
+                    continue;
+                }
+                if surrogate != 0
+                    && decode_direct(ch)
+                    && let Some(cp) = CodePoint::from_u32(surrogate)
+                {
+                    out.push(cp);
+                }
+                surrogate = 0;
+                if ch == b'-' {
+                    ctx.advance(1);
+                }
+            }
+        } else if ch == b'+' {
+            startinpos = ctx.position();
+            if rest.len() >= 2 && rest[1] == b'-' {
+                ctx.advance(2);
+                out.push_char('+');
+            } else if rest.len() >= 2 && !is_base64(rest[1]) {
+                let start = ctx.position();
+                let replace =
+                    ctx.handle_error(errors, start..start + 2, Some("ill-formed sequence"))?;
+                out.push_wtf8(replace.as_ref());
+            } else {
+                ctx.advance(1);
+                in_shift = true;
+                surrogate = 0;
+                shift_out_start = out.len();
+                bits = 0;
+                buffer = 0;
+            }
+        } else if decode_direct(ch) {
+            out.push_char(ch as char);
+            ctx.advance(1);
+        } else {
+            let start = ctx.position();
+            ctx.advance(1);
+            let replace = ctx.handle_error(
+                errors,
+                start..ctx.position(),
+                Some("unexpected special character"),
+            )?;
+            out.push_wtf8(replace.as_ref());
+        }
+    }
+    let mut consumed = ctx.position();
+    if in_shift && final_decode {
+        if surrogate != 0 || bits >= 6 || (bits > 0 && buffer != 0) {
+            let start = startinpos;
+            let end = ctx.position();
+            let replace =
+                ctx.handle_error(errors, start..end, Some("unterminated shift sequence"))?;
+            out.push_wtf8(replace.as_ref());
+        }
+    } else if in_shift {
+        consumed = startinpos;
+        out.truncate(shift_out_start);
+    }
+    Ok((out, consumed))
+}
@@ -0,0 +1,157 @@
+//! Shared utf-16 / utf-32 unit helpers.
+
+use super::*;
+use crate::wtf8::CodePoint;
+
+/// Byte order for a wide Unicode encoding.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ByteOrder {
+    /// Platform endianness; a leading BOM selects the other side on decode.
+    Native,
+    Little,
+    Big,
+}
+
+impl ByteOrder {
+    #[must_use]
+    pub const fn is_big_endian(self) -> bool {
+        match self {
+            Self::Native => cfg!(target_endian = "big"),
+            Self::Little => false,
+            Self::Big => true,
+        }
+    }
+}
+
+pub(super) fn push_u16(out: &mut Vec<u8>, unit: u16, big_endian: bool) {
+    out.extend_from_slice(&if big_endian {
+        unit.to_be_bytes()
+    } else {
+        unit.to_le_bytes()
+    });
+}
+
+pub(super) fn push_u32(out: &mut Vec<u8>, unit: u32, big_endian: bool) {
+    out.extend_from_slice(&if big_endian {
+        unit.to_be_bytes()
+    } else {
+        unit.to_le_bytes()
+    });
+}
+
+pub(super) fn emit_utf16(out: &mut Vec<u8>, cp: u32, big_endian: bool) {
+    if cp <= 0xffff {
+        push_u16(out, cp as u16, big_endian);
+    } else {
+        let v = cp - 0x10000;
+        push_u16(out, 0xd800 | ((v >> 10) as u16), big_endian);
+        push_u16(out, 0xdc00 | ((v & 0x3ff) as u16), big_endian);
+    }
+}
+
+pub(super) fn emit_utf32(out: &mut Vec<u8>, cp: u32, big_endian: bool) {
+    push_u32(out, cp, big_endian);
+}
+
+pub(super) fn read_u16(data: &[u8], big_endian: bool) -> u16 {
+    let arr = [data[0], data[1]];
+    if big_endian {
+        u16::from_be_bytes(arr)
+    } else {
+        u16::from_le_bytes(arr)
+    }
+}
+
+pub(super) fn read_u32(data: &[u8], big_endian: bool) -> u32 {
+    let arr = [data[0], data[1], data[2], data[3]];
+    if big_endian {
+        u32::from_be_bytes(arr)
+    } else {
+        u32::from_le_bytes(arr)
+    }
+}
+
+pub(super) const fn is_surrogate(cp: u32) -> bool {
+    matches!(cp, 0xd800..=0xdfff)
+}
+
+/// `(big_endian, bom_len, reported_byteorder)` for a native or fixed decode.
+///
+/// `reported_byteorder` is CPython's `-1` / `0` / `1` (`le` / native-no-BOM / `be`).
+pub(super) fn resolve_bom(data: &[u8], order: ByteOrder, unit: usize) -> (bool, usize, i32) {
+    match order {
+        ByteOrder::Little => (false, 0, -1),
+        ByteOrder::Big => (true, 0, 1),
+        ByteOrder::Native if unit == 2 && data.starts_with(&[0xff, 0xfe]) => (false, 2, -1),
+        ByteOrder::Native if unit == 2 && data.starts_with(&[0xfe, 0xff]) => (true, 2, 1),
+        ByteOrder::Native if unit == 4 && data.starts_with(&[0xff, 0xfe, 0x00, 0x00]) => {
+            (false, 4, -1)
+        }
+        ByteOrder::Native if unit == 4 && data.starts_with(&[0x00, 0x00, 0xfe, 0xff]) => {
+            (true, 4, 1)
+        }
+        ByteOrder::Native => (cfg!(target_endian = "big"), 0, 0),
+    }
+}
+
+pub(super) fn encode_wide<Ctx, E>(
+    mut ctx: Ctx,
+    errors: &E,
+    big_endian: bool,
+    bom: bool,
+    unit: usize,
+    reason: &str,
+) -> Result<Vec<u8>, Ctx::Error>
+where
+    Ctx: EncodeContext,
+    E: EncodeErrorHandler<Ctx>,
+{
+    let emit = if unit == 2 { emit_utf16 } else { emit_utf32 };
+    let mut out = Vec::new();
+    if bom {
+        emit(&mut out, 0xfeff, big_endian);
+    }
+    loop {
+        let data = ctx.remaining_data();
+        let mut iter = iter_code_points(data);
+        let Some((i, _)) = iter.find(|(_, c)| is_surrogate(c.to_u32())) else {
+            for (_, c) in iter_code_points(data) {
+                emit(&mut out, c.to_u32(), big_endian);
+            }
+            break;
+        };
+        for (_, c) in iter_code_points(&data[..i.bytes]) {
+            emit(&mut out, c.to_u32(), big_endian);
+        }
+        let err_end = match { iter }.find(|(_, c)| !is_surrogate(c.to_u32())) {
+            Some((j, _)) => ctx.position() + j,
+            None => ctx.data_len(),
+        };
+        let range = (ctx.position() + i)..err_end;
+        let replace = ctx.handle_error(errors, range.clone(), Some(reason))?;
+        match replace {
+            EncodeReplace::Str(s) => {
+                for c in s.as_ref().code_points() {
+                    let cp = c.to_u32();
+                    if is_surrogate(cp) {
+                        return Err(ctx.error_encoding(range, Some(reason)));
+                    }
+                    emit(&mut out, cp, big_endian);
+                }
+            }
+            EncodeReplace::Bytes(b) => {
+                if b.as_ref().len() % unit != 0 {
+                    return Err(ctx.error_encoding(range, Some(reason)));
+                }
+                out.extend_from_slice(b.as_ref());
+            }
+        }
+    }
+    Ok(out)
+}
+
+pub(super) fn push_codepoint(out: &mut Wtf8Buf, cp: u32) {
+    if let Some(c) = CodePoint::from_u32(cp) {
+        out.push(c);
+    }
+}
@@ -88,6 +88,7 @@ windows-sys = { workspace = true, features = [
    "Win32_System_Pipes",
    "Win32_System_Performance",
    "Win32_System_Registry",
+    "Win32_System_Rpc",
    "Win32_System_SystemInformation",
    "Win32_System_SystemServices",
    "Win32_System_Threading",
@@ -93,6 +93,8 @@ pub mod overlapped;
 #[cfg(windows)]
 pub mod testconsole;
 #[cfg(windows)]
+pub mod uuid;
+#[cfg(windows)]
 pub mod winapi;
 #[cfg(windows)]
 pub mod winreg;
@@ -94,6 +94,47 @@ fn copy_grouping(ptr: *const libc::c_char) -> Vec<libc::c_char> {
    out
 }

+/// Every byte of a NUL-terminated C string, `CHAR_MAX` included.
+///
+/// `localeconv().grouping` spells a "repeat last group" / "stop" terminator
+/// as `CHAR_MAX`. Stopping at that value drops the distinction; this reader
+/// keeps it, the way a NUL-only C-string walk does.
+///
+/// # Safety
+///
+/// `ptr` must be null or point to a NUL-terminated C string that remains valid
+/// for the duration of the call.
+pub unsafe fn charp2bytes(ptr: *const libc::c_char) -> Vec<u8> {
+    let mut out = Vec::new();
+    if !ptr.is_null() {
+        let mut cur = ptr;
+        unsafe {
+            while *cur != 0 {
+                out.push(*cur as u8);
+                cur = cur.add(1);
+            }
+        }
+    }
+    out
+}
+
+/// Decimal point, thousands separator and grouping of the current locale,
+/// as the raw bytes `localeconv()` reports. Grouping keeps a `CHAR_MAX`
+/// terminator when the C locale spells one.
+pub fn localeconv_numeric() -> (Vec<u8>, Vec<u8>, Vec<u8>) {
+    let lc = unsafe { localeconv() };
+    if lc.is_null() {
+        return (b".".to_vec(), Vec::new(), Vec::new());
+    }
+    unsafe {
+        (
+            charp2bytes((*lc).decimal_point),
+            charp2bytes((*lc).thousands_sep),
+            charp2bytes((*lc).grouping),
+        )
+    }
+}
+
 pub fn localeconv_data() -> LocaleConv {
    let lc = unsafe { localeconv() };
    unsafe {
@@ -0,0 +1,35 @@
+//! Windows `UuidCreateSequential` — the `_uuid` door `uuid.getnode()` reads.
+
+use windows_sys::Win32::System::Rpc::{
+    RPC_S_OK, RPC_S_UUID_LOCAL_ONLY, RPC_S_UUID_NO_ADDRESS, UuidCreateSequential,
+};
+use windows_sys::core::GUID;
+
+pub const STATUS_OK: i32 = RPC_S_OK;
+pub const STATUS_LOCAL_ONLY: i32 = RPC_S_UUID_LOCAL_ONLY;
+pub const STATUS_NO_ADDRESS: i32 = RPC_S_UUID_NO_ADDRESS;
+
+#[derive(Clone, Copy, Debug)]
+pub struct SequentialUuid {
+    pub bytes: [u8; 16],
+    pub status: i32,
+}
+
+/// One `UuidCreateSequential` call. `status` is the raw RPC status.
+#[must_use]
+pub fn create_sequential() -> SequentialUuid {
+    let mut uuid = GUID::from_u128(0);
+    // SAFETY: `uuid` is a live, aligned `GUID` the callee only writes into.
+    let status = unsafe { UuidCreateSequential(&raw mut uuid) };
+    let mut bytes = [0u8; 16];
+    bytes[0..4].copy_from_slice(&uuid.data1.to_le_bytes());
+    bytes[4..6].copy_from_slice(&uuid.data2.to_le_bytes());
+    bytes[6..8].copy_from_slice(&uuid.data3.to_le_bytes());
+    bytes[8..16].copy_from_slice(&uuid.data4);
+    SequentialUuid { bytes, status }
+}
+
+#[must_use]
+pub fn has_stable_node() -> bool {
+    create_sequential().status == STATUS_OK
+}
@@ -1285,6 +1285,23 @@ pub fn read_file(handle: HANDLE, size: u32) -> io::Result<ReadFileResult> {
    Ok(ReadFileResult { data, error: err })
 }

+pub fn get_named_pipe_handle_state(handle: HANDLE) -> io::Result<u32> {
+    let mut mode = 0u32;
+    unsafe {
+        windows_sys::Win32::System::Pipes::GetNamedPipeHandleStateW(
+            handle,
+            &mut mode,
+            core::ptr::null_mut(),
+            core::ptr::null_mut(),
+            core::ptr::null_mut(),
+            core::ptr::null_mut(),
+            0,
+        )
+    }
+    .check_win32_bool()?;
+    Ok(mode)
+}
+
 pub fn set_named_pipe_handle_state(
    handle: HANDLE,
    mode: Option<u32>,
@@ -13,10 +13,11 @@ mod _codecs {
    use crate::common::wtf8::Wtf8Buf;
    use crate::{
        AsObject, PyObjectRef, PyResult, VirtualMachine,
-        builtins::{PyStrRef, PyUtf8StrRef},
+        builtins::{PyBytesRef, PyStrRef, PyUtf8StrRef},
        codecs,
+        convert::TryFromObject,
        exceptions::nul_char_error,
-        function::{ArgBytesLike, PosArgs},
+        function::{ArgBytesLike, OptionalArg, PosArgs},
    };

    #[pyfunction]
@@ -244,7 +245,345 @@ mod _codecs {
        do_codec!(ascii::decode, args, vm)
    }

-    // TODO: implement these codecs in Rust!
+    fn wide_order(byteorder: i32) -> encodings::utf16::ByteOrder {
+        match byteorder.cmp(&0) {
+            core::cmp::Ordering::Less => encodings::utf16::ByteOrder::Little,
+            core::cmp::Ordering::Greater => encodings::utf16::ByteOrder::Big,
+            core::cmp::Ordering::Equal => encodings::utf16::ByteOrder::Native,
+        }
+    }
+
+    fn warn_escape(
+        note: Option<encodings::unicode_escape::EscapeNote>,
+        vm: &VirtualMachine,
+    ) -> PyResult<()> {
+        if let Some(note) = note {
+            crate::stdlib::_warnings::warn(
+                vm.ctx.exceptions.deprecation_warning,
+                note.message,
+                1,
+                vm,
+            )?;
+        }
+        Ok(())
+    }
+
+    #[pyfunction]
+    fn readbuffer_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
+        rustpython_common::static_cell!(
+            static FUNC: PyObjectRef;
+        );
+        super::delegate_pycodecs(&FUNC, "readbuffer_encode", args, vm)
+    }
+
+    #[derive(FromArgs)]
+    struct EscapeEncodeArgs {
+        #[pyarg(positional)]
+        data: PyBytesRef,
+        #[pyarg(positional, optional)]
+        _errors: OptionalArg<PyUtf8StrRef>,
+    }
+
+    #[pyfunction]
+    fn escape_encode(args: EscapeEncodeArgs, _vm: &VirtualMachine) -> (Vec<u8>, usize) {
+        let encoded = encodings::escape::encode(args.data.as_bytes());
+        (encoded, args.data.len())
+    }
+
+    #[derive(FromArgs)]
+    struct EscapeDecodeArgs {
+        #[pyarg(positional)]
+        data: PyObjectRef,
+        #[pyarg(positional, optional)]
+        errors: OptionalArg<PyUtf8StrRef>,
+    }
+
+    #[pyfunction]
+    fn escape_decode(args: EscapeDecodeArgs, vm: &VirtualMachine) -> PyResult<(Vec<u8>, usize)> {
+        let data = if let Ok(s) = args.data.clone().downcast::<crate::builtins::PyStr>() {
+            s.as_bytes().to_vec()
+        } else {
+            ArgBytesLike::try_from_object(vm, args.data)?
+                .borrow_buf()
+                .to_vec()
+        };
+        let name = args.errors.as_option().map_or("strict", |s| s.as_str());
+        let mode = encodings::escape::EscapeErrorMode::from_name(name).ok_or_else(|| {
+            vm.new_value_error(format!(
+                "decoding error; unknown error handling code: {name}"
+            ))
+        })?;
+        match encodings::escape::decode(&data, mode) {
+            Ok((out, warning)) => {
+                if let Some(message) = warning {
+                    crate::stdlib::_warnings::warn(
+                        vm.ctx.exceptions.deprecation_warning,
+                        message,
+                        1,
+                        vm,
+                    )?;
+                }
+                let consumed = data.len();
+                Ok((out, consumed))
+            }
+            Err(encodings::escape::EscapeDecodeError::TrailingBackslash) => {
+                Err(vm.new_value_error("Trailing \\ in string"))
+            }
+            Err(encodings::escape::EscapeDecodeError::InvalidHex { position }) => {
+                Err(vm.new_value_error(format!("invalid \\x escape at position {position}")))
+            }
+            Err(encodings::escape::EscapeDecodeError::UnknownHandler { name }) => Err(vm
+                .new_value_error(format!(
+                    "decoding error; unknown error handling code: {name}"
+                ))),
+        }
+    }
+
+    #[pyfunction]
+    fn unicode_escape_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(
+            encodings::unicode_escape::ENCODING_NAME,
+            encodings::unicode_escape::encode,
+            vm,
+        )
+    }
+
+    #[derive(FromArgs)]
+    struct EscapeTextDecodeArgs {
+        #[pyarg(positional)]
+        data: PyObjectRef,
+        #[pyarg(positional, optional)]
+        errors: Option<PyUtf8StrRef>,
+        #[pyarg(positional, default = true)]
+        final_decode: bool,
+    }
+
+    impl EscapeTextDecodeArgs {
+        fn bytes(&self, vm: &VirtualMachine) -> PyResult<ArgBytesLike> {
+            if let Ok(s) = self.data.clone().downcast::<crate::builtins::PyStr>() {
+                let bytes = vm.ctx.new_bytes(s.as_bytes().to_vec());
+                ArgBytesLike::try_from_object(vm, bytes.into())
+            } else {
+                ArgBytesLike::try_from_object(vm, self.data.clone())
+            }
+        }
+    }
+
+    #[pyfunction]
+    fn unicode_escape_decode(args: EscapeTextDecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let data = args.bytes(vm)?;
+        let ctx = PyDecodeContext::new(encodings::unicode_escape::ENCODING_NAME, &data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        let (text, consumed, note) =
+            encodings::unicode_escape::decode(ctx, &errors, args.final_decode)?;
+        warn_escape(note, vm)?;
+        Ok((text, consumed))
+    }
+
+    #[pyfunction]
+    fn raw_unicode_escape_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(
+            encodings::raw_unicode_escape::ENCODING_NAME,
+            encodings::raw_unicode_escape::encode,
+            vm,
+        )
+    }
+
+    #[pyfunction]
+    fn raw_unicode_escape_decode(args: EscapeTextDecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let data = args.bytes(vm)?;
+        let ctx = PyDecodeContext::new(encodings::raw_unicode_escape::ENCODING_NAME, &data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        encodings::raw_unicode_escape::decode(ctx, &errors, args.final_decode)
+    }
+
+    #[pyfunction]
+    fn utf_7_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(encodings::utf7::ENCODING_NAME, encodings::utf7::encode, vm)
+    }
+
+    #[pyfunction]
+    fn utf_7_decode(args: DecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let ctx = PyDecodeContext::new(encodings::utf7::ENCODING_NAME, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        encodings::utf7::decode(ctx, &errors, args.final_decode)
+    }
+
+    #[pyfunction]
+    fn utf_16_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(
+            encodings::utf16::ENCODING_NAME,
+            |ctx, errors| {
+                encodings::utf16::encode(ctx, errors, encodings::utf16::ByteOrder::Native, true)
+            },
+            vm,
+        )
+    }
+
+    #[pyfunction]
+    fn utf_16_decode(args: DecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let ctx = PyDecodeContext::new(encodings::utf16::ENCODING_NAME, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        let (text, consumed, _) = encodings::utf16::decode(
+            ctx,
+            &errors,
+            encodings::utf16::ByteOrder::Native,
+            args.final_decode,
+        )?;
+        Ok((text, consumed))
+    }
+
+    #[pyfunction]
+    fn utf_16_le_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(
+            encodings::utf16::ENCODING_NAME_LE,
+            |ctx, errors| {
+                encodings::utf16::encode(ctx, errors, encodings::utf16::ByteOrder::Little, false)
+            },
+            vm,
+        )
+    }
+
+    #[pyfunction]
+    fn utf_16_le_decode(args: DecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let ctx = PyDecodeContext::new(encodings::utf16::ENCODING_NAME_LE, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        let (text, consumed, _) = encodings::utf16::decode(
+            ctx,
+            &errors,
+            encodings::utf16::ByteOrder::Little,
+            args.final_decode,
+        )?;
+        Ok((text, consumed))
+    }
+
+    #[pyfunction]
+    fn utf_16_be_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(
+            encodings::utf16::ENCODING_NAME_BE,
+            |ctx, errors| {
+                encodings::utf16::encode(ctx, errors, encodings::utf16::ByteOrder::Big, false)
+            },
+            vm,
+        )
+    }
+
+    #[pyfunction]
+    fn utf_16_be_decode(args: DecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let ctx = PyDecodeContext::new(encodings::utf16::ENCODING_NAME_BE, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        let (text, consumed, _) = encodings::utf16::decode(
+            ctx,
+            &errors,
+            encodings::utf16::ByteOrder::Big,
+            args.final_decode,
+        )?;
+        Ok((text, consumed))
+    }
+
+    #[derive(FromArgs)]
+    struct ExDecodeArgs {
+        #[pyarg(positional)]
+        data: ArgBytesLike,
+        #[pyarg(positional, optional)]
+        errors: Option<PyUtf8StrRef>,
+        #[pyarg(positional, default = 0)]
+        byteorder: i32,
+        #[pyarg(positional, default = false)]
+        final_decode: bool,
+    }
+
+    #[pyfunction]
+    fn utf_16_ex_decode(
+        args: ExDecodeArgs,
+        vm: &VirtualMachine,
+    ) -> PyResult<(Wtf8Buf, usize, i32)> {
+        let ctx = PyDecodeContext::new(encodings::utf16::ENCODING_NAME, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        encodings::utf16::decode(ctx, &errors, wide_order(args.byteorder), args.final_decode)
+    }
+
+    #[pyfunction]
+    fn utf_32_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(
+            encodings::utf32::ENCODING_NAME,
+            |ctx, errors| {
+                encodings::utf32::encode(ctx, errors, encodings::utf32::ByteOrder::Native, true)
+            },
+            vm,
+        )
+    }
+
+    #[pyfunction]
+    fn utf_32_decode(args: DecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let ctx = PyDecodeContext::new(encodings::utf32::ENCODING_NAME, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        let (text, consumed, _) = encodings::utf32::decode(
+            ctx,
+            &errors,
+            encodings::utf32::ByteOrder::Native,
+            args.final_decode,
+        )?;
+        Ok((text, consumed))
+    }
+
+    #[pyfunction]
+    fn utf_32_le_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(
+            encodings::utf32::ENCODING_NAME_LE,
+            |ctx, errors| {
+                encodings::utf32::encode(ctx, errors, encodings::utf32::ByteOrder::Little, false)
+            },
+            vm,
+        )
+    }
+
+    #[pyfunction]
+    fn utf_32_le_decode(args: DecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let ctx = PyDecodeContext::new(encodings::utf32::ENCODING_NAME_LE, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        let (text, consumed, _) = encodings::utf32::decode(
+            ctx,
+            &errors,
+            encodings::utf32::ByteOrder::Little,
+            args.final_decode,
+        )?;
+        Ok((text, consumed))
+    }
+
+    #[pyfunction]
+    fn utf_32_be_encode(args: EncodeArgs, vm: &VirtualMachine) -> EncodeResult {
+        args.encode(
+            encodings::utf32::ENCODING_NAME_BE,
+            |ctx, errors| {
+                encodings::utf32::encode(ctx, errors, encodings::utf32::ByteOrder::Big, false)
+            },
+            vm,
+        )
+    }
+
+    #[pyfunction]
+    fn utf_32_be_decode(args: DecodeArgs, vm: &VirtualMachine) -> DecodeResult {
+        let ctx = PyDecodeContext::new(encodings::utf32::ENCODING_NAME_BE, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        let (text, consumed, _) = encodings::utf32::decode(
+            ctx,
+            &errors,
+            encodings::utf32::ByteOrder::Big,
+            args.final_decode,
+        )?;
+        Ok((text, consumed))
+    }
+
+    #[pyfunction]
+    fn utf_32_ex_decode(
+        args: ExDecodeArgs,
+        vm: &VirtualMachine,
+    ) -> PyResult<(Wtf8Buf, usize, i32)> {
+        let ctx = PyDecodeContext::new(encodings::utf32::ENCODING_NAME, &args.data, vm);
+        let errors = ErrorsHandler::new(args.errors.as_deref(), vm);
+        encodings::utf32::decode(ctx, &errors, wide_order(args.byteorder), args.final_decode)
+    }

    macro_rules! delegate_pycodecs {
        ($name:ident, $args:ident, $vm:ident) => {{
@@ -255,50 +594,6 @@ mod _codecs {
        }};
    }

-    #[pyfunction]
-    fn readbuffer_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(readbuffer_encode, args, vm)
-    }
-    #[pyfunction]
-    fn escape_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(escape_encode, args, vm)
-    }
-    #[pyfunction]
-    fn escape_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(escape_decode, args, vm)
-    }
-    #[pyfunction]
-    fn unicode_escape_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(unicode_escape_encode, args, vm)
-    }
-    #[pyfunction]
-    fn unicode_escape_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(unicode_escape_decode, args, vm)
-    }
-    #[pyfunction]
-    fn raw_unicode_escape_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(raw_unicode_escape_encode, args, vm)
-    }
-    #[pyfunction]
-    fn raw_unicode_escape_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(raw_unicode_escape_decode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_7_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_7_encode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_7_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_7_decode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_16_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_16_encode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_16_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_16_decode, args, vm)
-    }
    #[pyfunction]
    fn charmap_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
        delegate_pycodecs!(charmap_encode, args, vm)
@@ -311,54 +606,6 @@ mod _codecs {
    fn charmap_build(args: PosArgs, vm: &VirtualMachine) -> PyResult {
        delegate_pycodecs!(charmap_build, args, vm)
    }
-    #[pyfunction]
-    fn utf_16_le_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_16_le_encode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_16_le_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_16_le_decode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_16_be_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_16_be_encode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_16_be_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_16_be_decode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_16_ex_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_16_ex_decode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_32_ex_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_32_ex_decode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_32_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_32_encode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_32_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_32_decode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_32_le_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_32_le_encode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_32_le_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_32_le_decode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_32_be_encode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_32_be_encode, args, vm)
-    }
-    #[pyfunction]
-    fn utf_32_be_decode(args: PosArgs, vm: &VirtualMachine) -> PyResult {
-        delegate_pycodecs!(utf_32_be_decode, args, vm)
-    }
 }

 #[inline]
@@ -0,0 +1,30 @@
+# Shared wide / escape codecs from rustpython-common.
+
+assert "hi".encode("utf-16-le") == b"h\x00i\x00"
+assert "hi".encode("utf-16-be") == b"\x00h\x00i"
+assert b"h\x00i\x00".decode("utf-16-le") == "hi"
+assert b"\x00h\x00i".decode("utf-16-be") == "hi"
+
+# BOM form: native utf-16 writes a BOM and decodes it back.
+assert "hi".encode("utf-16").decode("utf-16") == "hi"
+assert "hi".encode("utf-32").decode("utf-32") == "hi"
+assert "hi".encode("utf-32-le") == b"h\x00\x00\x00i\x00\x00\x00"
+assert b"h\x00\x00\x00i\x00\x00\x00".decode("utf-32-le") == "hi"
+
+assert "café".encode("unicode-escape") == b"caf\\xe9"
+assert b"caf\\xe9".decode("unicode-escape") == "café"
+assert "A\u0100".encode("raw-unicode-escape") == b"A\\u0100"
+assert b"A\\u0100".decode("raw-unicode-escape") == "A\u0100"
+
+assert "+".encode("utf-7") == b"+-"
+assert b"+-".decode("utf-7") == "+"
+
+import _codecs
+
+assert _codecs.escape_encode(b"a\tb") == (b"a\\tb", 3)
+assert _codecs.escape_decode(rb"a\tb") == (b"a\tb", 4)
+assert _codecs.utf_16_ex_decode(b"\xff\xfeh\x00i\x00", "strict", 0, True)[0] == "hi"
+assert (
+    _codecs.utf_32_ex_decode(b"\xff\xfe\x00\x00h\x00\x00\x00", "strict", 0, True)[0]
+    == "h"
+)