dodil/erp-platform

tests/scenario_bricks/scn_mailer/infrastructure/tables.py

"""mailer's outbox: one row per message sent."""
from sqlalchemy import Column, MetaData, String, Table, TIMESTAMP, insert, select

md = MetaData()
outbox = Table("mailer_outbox", md, Column("message_id", String, primary_key=True), Column("to_addr", String),
               Column("subject", String), Column("sent_at", TIMESTAMP))


class MailerRepository:
    def __init__(self, conn):
        self.conn = conn

    def sent(self, message_id: str) -> bool:
        return self.conn.execute(select(outbox.c.message_id).where(outbox.c.message_id == message_id)).first() is not None

    def add(self, row: dict) -> None:
        self.conn.execute(insert(outbox).values(**row))

    def to(self, address: str) -> list:
        return [dict(r._mapping) for r in self.conn.execute(select(outbox).where(outbox.c.to_addr == address)
                                                            .order_by(outbox.c.message_id))]