Source code for boltons.mboxutils

# -*- 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

[docs]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`_. Args: 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 case. .. _Heirloom mailx: """ def __init__(self, path, factory=None, create=True, maxmem=1024 * 1024): mailbox.mbox.__init__(self, path, factory, create) self.maxmem = maxmem
[docs] 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. mailbox._sync_flush(self._file) self._pending_sync = False return # 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 = {} self._pre_mailbox_hook(new_file) for key in sorted(self._toc.keys()): start, stop = self._toc[key] self._pre_message_hook(new_file) new_start = new_file.tell() while True: buffer =, stop - self._file.tell())) if buffer == '': break new_file.write(buffer) new_toc[key] = (new_start, new_file.tell()) self._post_message_hook(new_file) self._file_length = new_file.tell() # Copy back our messages if self._file_length <= self.maxmem: self._file.write( else: while True: buffer = if not buffer: break self._file.write(buffer) # Delete the rest. self._file.truncate() # Same wrap up. self._toc = new_toc self._pending = False self._pending_sync = False if self._locked: mailbox._lock_file(self._file, dotlock=False)