view env/lib/python3.9/site-packages/boltons/ @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
line wrap: on
line source

# -*- coding: utf-8 -*-
"""Useful utilities for working with the `mbox`_-formatted
mailboxes. Credit to Mark Williams for these.

.. _mbox:

import mailbox
import tempfile

DEFAULT_MAXMEM = 4 * 1024 * 1024  # 4MB

class mbox_readonlydir(mailbox.mbox):
    """A subclass of :class:`mailbox.mbox` suitable for use with mboxs
    insides a read-only mail directory, e.g., ``/var/mail``. Otherwise
    the API is exactly the same as the built-in mbox.

    Deletes messages via truncation, in the manner of `Heirloom mailx`_.

        path (str): Path to the mbox file.
        factory (type): Message type (defaults to :class:`rfc822.Message`)
        create (bool): Create mailbox if it does not exist. (defaults
                       to ``True``)
        maxmem (int): Specifies, in bytes, the largest sized mailbox
                      to attempt to copy into memory. Larger mailboxes
                      will be copied incrementally which is more
                      hazardous. (defaults to 4MB)

    .. note::

       Because this truncates and rewrites parts of the mbox file,
       this class can corrupt your mailbox.  Only use this if you know
       the built-in :class:`mailbox.mbox` does not work for your use

    .. _Heirloom mailx:
    def __init__(self, path, factory=None, create=True, maxmem=1024 * 1024):
        mailbox.mbox.__init__(self, path, factory, create)
        self.maxmem = maxmem

    def flush(self):
        """Write any pending changes to disk. This is called on mailbox
        close and is usually not called explicitly.

        .. note::

           This deletes messages via truncation. Interruptions may
           corrupt your mailbox.

        # Appending and basic assertions are the same as in mailbox.mbox.flush.
        if not self._pending:
            if self._pending_sync:
                # Messages have only been added, so syncing the file
                # is enough.
                self._pending_sync = False

        # In order to be writing anything out at all, self._toc must
        # already have been generated (and presumably has been modified
        # by adding or deleting an item).
        assert self._toc is not None

        # Check length of self._file; if it's changed, some other process
        # has modified the mailbox since we scanned it., 2)
        cur_len = self._file.tell()
        if cur_len != self._file_length:
            raise mailbox.ExternalClashError('Size of mailbox file changed '
                                             '(expected %i, found %i)' %
                                             (self._file_length, cur_len))

        # Truncation logic begins here.  Mostly the same except we
        # can use tempfile because we're not doing rename(2).
        with tempfile.TemporaryFile() as new_file:
            new_toc = {}
            for key in sorted(self._toc.keys()):
                start, stop = self._toc[key]
                new_start = new_file.tell()
                while True:
                    buffer =,
                                                 stop - self._file.tell()))
                    if buffer == '':
                new_toc[key] = (new_start, new_file.tell())
            self._file_length = new_file.tell()


            # Copy back our messages
            if self._file_length <= self.maxmem:
                while True:
                    buffer =
                    if not buffer:

            # Delete the rest.

        # Same wrap up.
        self._toc = new_toc
        self._pending = False
        self._pending_sync = False
        if self._locked:
            mailbox._lock_file(self._file, dotlock=False)