Skip to content

Fetch

email_profile.clients.imap.fetch.Fetch

Execute UID FETCH in chunks and yield parsed entries.

Source code in email_profile/clients/imap/fetch.py 
 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
class Fetch:
    """Execute UID FETCH in chunks and yield parsed entries."""

    CHUNK_SIZE = 100

    def __init__(
        self,
        *,
        client: imaplib.IMAP4_SSL,
        mailbox: MailBox,
        spec: F,
        chunk_size: int = CHUNK_SIZE,
    ) -> None:
        self._client = client
        self._mailbox = mailbox
        self._spec = spec.mount()
        self._chunk_size = chunk_size

    def _select(self) -> None:
        from email_profile.clients.imap.mailbox import _quote

        Status.state(self._client.select(_quote(self._mailbox.name)))

    def _fetch_chunk(self, group: list[str]) -> list:
        return Status.state(
            self._client.uid("fetch", ",".join(group), self._spec)
        )

    def chunks(self, uids: list[str]) -> Iterator[list]:
        """Yield raw IMAP response lists per chunk."""
        self._select()
        fetch = with_retry()(self._fetch_chunk)

        for start in range(0, len(uids), self._chunk_size):
            group = uids[start : start + self._chunk_size]
            yield fetch(group)

    def parsed(self, uids: list[str]) -> Iterator[FetchParser]:
        """Yield parsed entries with resolved flags."""
        for fetched in self.chunks(uids):
            yield from FetchParser.iter_entries(fetched)

    @staticmethod
    def fetch_message_ids(
        client: object, uids: list[str], chunk_size: int = 500
    ) -> dict[str, str]:
        """Fetch Message-IDs from server. Returns {uid: message_id}."""

        result: dict[str, str] = {}

        for start in range(0, len(uids), chunk_size):
            group = uids[start : start + chunk_size]

            status, fetched = client.uid(
                "fetch",
                ",".join(group),
                "(BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)])",
            )
            if status != "OK":
                continue

            for entry in fetched:
                if not FetchParser.is_valid(entry):
                    continue

                d = FetchParser(entry)
                uid = d._parse_uid()
                msg_id = d._parse_message_id()

                if uid and msg_id:
                    result[uid] = msg_id

        return result

CHUNK_SIZE = 100 class-attribute instance-attribute

__init__(*, client, mailbox, spec, chunk_size=CHUNK_SIZE)

Source code in email_profile/clients/imap/fetch.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(
    self,
    *,
    client: imaplib.IMAP4_SSL,
    mailbox: MailBox,
    spec: F,
    chunk_size: int = CHUNK_SIZE,
) -> None:
    self._client = client
    self._mailbox = mailbox
    self._spec = spec.mount()
    self._chunk_size = chunk_size

chunks(uids)

Yield raw IMAP response lists per chunk.

Source code in email_profile/clients/imap/fetch.py 
112
113
114
115
116
117
118
119
def chunks(self, uids: list[str]) -> Iterator[list]:
    """Yield raw IMAP response lists per chunk."""
    self._select()
    fetch = with_retry()(self._fetch_chunk)

    for start in range(0, len(uids), self._chunk_size):
        group = uids[start : start + self._chunk_size]
        yield fetch(group)

fetch_message_ids(client, uids, chunk_size=500) staticmethod

Fetch Message-IDs from server. Returns {uid: message_id}.

Source code in email_profile/clients/imap/fetch.py 
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
@staticmethod
def fetch_message_ids(
    client: object, uids: list[str], chunk_size: int = 500
) -> dict[str, str]:
    """Fetch Message-IDs from server. Returns {uid: message_id}."""

    result: dict[str, str] = {}

    for start in range(0, len(uids), chunk_size):
        group = uids[start : start + chunk_size]

        status, fetched = client.uid(
            "fetch",
            ",".join(group),
            "(BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)])",
        )
        if status != "OK":
            continue

        for entry in fetched:
            if not FetchParser.is_valid(entry):
                continue

            d = FetchParser(entry)
            uid = d._parse_uid()
            msg_id = d._parse_message_id()

            if uid and msg_id:
                result[uid] = msg_id

    return result

parsed(uids)

Yield parsed entries with resolved flags.

Source code in email_profile/clients/imap/fetch.py 
121
122
123
124
def parsed(self, uids: list[str]) -> Iterator[FetchParser]:
    """Yield parsed entries with resolved flags."""
    for fetched in self.chunks(uids):
        yield from FetchParser.iter_entries(fetched)