### Install and Run Tests with Poetry Source: https://github.com/iroco-co/aioimaplib/blob/main/README.rst Install project dependencies and run tests using Poetry. This is the recommended development workflow. ```bash poetry install poetry run pytest ``` ```bash poetry install poetry shell pytest ``` -------------------------------- ### Configure aioimaplib Logger with YAML Source: https://github.com/iroco-co/aioimaplib/blob/main/README.rst An example of a YAML configuration for the aioimaplib logger, specifying log level and handlers. This requires a YAML parser like PyYAML. ```yaml loggers: ... aioimaplib.aioimaplib: level: DEBUG handlers: [syslog] propagate: no ... ``` -------------------------------- ### Manage IMAP Mailboxes with aioimaplib Source: https://context7.com/iroco-co/aioimaplib/llms.txt Provides examples for common mailbox management tasks including listing, creating, renaming, subscribing, unsubscribing, checking status, and deleting mailboxes. ```python import asyncio from aioimaplib import aioimaplib async def manage_mailboxes(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') # List all mailboxes response = await imap_client.list('""', '*') print('All mailboxes:') for line in response.lines[:-1]: print(f' {line}') # Create a new mailbox response = await imap_client.create('MyFolder') print(f'Create folder: {response.result}') # Create nested mailbox response = await imap_client.create('MyFolder/Subfolder') print(f'Create subfolder: {response.result}') # Rename mailbox response = await imap_client.rename('MyFolder', 'RenamedFolder') print(f'Rename folder: {response.result}') # Subscribe to mailbox (appears in LSUB) response = await imap_client.subscribe('RenamedFolder') print(f'Subscribe: {response.result}') # List subscribed mailboxes response = await imap_client.lsub('""', '*') print('Subscribed mailboxes:') for line in response.lines[:-1]: print(f' {line}') # Unsubscribe from mailbox response = await imap_client.unsubscribe('RenamedFolder') print(f'Unsubscribe: {response.result}') # Get mailbox status response = await imap_client.status('INBOX', '(MESSAGES UNSEEN RECENT)') print(f'INBOX status: {response.lines}') # Delete mailbox response = await imap_client.delete('RenamedFolder') print(f'Delete folder: {response.result}') await imap_client.logout() if __name__ == '__main__': asyncio.run(manage_mailboxes()) ``` -------------------------------- ### Event-Based IDLE Loop Source: https://github.com/iroco-co/aioimaplib/blob/main/README.rst An event-based approach to the IDLE command where the IDLE command is closed at each message from the server. This example fetches messages and then enters IDLE mode. ```python async def idle_loop(host, user, password): imap_client = aioimaplib.IMAP4_SSL(host=host, timeout=30) await imap_client.wait_hello_from_server() await imap_client.login(user, password) await imap_client.select() while True: print((await imap_client.uid('fetch', '1:*', 'FLAGS'))) idle = await imap_client.idle_start(timeout=60) print((await imap_client.wait_server_push())) imap_client.idle_done() await asyncio.wait_for(idle, 30) ``` -------------------------------- ### Send ID and Get Namespace Information Source: https://context7.com/iroco-co/aioimaplib/llms.txt Use the `id` method to exchange client/server identification and `namespace` to retrieve mailbox namespace information. Ensure the server supports the NAMESPACE capability before calling `namespace`. ```python import asyncio from aioimaplib import aioimaplib async def server_identification(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') # Send client identification and receive server info response = await imap_client.id( name='MyMailClient', version='1.0.0', vendor='My Company' ) print(f'Server ID: {response.lines}') # Get namespace information if imap_client.has_capability('NAMESPACE'): response = await imap_client.namespace() print(f'Namespace: {response.lines}') # Returns personal, other users, and shared namespaces await imap_client.logout() if __name__ == '__main__': asyncio.run(server_identification()) ``` -------------------------------- ### Select and Examine Mailboxes, List Mailboxes Source: https://context7.com/iroco-co/aioimaplib/llms.txt Demonstrates selecting a mailbox for read-write access, examining a mailbox in read-only mode, and listing all available mailboxes. Includes extracting message counts using helper functions. ```python import asyncio from aioimaplib import aioimaplib async def work_with_mailboxes(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') # Select INBOX for read-write access response = await imap_client.select('INBOX') print(f'SELECT result: {response.result}') for line in response.lines: print(f' {line}') # Output includes: EXISTS count, RECENT count, FLAGS, UIDVALIDITY # Use extract_exists helper to get message count message_count = aioimaplib.extract_exists(response) print(f'Total messages: {message_count}') # Examine another folder (read-only) response = await imap_client.examine('Sent') print(f'Sent folder has {aioimaplib.extract_exists(response)} messages') # List all mailboxes response = await imap_client.list('""', '*') print('Available mailboxes:') for line in response.lines: print(f' {line}') # Close the selected mailbox (returns to AUTH state) await imap_client.close() await imap_client.logout() if __name__ == '__main__': asyncio.run(work_with_mailboxes()) ``` -------------------------------- ### Connect to IMAP Server with SSL and Authenticate Source: https://context7.com/iroco-co/aioimaplib/llms.txt Establishes an SSL/TLS connection to an IMAP server using IMAP4_SSL, authenticates with username and password, and selects the INBOX. Includes custom timeout configuration. ```python import asyncio from aioimaplib import aioimaplib async def connect_to_mailbox(): # Create SSL client with custom timeout (default is 10 seconds) imap_client = aioimaplib.IMAP4_SSL( host='imap.gmail.com', port=993, timeout=30.0 ) # Wait for server greeting await imap_client.wait_hello_from_server() # Authenticate with username and password response = await imap_client.login('user@gmail.com', 'app-password') if response.result == 'OK': print('Login successful') print(f'Capabilities: {imap_client.protocol.capabilities}') # Select mailbox (INBOX is default) response = await imap_client.select('INBOX') print(f'Messages in INBOX: {response.lines[0]}') # Always logout to properly close connection await imap_client.logout() if __name__ == '__main__': asyncio.run(connect_to_mailbox()) ``` -------------------------------- ### Implement IDLE for Real-Time Push Notifications Source: https://context7.com/iroco-co/aioimaplib/llms.txt Use `idle_start`, `wait_server_push`, and `idle_done` to receive instant notifications for new messages without polling. Handles `STOP_WAIT_SERVER_PUSH` for timeouts and processes different push message types. ```python import asyncio from aioimaplib import aioimaplib from aioimaplib.aioimaplib import STOP_WAIT_SERVER_PUSH async def wait_for_new_messages(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com', timeout=30) await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') await imap_client.select('INBOX') print('Starting IDLE mode - waiting for new messages...') # Start IDLE with 60-second timeout idle = await imap_client.idle_start(timeout=60) while imap_client.has_pending_idle(): # Wait for server push notification msg = await imap_client.wait_server_push() print(f'Server push received: {msg}') # Check for stop signal (timeout reached) if msg == STOP_WAIT_SERVER_PUSH: print('IDLE timeout - ending idle mode') imap_client.idle_done() await asyncio.wait_for(idle, 1) break # Process push messages for push in msg: if push.endswith(b'EXISTS'): print(f'New message arrived: {push}') elif push.endswith(b'EXPUNGE'): print(f'Message deleted: {push}') elif b'FETCH' in push: print(f'Message flags changed: {push}') await imap_client.logout() async def continuous_idle_loop(): """Event-based IDLE loop that processes messages between IDLE cycles.""" imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com', timeout=30) await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') await imap_client.select('INBOX') while True: # Fetch current state before idling response = await imap_client.uid('fetch', '1:*', 'FLAGS') print(f'Current flags: {response.lines}') # Start IDLE idle = await imap_client.idle_start(timeout=60) # Wait for push notification push_msg = await imap_client.wait_server_push() print(f'Push received: {push_msg}') # End IDLE to process the notification imap_client.idle_done() await asyncio.wait_for(idle, 30) # Now we can execute other IMAP commands # Loop continues to re-enter IDLE if __name__ == '__main__': asyncio.run(wait_for_new_messages()) ``` -------------------------------- ### Configure aioimaplib Logger Source: https://github.com/iroco-co/aioimaplib/blob/main/README.rst Set up a stream handler for the aioimaplib logger to view debug messages. This requires importing the logging module. ```python logger = logging.getLogger(__name__) ``` ```python aioimaplib_logger = logging.getLogger('aioimaplib.aioimaplib') sh = logging.StreamHandler() sh.setLevel(logging.DEBUG) sh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s [%(module)s:%(lineno)d] %(message)s")) aioimaplib_logger.addHandler(sh) ``` -------------------------------- ### Connect and Check Mailbox Source: https://github.com/iroco-co/aioimaplib/blob/main/README.rst Connects to an IMAP server using SSL, logs in, selects the INBOX, prints the message count, and logs out. Requires host, user, and password credentials. ```python import asyncio from aioimaplib import aioimaplib async def check_mailbox(host, user, password): imap_client = aioimaplib.IMAP4_SSL(host=host) await imap_client.wait_hello_from_server() await imap_client.login(user, password) res, data = await imap_client.select() print('there is %s messages INBOX' % data[0]) await imap_client.logout() if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(check_mailbox('my.imap.server', 'user', 'pass')) ``` -------------------------------- ### Copy and Move Messages with aioimaplib Source: https://context7.com/iroco-co/aioimaplib/llms.txt Demonstrates copying and moving messages between IMAP folders using UID. MOVE requires server support; a fallback to copy-then-delete is provided. ```python import asyncio from aioimaplib import aioimaplib async def copy_and_move_messages(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') await imap_client.select('INBOX') # Copy message to another folder (by UID) response = await imap_client.uid('copy', '100', 'Archive') if response.result == 'OK': print('Message copied to Archive') # Copy multiple messages response = await imap_client.uid('copy', '100:110', 'Backup') print(f'Copy result: {response.result}') # Move message (requires MOVE capability) if imap_client.has_capability('MOVE'): response = await imap_client.move('100', 'Archive') print(f'Move result: {response.result}') else: # Fallback: copy then delete await imap_client.uid('copy', '100', 'Archive') await imap_client.uid('store', '100', '+FLAGS', '(\Deleted)') await imap_client.expunge() print('Message moved using copy+delete') await imap_client.logout() if __name__ == '__main__': asyncio.run(copy_and_move_messages()) ``` -------------------------------- ### Fetch Message Data with aioimaplib Source: https://context7.com/iroco-co/aioimaplib/llms.txt Use `fetch` or `uid('fetch', ...)` to retrieve message data. The response contains raw bytes that can be parsed using Python's `email` module. Efficiently fetch headers only or specific parts of a message. ```python import asyncio import re from email.parser import BytesParser, BytesHeaderParser from aioimaplib import aioimaplib FETCH_MESSAGE_DATA_UID = re.compile(rb'.*UID (?P\d+).*') FETCH_MESSAGE_DATA_FLAGS = re.compile(rb'.*FLAGS \((?P.*?)\).*') async def fetch_messages(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') await imap_client.select('INBOX') # Fetch headers only (efficient for listing) response = await imap_client.fetch('1:5', '(FLAGS BODY.PEEK[HEADER.FIELDS (FROM SUBJECT DATE)])') if response.result == 'OK': for i in range(0, len(response.lines) - 1, 3): header_data = response.lines[i + 1] headers = BytesHeaderParser().parsebytes(header_data) print(f"From: {headers['From']}") print(f"Subject: {headers['Subject']}") print(f"Date: {headers['Date']}") print('---') # Fetch by UID (recommended) response = await imap_client.uid('fetch', '100', '(UID FLAGS BODY.PEEK[])') if response.result == 'OK' and len(response.lines) > 1: # Parse the complete message message = BytesParser().parsebytes(response.lines[1]) print(f"Subject: {message['Subject']}") print(f"Content-Type: {message.get_content_type()}") # Get message body if message.is_multipart(): for part in message.walk(): if part.get_content_type() == 'text/plain': print(f"Body: {part.get_payload(decode=True)[:200]}...") else: print(f"Body: {message.get_payload(decode=True)[:200]}...") # Fetch only specific parts response = await imap_client.uid('fetch', '1:*', '(UID FLAGS)') if response.result == 'OK': for line in response.lines[:-1]: if isinstance(line, bytes): uid_match = FETCH_MESSAGE_DATA_UID.match(line) flags_match = FETCH_MESSAGE_DATA_FLAGS.match(line) if uid_match and flags_match: print(f"UID: {uid_match.group('uid')}, Flags: {flags_match.group('flags')}") await imap_client.logout() if __name__ == '__main__': asyncio.run(fetch_messages()) ``` -------------------------------- ### Configure Logging for IMAP Debugging Source: https://context7.com/iroco-co/aioimaplib/llms.txt Configure the 'aioimaplib.aioimaplib' logger to DEBUG level and add a StreamHandler to see detailed IMAP protocol communication. This is useful for troubleshooting connection and command issues. Passwords are automatically scrubbed from logs. ```python import asyncio import logging from aioimaplib import aioimaplib # Configure aioimaplib logger aioimaplib_logger = logging.getLogger('aioimaplib.aioimaplib') aioimaplib_logger.setLevel(logging.DEBUG) # Add console handler with formatting handler = logging.StreamHandler() handler.setLevel(logging.DEBUG) handler.setFormatter(logging.Formatter( '%(asctime)s %(levelname)s [%(module)s:%(lineno)d] %(message)s' )) aioimaplib_logger.addHandler(handler) async def debug_imap_session(): # All IMAP protocol messages will now be logged imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() # Login - password will be scrubbed in logs await imap_client.login('user', 'password') await imap_client.select('INBOX') await imap_client.search('ALL') await imap_client.logout() if __name__ == '__main__': asyncio.run(debug_imap_session()) ``` -------------------------------- ### Search Messages with aioimaplib Source: https://context7.com/iroco-co/aioimaplib/llms.txt Use `search` or `uid_search` to find messages based on criteria like ALL, UNSEEN, FROM, SINCE, SUBJECT, or combinations using OR. UID search is recommended for persistent references. ```python import asyncio from aioimaplib import aioimaplib async def search_messages(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') await imap_client.select('INBOX') # Search for all messages response = await imap_client.search('ALL') print(f'All message sequence numbers: {response.lines}') # Search by UID (recommended for persistent references) response = await imap_client.uid_search('ALL') print(f'All message UIDs: {response.lines}') # Search for unread messages response = await imap_client.search('UNSEEN') print(f'Unread messages: {response.lines}') # Search with multiple criteria response = await imap_client.search('FROM', '"sender@example.com"', 'SINCE', '01-Jan-2024') print(f'Messages from sender since Jan 2024: {response.lines}') # Search for messages with specific subject response = await imap_client.search('SUBJECT', '"Important Meeting"') print(f'Messages with subject: {response.lines}') # Combine criteria (AND is implicit) response = await imap_client.search('UNSEEN', 'FLAGGED') print(f'Unread and flagged: {response.lines}') # Use OR for disjunction response = await imap_client.search('OR', 'FROM', '"alice@example.com"', 'FROM', '"bob@example.com"') print(f'Messages from Alice or Bob: {response.lines}') await imap_client.logout() if __name__ == '__main__': asyncio.run(search_messages()) ``` -------------------------------- ### Append Message to Mailbox with aioimaplib Source: https://context7.com/iroco-co/aioimaplib/llms.txt Shows how to append a new email message to an IMAP mailbox, with options for setting flags and internal date. Requires constructing the email message as bytes. ```python import asyncio from datetime import datetime, timezone from email.message import EmailMessage from aioimaplib import aioimaplib async def append_message(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') # Create an email message msg = EmailMessage() msg['From'] = 'sender@example.com' msg['To'] = 'recipient@example.com' msg['Subject'] = 'Test Message' msg['Date'] = datetime.now(timezone.utc).strftime('%a, %d %b %Y %H:%M:%S %z') msg.set_content('This is a test message body.') # Convert to bytes message_bytes = msg.as_bytes() # Append to INBOX response = await imap_client.append( message_bytes, mailbox='INBOX', flags='(\Seen)', # Mark as read date=datetime.now(timezone.utc) # Internal date ) print(f'Append result: {response.result}') print(f'Response: {response.lines}') # Append to Drafts folder without flags response = await imap_client.append( message_bytes, mailbox='Drafts' ) print(f'Append to Drafts: {response.result}') await imap_client.logout() if __name__ == '__main__': asyncio.run(append_message()) ``` -------------------------------- ### Authenticate with OAuth2 using IMAP4_SSL Source: https://context7.com/iroco-co/aioimaplib/llms.txt Connects to an IMAP server using IMAP4_SSL and authenticates using an OAuth2 access token. This is required for providers like Microsoft Outlook and Google Gmail. ```python import asyncio from aioimaplib import IMAP4_SSL async def authenticate_with_oauth2(): # Connect to Outlook using OAuth2 client = IMAP4_SSL('outlook.office365.com') await client.wait_hello_from_server() # Authenticate using OAuth2 access token # Token must be obtained through OAuth2 flow beforehand access_token = 'your-oauth2-access-token' response = await client.xoauth2('user@outlook.com', access_token) if response.result == 'OK': print('OAuth2 authentication successful') # Now you can perform IMAP operations await client.select('INBOX') response = await client.uid_search('1:*', charset='us-ascii') print(f'Message UIDs: {response.lines}') await client.close() else: print(f'Authentication failed: {response.lines}') await client.logout() if __name__ == '__main__': asyncio.run(authenticate_with_oauth2()) ``` -------------------------------- ### Wait for New Mail with IDLE Source: https://github.com/iroco-co/aioimaplib/blob/main/README.rst Uses the IMAP IDLE command to wait for new messages without consuming CPU. Responses are pushed to an async queue. The IDLE mode is exited by sending 'DONE'. ```python async def wait_for_new_message(host, user, password): imap_client = aioimaplib.IMAP4_SSL(host=host) await imap_client.wait_hello_from_server() await imap_client.login(user, password) await imap_client.select() idle = await imap_client.idle_start(timeout=10) while imap_client.has_pending_idle(): msg = await imap_client.wait_server_push() print(msg) if msg == STOP_WAIT_SERVER_PUSH: imap_client.idle_done() await asyncio.wait_for(idle, 1) await imap_client.logout() if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(wait_for_new_message('my.imap.server', 'user', 'pass')) ``` -------------------------------- ### Handle IMAP Response Object Structure Source: https://context7.com/iroco-co/aioimaplib/llms.txt Process IMAP command responses using the `Response` namedtuple, which contains `result` ('OK', 'NO', 'BAD') and `lines` (list of bytes). Check server capabilities using `has_capability` before executing commands that rely on them. ```python import asyncio from aioimaplib import aioimaplib async def handle_responses(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') await imap_client.select('INBOX') # Response structure response = await imap_client.search('ALL') # response.result is 'OK', 'NO', or 'BAD' print(f'Result: {response.result}') # response.lines is a list of bytes print(f'Lines type: {type(response.lines)}') print(f'Number of lines: {len(response.lines)}') for i, line in enumerate(response.lines): print(f'Line {i}: {line} (type: {type(line)})') # Error handling response = await imap_client.search('INVALID', 'CRITERIA') if response.result != 'OK': print(f'Search failed: {response.result}') print(f'Error details: {response.lines}') # Check capabilities before using commands if imap_client.has_capability('IDLE'): print('Server supports IDLE') else: print('Server does not support IDLE') await imap_client.logout() if __name__ == '__main__': asyncio.run(handle_responses()) ``` -------------------------------- ### Debug IMAP Traffic with ngrep Source: https://github.com/iroco-co/aioimaplib/blob/main/README.rst Use ngrep to capture network traffic on the IMAP test port for debugging. This command requires root privileges. ```bash sudo ngrep -d lo port 12345 ``` -------------------------------- ### Modify Message Flags with STORE Command Source: https://context7.com/iroco-co/aioimaplib/llms.txt Use the `uid` command with `STORE` to modify message flags like \Seen, \Flagged, \Deleted, and custom flags. Supports adding, removing, and setting flags for single messages or ranges. ```python import asyncio from aioimaplib import aioimaplib async def modify_message_flags(): imap_client = aioimaplib.IMAP4_SSL(host='imap.server.com') await imap_client.wait_hello_from_server() await imap_client.login('user', 'password') await imap_client.select('INBOX') # Mark message as read (add \Seen flag) response = await imap_client.uid('store', '100', '+FLAGS', '(\Seen)') print(f'Mark as read: {response.result}') # Mark message as flagged/starred response = await imap_client.uid('store', '100', '+FLAGS', '(\Flagged)') print(f'Mark as flagged: {response.result}') # Remove \Seen flag (mark as unread) response = await imap_client.uid('store', '100', '-FLAGS', '(\Seen)') print(f'Mark as unread: {response.result}') # Set flags (replaces all existing flags) response = await imap_client.uid('store', '100', 'FLAGS', '(\Seen \Flagged)') print(f'Set flags: {response.result}') # Mark message for deletion response = await imap_client.uid('store', '100', '+FLAGS', '(\Deleted)') print(f'Mark for deletion: {response.result}') # Permanently remove deleted messages response = await imap_client.expunge() print(f'Expunge result: {response.result}') # Operate on multiple messages response = await imap_client.uid('store', '100:110', '+FLAGS', '(\Seen)') print(f'Mark range as read: {response.result}') await imap_client.logout() if __name__ == '__main__': asyncio.run(modify_message_flags()) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.