Skip to content

MailBox

email_profile.clients.imap.mailbox.MailBox

One IMAP mailbox addressed by its server-side name.

Source code in email_profile/clients/imap/mailbox.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class MailBox:
    """One IMAP mailbox addressed by its server-side name."""

    PATTERN = re.compile(
        r'\((?P<flags>.*?)\) "(?P<delimiter>.*)" (?P<name>.*)'
    )

    def __init__(
        self,
        *,
        client: imaplib.IMAP4_SSL,
        name: str,
        delimiter: str = "/",
        flags: Optional[list[str]] = None,
    ) -> None:
        self._client = client
        self.name = name.strip().strip('"')
        self.delimiter = delimiter
        self.flags = flags or []

    @classmethod
    def from_imap_detail(
        cls,
        client: imaplib.IMAP4_SSL,
        detail: bytes,
    ) -> MailBox:
        """Parse one entry from IMAP.list()."""

        match = cls.PATTERN.match(detail.decode("utf-8"))

        if match is None:
            raise ValueError(f"Cannot parse mailbox detail: {detail!r}")

        flags_raw, delimiter, name = match.groups()
        flags = [flag.strip("\\") for flag in flags_raw.split()]

        return cls(client=client, name=name, delimiter=delimiter, flags=flags)

    def where(self, query: Optional[QueryLike] = None, **kwargs) -> Where:
        """Build a lazy search; pass a Query/Q/string OR filter kwargs."""

        if query is not None and kwargs:
            raise TypeError(
                "Pass either a Query/Q expression OR kwargs, not both."
            )

        if query is None:
            query = Query(**kwargs) if kwargs else Query()

        return Where(client=self._client, mailbox=self, query=query)

    def append(
        self,
        message: MessageLike,
        flags: str = "",
        date: Optional[datetime] = None,
    ) -> Optional[AppendedUID]:
        """Upload one message into this mailbox via IMAP APPEND.

        Returns the new ``AppendedUID`` when the server supports UIDPLUS
        (RFC 4315). Returns ``None`` otherwise — the message is still saved.
        """

        from email_profile.serializers.email import Message

        if isinstance(message, Message):
            raw = message.file.encode("utf-8")
            if date is None:
                date = message.date
        elif isinstance(message, str):
            raw = message.encode("utf-8")
        elif isinstance(message, bytes):
            raw = message
        else:
            raise TypeError(
                "append expects Message, bytes or str — "
                f"got {type(message).__name__}"
            )

        date_time = imaplib.Time2Internaldate(
            date.timestamp() if date else time.time()
        )

        do_append = with_retry()(self._client.append)
        status, payload = do_append(_quote(self.name), flags, date_time, raw)
        Status.state((status, payload))
        return _parse_append_uid(payload)

    def _store(self, target: UIDLike, command: str, flag: str) -> None:
        Status.state(self._client.select(_quote(self.name)))
        Status.state(self._client.uid("STORE", _uid_of(target), command, flag))

    def mark_seen(self, target: UIDLike) -> None:
        """Mark a message as read (``\\Seen``)."""
        self._store(target, "+FLAGS", "\\Seen")

    def mark_unseen(self, target: UIDLike) -> None:
        """Mark a message as unread."""
        self._store(target, "-FLAGS", "\\Seen")

    def flag(self, target: UIDLike) -> None:
        """Flag a message (``\\Flagged``)."""
        self._store(target, "+FLAGS", "\\Flagged")

    def unflag(self, target: UIDLike) -> None:
        """Remove the ``\\Flagged`` flag."""
        self._store(target, "-FLAGS", "\\Flagged")

    def delete(self, target: UIDLike, expunge: bool = False) -> None:
        """Mark a message as deleted. Call :meth:`expunge` to commit."""
        self._store(target, "+FLAGS", "\\Deleted")
        if expunge:
            self.expunge()

    def undelete(self, target: UIDLike) -> None:
        """Unmark a message as deleted before the next expunge."""
        self._store(target, "-FLAGS", "\\Deleted")

    def expunge(self) -> None:
        """Permanently remove every message marked as ``\\Deleted``."""
        Status.state(self._client.select(_quote(self.name)))
        Status.state(self._client.expunge())

    def copy(self, target: UIDLike, destination: str) -> None:
        """Copy a message into another mailbox (``UID COPY``)."""
        Status.state(self._client.select(_quote(self.name)))
        Status.state(self._client.uid("COPY", _uid_of(target), destination))

    def move(self, target: UIDLike, destination: str) -> None:
        """Move a message into another mailbox.

        Prefers ``UID MOVE`` (RFC 6851). Falls back to copy + delete + expunge
        when the server does not advertise the MOVE extension.
        """
        Status.state(self._client.select(_quote(self.name)))
        uid = _uid_of(target)
        try:
            Status.state(self._client.uid("MOVE", uid, destination))
        except Exception:
            Status.state(self._client.uid("COPY", uid, destination))
            Status.state(self._client.uid("STORE", uid, "+FLAGS", "\\Deleted"))
            Status.state(self._client.expunge())

    def create(self) -> None:
        """Create this mailbox on the server."""
        Status.state(self._client.create(_quote(self.name)))

    def delete_mailbox(self) -> None:
        """Delete this mailbox from the server."""
        Status.state(self._client.delete(_quote(self.name)))

    def rename_to(self, new_name: str) -> None:
        """Rename this mailbox. Updates ``self.name`` in place."""
        Status.state(self._client.rename(_quote(self.name), _quote(new_name)))
        self.name = new_name

    def __repr__(self) -> str:
        return f"MailBox(name={self.name!r}, flags={self.flags})"

PATTERN = re.compile('\\((?P<flags>.*?)\\) "(?P<delimiter>.*)" (?P<name>.*)') class-attribute instance-attribute

delimiter = delimiter instance-attribute

flags = flags or [] instance-attribute

name = name.strip().strip('"') instance-attribute

__init__(*, client, name, delimiter='/', flags=None)

Source code in email_profile/clients/imap/mailbox.py 
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    *,
    client: imaplib.IMAP4_SSL,
    name: str,
    delimiter: str = "/",
    flags: Optional[list[str]] = None,
) -> None:
    self._client = client
    self.name = name.strip().strip('"')
    self.delimiter = delimiter
    self.flags = flags or []

__repr__()

Source code in email_profile/clients/imap/mailbox.py 
212
213
def __repr__(self) -> str:
    return f"MailBox(name={self.name!r}, flags={self.flags})"

append(message, flags='', date=None)

Upload one message into this mailbox via IMAP APPEND.

Returns the new AppendedUID when the server supports UIDPLUS (RFC 4315). Returns None otherwise — the message is still saved.

Source code in email_profile/clients/imap/mailbox.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def append(
    self,
    message: MessageLike,
    flags: str = "",
    date: Optional[datetime] = None,
) -> Optional[AppendedUID]:
    """Upload one message into this mailbox via IMAP APPEND.

    Returns the new ``AppendedUID`` when the server supports UIDPLUS
    (RFC 4315). Returns ``None`` otherwise — the message is still saved.
    """

    from email_profile.serializers.email import Message

    if isinstance(message, Message):
        raw = message.file.encode("utf-8")
        if date is None:
            date = message.date
    elif isinstance(message, str):
        raw = message.encode("utf-8")
    elif isinstance(message, bytes):
        raw = message
    else:
        raise TypeError(
            "append expects Message, bytes or str — "
            f"got {type(message).__name__}"
        )

    date_time = imaplib.Time2Internaldate(
        date.timestamp() if date else time.time()
    )

    do_append = with_retry()(self._client.append)
    status, payload = do_append(_quote(self.name), flags, date_time, raw)
    Status.state((status, payload))
    return _parse_append_uid(payload)

copy(target, destination)

Copy a message into another mailbox (UID COPY).

Source code in email_profile/clients/imap/mailbox.py 
179
180
181
182
def copy(self, target: UIDLike, destination: str) -> None:
    """Copy a message into another mailbox (``UID COPY``)."""
    Status.state(self._client.select(_quote(self.name)))
    Status.state(self._client.uid("COPY", _uid_of(target), destination))

create()

Create this mailbox on the server.

Source code in email_profile/clients/imap/mailbox.py 
199
200
201
def create(self) -> None:
    """Create this mailbox on the server."""
    Status.state(self._client.create(_quote(self.name)))

delete(target, expunge=False)

Mark a message as deleted. Call :meth:expunge to commit.

Source code in email_profile/clients/imap/mailbox.py 
164
165
166
167
168
def delete(self, target: UIDLike, expunge: bool = False) -> None:
    """Mark a message as deleted. Call :meth:`expunge` to commit."""
    self._store(target, "+FLAGS", "\\Deleted")
    if expunge:
        self.expunge()

delete_mailbox()

Delete this mailbox from the server.

Source code in email_profile/clients/imap/mailbox.py 
203
204
205
def delete_mailbox(self) -> None:
    """Delete this mailbox from the server."""
    Status.state(self._client.delete(_quote(self.name)))

expunge()

Permanently remove every message marked as \Deleted.

Source code in email_profile/clients/imap/mailbox.py 
174
175
176
177
def expunge(self) -> None:
    """Permanently remove every message marked as ``\\Deleted``."""
    Status.state(self._client.select(_quote(self.name)))
    Status.state(self._client.expunge())

flag(target)

Flag a message (\Flagged).

Source code in email_profile/clients/imap/mailbox.py 
156
157
158
def flag(self, target: UIDLike) -> None:
    """Flag a message (``\\Flagged``)."""
    self._store(target, "+FLAGS", "\\Flagged")

from_imap_detail(client, detail) classmethod

Parse one entry from IMAP.list().

Source code in email_profile/clients/imap/mailbox.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@classmethod
def from_imap_detail(
    cls,
    client: imaplib.IMAP4_SSL,
    detail: bytes,
) -> MailBox:
    """Parse one entry from IMAP.list()."""

    match = cls.PATTERN.match(detail.decode("utf-8"))

    if match is None:
        raise ValueError(f"Cannot parse mailbox detail: {detail!r}")

    flags_raw, delimiter, name = match.groups()
    flags = [flag.strip("\\") for flag in flags_raw.split()]

    return cls(client=client, name=name, delimiter=delimiter, flags=flags)

mark_seen(target)

Mark a message as read (\Seen).

Source code in email_profile/clients/imap/mailbox.py 
148
149
150
def mark_seen(self, target: UIDLike) -> None:
    """Mark a message as read (``\\Seen``)."""
    self._store(target, "+FLAGS", "\\Seen")

mark_unseen(target)

Mark a message as unread.

Source code in email_profile/clients/imap/mailbox.py 
152
153
154
def mark_unseen(self, target: UIDLike) -> None:
    """Mark a message as unread."""
    self._store(target, "-FLAGS", "\\Seen")

move(target, destination)

Move a message into another mailbox.

Prefers UID MOVE (RFC 6851). Falls back to copy + delete + expunge when the server does not advertise the MOVE extension.

Source code in email_profile/clients/imap/mailbox.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def move(self, target: UIDLike, destination: str) -> None:
    """Move a message into another mailbox.

    Prefers ``UID MOVE`` (RFC 6851). Falls back to copy + delete + expunge
    when the server does not advertise the MOVE extension.
    """
    Status.state(self._client.select(_quote(self.name)))
    uid = _uid_of(target)
    try:
        Status.state(self._client.uid("MOVE", uid, destination))
    except Exception:
        Status.state(self._client.uid("COPY", uid, destination))
        Status.state(self._client.uid("STORE", uid, "+FLAGS", "\\Deleted"))
        Status.state(self._client.expunge())

rename_to(new_name)

Rename this mailbox. Updates self.name in place.

Source code in email_profile/clients/imap/mailbox.py 
207
208
209
210
def rename_to(self, new_name: str) -> None:
    """Rename this mailbox. Updates ``self.name`` in place."""
    Status.state(self._client.rename(_quote(self.name), _quote(new_name)))
    self.name = new_name

undelete(target)

Unmark a message as deleted before the next expunge.

Source code in email_profile/clients/imap/mailbox.py 
170
171
172
def undelete(self, target: UIDLike) -> None:
    """Unmark a message as deleted before the next expunge."""
    self._store(target, "-FLAGS", "\\Deleted")

unflag(target)

Remove the \Flagged flag.

Source code in email_profile/clients/imap/mailbox.py 
160
161
162
def unflag(self, target: UIDLike) -> None:
    """Remove the ``\\Flagged`` flag."""
    self._store(target, "-FLAGS", "\\Flagged")

where(query=None, **kwargs)

Build a lazy search; pass a Query/Q/string OR filter kwargs.

Source code in email_profile/clients/imap/mailbox.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def where(self, query: Optional[QueryLike] = None, **kwargs) -> Where:
    """Build a lazy search; pass a Query/Q/string OR filter kwargs."""

    if query is not None and kwargs:
        raise TypeError(
            "Pass either a Query/Q expression OR kwargs, not both."
        )

    if query is None:
        query = Query(**kwargs) if kwargs else Query()

    return Where(client=self._client, mailbox=self, query=query)