multipart API¶

def extract_boundary(content_type: str) -> str:
    """Extract the boundary string from a Content-Type header.

    Handles both quoted and unquoted boundaries.  If the string contains
    no ``boundary=``, it is returned as-is (assumed to be a bare boundary).

    Args:
        content_type: Full Content-Type header value or bare boundary string.

    Returns:
        The boundary string.

    Raises:
        MultipartParseError: If Content-Type is multipart/* but has no boundary.
    """
    # Try to find boundary= in the header
    match = re.search(
        r"boundary\s*=\s*(?:\"([^\"]*)\"|([^\s;]+))",
        content_type,
        re.IGNORECASE,
    )
    if match:
        return match.group(1) if match.group(1) is not None else match.group(2)

    # If it looks like a Content-Type header but has no boundary, error
    if "multipart/" in content_type.lower():
        raise MultipartParseError(
            "Content-Type is multipart/* but contains no boundary parameter"
        )

    # Treat the whole string as a bare boundary value
    return content_type.strip()

def parse_multipart(
    body: bytes,
    content_type: str,
    *,
    max_part_size: int = _DEFAULT_MAX_PART_SIZE,
    max_parts: int = _DEFAULT_MAX_PARTS,
) -> list[Part]:
    """Parse a multipart/form-data request body into a list of parts.

    Args:
        body: The raw request body bytes.
        content_type: Full Content-Type header value
            (e.g. ``"multipart/form-data; boundary=abc123"``), or just the
            boundary string itself.
        max_part_size: Maximum size in bytes for any single part.
            Set to 0 to disable.
        max_parts: Maximum number of parts allowed.
            Set to 0 to disable.

    Returns:
        List of Part objects in the order they appeared.

    Raises:
        MultipartParseError: If the body is malformed or limits are exceeded.
    """
    boundary = extract_boundary(content_type)
    return list(_iter_parts(body, boundary, max_part_size, max_parts))

def encode_multipart(
    fields: dict[str, str | bytes] | list[tuple[str, str | bytes]] | None = None,
    files: (
        dict[str, bytes | tuple[str, bytes] | tuple[str, bytes, str]]
        | list[tuple[str, bytes | tuple[str, bytes] | tuple[str, bytes, str]]]
        | None
    ) = None,
    *,
    boundary: str | None = None,
) -> tuple[bytes, str]:
    """Encode form fields and files as a multipart/form-data body.

    Args:
        fields: Text form fields as ``{name: value}`` dict or
            ``[(name, value)]`` list.  Values can be str or bytes.
        files: File uploads as dict or list of tuples.  Each value can be:

            - ``bytes`` -- raw content (auto filename, octet-stream)
            - ``(filename, bytes)`` -- named file
            - ``(filename, bytes, content_type)`` -- named file with MIME type
        boundary: Optional boundary string.  If None, a random one is
            generated.

    Returns:
        ``(body_bytes, content_type_header)`` where content_type_header is
        the full ``"multipart/form-data; boundary=..."`` string.

    Raises:
        MultipartEncodeError: If arguments are invalid.
    """
    if boundary is None:
        boundary = os.urandom(16).hex()

    if len(boundary) > _MAX_BOUNDARY_LEN:
        raise MultipartEncodeError(
            f"Boundary exceeds maximum length of {_MAX_BOUNDARY_LEN}"
        )

    buf = bytearray()
    delim = f"--{boundary}\r\n".encode()

    # Encode text fields
    field_items: list[tuple[str, str | bytes]] = []
    if isinstance(fields, dict):
        field_items = list(fields.items())
    elif fields is not None:
        field_items = list(fields)

    for name, value in field_items:
        buf += delim
        cd = f'Content-Disposition: form-data; name="{_quote_name(name)}"\r\n'
        buf += cd.encode()
        buf += b"\r\n"
        if isinstance(value, bytes):
            buf += value
        else:
            buf += str(value).encode()
        buf += b"\r\n"

    # Encode file uploads
    file_items: list[tuple[str, Any]] = []
    if isinstance(files, dict):
        file_items = list(files.items())
    elif files is not None:
        file_items = list(files)

    for name, file_val in file_items:
        buf += delim
        filename: str
        data: bytes
        ct: str

        if isinstance(file_val, bytes):
            filename = name
            data = file_val
            ct = "application/octet-stream"
        elif isinstance(file_val, tuple):
            if len(file_val) == 2:
                filename, data = file_val  # type: ignore[misc]
                ct = "application/octet-stream"
            elif len(file_val) == 3:
                filename, data, ct = file_val  # type: ignore[misc]
            else:
                raise MultipartEncodeError(
                    f"File tuple for '{name}' must have 2 or 3 elements, "
                    f"got {len(file_val)}"
                )
        else:
            raise MultipartEncodeError(
                f"File value for '{name}' must be bytes or tuple, "
                f"got {type(file_val).__name__}"
            )

        buf += (
            f'Content-Disposition: form-data; name="{_quote_name(name)}"; '
            f'filename="{_quote_name(filename)}"\r\n'
        ).encode()
        buf += f"Content-Type: {ct}\r\n".encode()
        buf += b"\r\n"
        buf += data
        buf += b"\r\n"

    buf += f"--{boundary}--\r\n".encode()

    content_type_header = f"multipart/form-data; boundary={boundary}"
    return bytes(buf), content_type_header