wr-common-lib 0.1.10

Shared enums, Resend provider, voyage matching, and optional PostgreSQL email helpers

wr-common-lib

Shared enums, Resend provider, inbound voyage matching helpers, and optional PostgreSQL write helpers for the email table.

PyPI	wr-common-lib
import	wr_common_lib

Requires Python 3.12+.

Install

pip install wr-common-lib
pip install "wr-common-lib[resend]"   # ResendProvider (httpx)
pip install "wr-common-lib[db]"       # EmailDbOper (async-db-tools)
pip install "wr-common-lib[db,resend]"

Package layout

src/wr_common_lib/
├── __init__.py          # __version__
└── email/
    ├── __init__.py      # public exports
    ├── constants.py     # MailFlow, MailStatus, VoyageStatus, PassVia, CustomerRole
    ├── encoding.py      # file_to_base64, file_to_attachment
    ├── provider.py      # ResendProvider, SendResult, ReceivedEmail
    └── db_oper.py       # EmailDbOper, get_email_content_hash

Enums

Aligned with PostgreSQL enum types.

from wr_common_lib.email import (
    MailFlow,
    MailStatus,
    VoyageStatus,
    PassVia,
    CustomerRole,
)

MailFlow.INBOUND
MailStatus.PENDING.value
VoyageStatus.UNDERWAY
PassVia.SUE
CustomerRole.OWNER

Python	PostgreSQL
MailFlow	mail_flow
MailStatus	mail_status
VoyageStatus	voyage_status
PassVia	pass_via_enum
CustomerRole	customer_role_enum

Mail status flows

Outbound:  PENDING → SENT | FAILED；webhook → DELIVERED | BOUNCED
Inbound:   RECEIVED → PARSED | PARSE_FAILED

Content hash

from wr_common_lib.email import get_email_content_hash

content_hash = get_email_content_hash(
    to="[e-mail],[e-mail]",
    subject="...",
    content="...",
    cc="",
    attachments=[{"filename": "report.pdf"}],
)

Normalizes to / cc (comma-separated), strips subject/content, and hashes attachment filenames only (not file bytes).

Inbound parsing helpers

from wr_common_lib.email import extract_ship_candidates, extract_imo_from_content

extract_ship_candidates("MV OCEAN STAR - noon report")
extract_imo_from_content("IMO 9682930")   # "9682930" or None

Attachment encoding

Converts local files to Base64 for Resend attachments[].content.

from wr_common_lib.email import file_to_base64, file_to_attachment

b64 = file_to_base64("/path/to/report.pdf")

att = file_to_attachment("/path/to/report.pdf")
# {"filename": "report.pdf", "content": "<base64>", "content_type": "application/pdf"}

ResendProvider

Requires httpx ([resend] extra). Uses the Resend API.

from wr_common_lib.email import EmailProvider, ResendProvider, file_to_attachment

provider = ResendProvider(
    api_key="re_...",
    default_from="[e-mail]",
    default_from_name="My App",
)

# Send (attachments from file_to_attachment)
result = await provider.send(
    to="[e-mail]",
    cc="",
    subject="Hello",
    content="Plain text body",
    attachments=[file_to_attachment("/path/to/file.pdf")],
    mail_from=None,           # optional override
    email_id=email_id,        # optional, sets X-Email-ID header
)
# result.ok, result.message_id

# Inbound (Receiving API)
mail = await provider.received(message_id)
# mail.ok, mail.subject, mail.text, mail.html, mail.from_addr, mail.to, mail.cc, ...
body = mail.content   # text, else html

EmailDbOper

Requires the [db] extra. Pass an async-db-tools PostgresPool (or compatible pool with fetchval / execute).

from wr_common_lib.email import EmailDbOper, MailFlow, MailStatus, get_email_content_hash

db_oper = EmailDbOper(pool)

Write operations only; reads should use db_oper._db directly (e.g. fetchrow, fetchval).

insert_email

email_id, created = await db_oper.insert_email(
    task,
    MailFlow.OUTBOUND,
    MailStatus.PENDING,
)
# created is True when a new row was inserted, False when content_hash already existed

Task fields (inbound and outbound use the same shape):

Field	Required	Notes
imo	yes
voyage_id	yes	UUID string
mail_from	yes	Sender
to	yes	Recipients (comma-separated allowed)
cc	no	Default ""
subject	no	Default ""
content	no	Default ""
attachments	no	List of dicts with filename
content_hash	no	Computed via get_email_content_hash when omitted
created_user_id	no	Outbound only

On duplicate content_hash, the insert is skipped (ON CONFLICT DO NOTHING); the existing row id is returned and created is False.

update_status / update_parsed_data

await db_oper.update_status(email_id, MailStatus.SENT)
await db_oper.update_parsed_data(email_id, {"parsed": "..."})

get_imo_and_voyage_id

Resolves (imo, voyage_id) for inbound mail when exactly one active voyage matches (READY / UNDERWAY):

1. mail_from in voyage.ship_emails
2. Vessel name from subject (extract_ship_candidates)
3. IMO in body (extract_imo_from_content)

Returns (None, None) if ambiguous or not found.

imo, voyage_id = await db_oper.get_imo_and_voyage_id(
    mail_from="[e-mail]",
    subject="MV OCEAN STAR report",
    content="IMO 9682930 ...",
)

Dependencies

install	brings in
wr-common-lib	enums, encoding, hash/parsing helpers, ResendProvider (needs httpx — use [resend])
wr-common-lib[db]	async-db-tools, EmailDbOper, get_imo_and_voyage_id
wr-common-lib[resend]	httpx

Your application owns API keys, database URL, and pool lifecycle.

Publish

pip install -e ".[db,resend]"

# 1. Update CHANGELOG.md for the new version
# 2. Bump version and publish
uv version 0.1.11
export UV_PUBLISH_TOKEN=pypi-...
uv build && uv publish

git tag v0.1.11 && git push --tags

PyPI does not allow re-uploading the same version.

License

MIT — see LICENSE.