wr-common-lib 0.1.5

Shared enums and optional PostgreSQL helpers for email workflows

wr-common-lib

Shared enums 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[db]"   # adds async-db-tools

Package layout

src/wr_common_lib/
├── __init__.py          # __version__
└── email/
    ├── __init__.py      # MailFlow, MailStatus, get_email_content_hash
    ├── constants.py     # MailFlow, MailStatus
    └── db_oper.py       # EmailDbOper, get_email_content_hash

Enums

MailFlow and MailStatus match PostgreSQL mail_flow / mail_status.

from wr_common_lib.email import MailFlow, MailStatus

MailFlow.INBOUND
MailStatus.PENDING.value

Status flows

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

Helpers

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"}],
)

get_email_content_hash normalizes to / cc (comma-separated), strips subject/content, and hashes attachment filenames only.

EmailDbOper

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

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

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

Stored as DB columns mail_to, mail_cc, attachment (jsonb).

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

Example:

task = {
    "imo": 1234567,
    "voyage_id": "550e8400-e29b-41d4-a716-446655440000",
    "mail_from": "[e-mail]",
    "to": "[e-mail]",
    "cc": "",
    "subject": "Noon report",
    "content": "...",
    "attachments": [],
}

email_id, created = await db_oper.insert_email(
    task, MailFlow.INBOUND, MailStatus.RECEIVED
)

update_status

await db_oper.update_status(email_id, MailStatus.SENT)
await db_oper.update_status(email_id, MailStatus.DELIVERED)
await db_oper.update_status(email_id, MailStatus.FAILED)

update_parsed_data

Sets mail_status to PARSED and writes parsed_data (jsonb).

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

Dependencies

install	brings in
wr-common-lib	—
wr-common-lib[db]	async-db-tools

Your application owns the database URL and pool lifecycle.

Publish

pip install -e ".[db]"

uv version 0.1.5          # bump before each PyPI release
export UV_PUBLISH_TOKEN=pypi-...
uv build && uv publish

PyPI does not allow re-uploading the same version.

License

MIT — see LICENSE.