Ahrara for Python

Create an email address from a Python application or test, wait for messages, and extract verification codes or links. Requires Python 3.11+; install with pip install ahrara. It provides synchronous and asynchronous APIs, plus a Robot Framework library. The SDK connects directly to the relay without a running Ahrara CLI.

from ahrara import create_inbox

with create_inbox() as inbox:
    # Submit inbox.address in your application.
    email = inbox.wait_for_email()  # No implicit timeout.

from ahrara import create_inbox_async

async def verify_signup():
    async with await create_inbox_async() as inbox:
        # Submit inbox.address in your application.
        email = await inbox.wait_for_email()
        return email

Creation waits for authentication/subscription. Reception continues before the first wait. Matching receives are consumed once, while stored messages remain available until deleted. Use separate inboxes for parallel tests. Concurrent waits on one inbox are queued and run one at a time. Always close the inbox or use its context manager.

Waiting and continuous reception

wait_for_email(timeout=None, *, subject=None, from_address=None, filter=None, cancel=None) waits indefinitely by default. A positive finite timeout is in seconds and raises TimeoutError. subject and from_address are ASCII case-insensitive substrings. Other native failures expose AhraraError.code. The sync API also accepts a threading.Event as cancel; cancellation raises AhraraError with code cancelled. Async task cancellation raises asyncio.CancelledError. Both preserve the inbox for reuse.

for email in inbox.wait_for_emails(timeout=10):
    verify(email)
    if finished():
        break

inbox.receive_emails(verify, timeout=10)

Async inboxes expose the same stream and callback methods:

from contextlib import aclosing

async with aclosing(inbox.wait_for_emails(timeout=10)) as emails:
    async for email in emails:
        await verify(email)
        if finished():
            break

await inbox.receive_emails(verify, timeout=10)

A finite stream timeout is the total duration, including consumer/callback work, and ends iteration normally. Omit it for an indefinite subscription. Callbacks execute sequentially; async receivers await async callbacks and also accept synchronous ones. Callback errors propagate. Breaking, closing a generator, normal completion, and cancellation keep the inbox open. No messages are prefetched while the consumer handles the previous message. Cancel an active async anext() task before closing its generator; synchronous waits accept a cancellation event.

Cancelled creation waits for its bounded connection attempt, then closes any result. A concurrently committed result may be delivered once before a stream propagates cancellation.

Filters and extracted values

EmailFilter is an optional TypedDict accepted by creation and wait/search calls. At creation, filter discards rejected arrivals before storage. On individual waits, it leaves rejected messages available for a later wait. Every field is ANDed: sender, subject, to, cc, text, and headers={name: value} match case-insensitive substrings. recipient, envelope_sender, and message_id match the whole value, ignoring case. has_attachments selects presence or absence; header names ignore case. Per-wait criteria add to the global receipt filter.

Use filter={"regex": True, "subject": r"^(Verification|Confirm email)$"} to interpret all string conditions as case-insensitive Rust regex patterns; regex queries below are case-sensitive unless ignore_case=True. The subject/from_address wait arguments stay literal. Robot Framework accepts the same dictionary, for example &{filter}= Create Dictionary regex=${TRUE} subject=^Verify$. See the shared filter rules for anchoring and limits.

match = inbox.wait_for_regex_match(
    r"code=(?<code>[0-9]+)",
    filter={"sender": "example.test", "subject": "Verify"},
    ignore_case=True,
)
code = next(group["value"] for group in match["groups"] if group.get("name") == "code")

for match in inbox.wait_for_regex_matches(r"[0-9]+", timeout=10):
    verify_code(match["value"])

link = inbox.wait_for_xpath('//a[@id="verify"]/@href')
for value in inbox.wait_for_xpath_values("//a/@href", timeout=10):
    verify_link(value)

AsyncInbox provides the same singular methods with await and plural methods with async for. Regex uses shared Rust syntax, including named captures; lookaround/backreferences are unsupported. Options are source="text" (full decoded text), "html", "subject", or "raw", plus ignore_case, multi_line, and dot_matches_new_line. Results contain email_id, value, and groups; group zero is the full match and optional unmatched groups contain None. XPath evaluates the original HTML and returns strings.

Singular searches throw TimeoutError on finite expiry. Plural searches yield all matches/values across multiple emails and end normally at their total deadline, including consumer work. None/omitted timeout is indefinite. Invalid queries or filters raise AhraraError before consuming messages. Sync cancellation uses cancel=threading.Event(); async cancellation uses the task. Closing a generator preserves the inbox. Each matched email is consumed once, so stopping midway through its results does not replay unused values; stored mail remains readable.

Persistence and credentials

Pass profile_path='/absolute/profile' to retain identity, address, encrypted history, and consumed-receive state across reopen. Omit it for an isolated temporary profile. The same active profile cannot be opened twice.

credentials = inbox.export_credentials()  # Await this for AsyncInbox.
inbox.close()
with create_inbox(credentials=credentials) as restored:
    email = restored.wait_for_email(timeout=10)

Exported InboxCredentials(address, recovery_key) restores identity/address with fresh local history unless a persistent profile is supplied. Its repr is redacted; access recovery_key only for deliberate secure storage and never log it. Alternatively pass recovery_key with optional owned address; these options cannot be combined with credentials. An address alone needs an existing profile identity. Mismatched keys or unowned addresses fail.

Connection options are server_url, server_public_key, tls_ca_path, and connect_timeout_ms (1..300000 ms, default 20000). Omit them for the hosted relay, whose verified public key the package embeds. Custom CA files keep peer and hostname verification enabled.

Email CRUD and attachments

All operations are scoped to the inbox's address and have async counterparts. Reading stored messages does not consume receive state.

Email fields follow the native schema, including from, to, subject, text_body, attachments, and parse_error. The core caps extracted text at 256 KiB and reports text_body_truncated. Message content is untrusted data.

pytest and Robot Framework

pytest can use the context manager in a yield fixture. Robot imports ahrara.robot.AhraraLibrary; its optional constructor arguments are server_url, server_public_key, tls_ca_path, and profile_path.

*** Settings ***
Library    ahrara.robot.AhraraLibrary
Test Teardown    Close Inbox

*** Test Cases ***
Verify Signup
    ${address}=    Create Inbox
    # Submit ${address} in the application here.
    ${email}=    Wait For Email    subject=Verify
    ${page}=    List Emails
    ${stored}=    Get Email    ${email}[id]

Robot's wait has no implicit timeout. The library also provides credentials, CRUD, attachment, raw-email, HTML, Wait For Regex Match/Matches, and Wait For XPath/Values keywords; always use Close Inbox as teardown.