~homeworkprod/byceps

18cb21764cf12f2762ba6c62bd4e8fa819a82b3c — Jochen Kupperschmidt a month ago 918cda2
Extract configuration-related email service
M byceps/blueprints/admin/brand/views.py => byceps/blueprints/admin/brand/views.py +6 -6
@@ 13,7 13,7 @@ from ....services.brand import (
    service as brand_service,
    settings_service as brand_settings_service,
)
from ....services.email import service as email_service
from ....services.email import config_service as email_config_service
from ....services.orga import service as orga_service
from ....services.party import service as party_service
from ....util.framework.blueprint import create_blueprint


@@ 53,7 53,7 @@ def view(brand_id):
    brand = _get_brand_or_404(brand_id)

    settings = brand_settings_service.get_settings(brand.id)
    email_config = email_service.get_config(brand.id)
    email_config = email_config_service.get_config(brand.id)

    return {
        'brand': brand,


@@ 88,7 88,7 @@ def create():

    brand = brand_service.create_brand(brand_id, title)

    email_service.create_config(
    email_config_service.create_config(
        brand.id,
        sender_address=f'noreply@{brand.id}.example',
        sender_name=brand.title,


@@ 151,7 151,7 @@ def email_config_update_form(brand_id, erroneous_form=None):
    """Show form to update e-mail config."""
    brand = _get_brand_or_404(brand_id)

    config = email_service.get_config(brand.id)
    config = email_config_service.get_config(brand.id)

    form = (
        erroneous_form


@@ 176,7 176,7 @@ def email_config_update(brand_id):
    """Update e-mail config."""
    brand = _get_brand_or_404(brand_id)

    config = email_service.get_config(brand.id)
    config = email_config_service.get_config(brand.id)

    form = EmailConfigUpdateForm(request.form)
    if not form.validate():


@@ 186,7 186,7 @@ def email_config_update(brand_id):
    sender_name = form.sender_name.data.strip()
    contact_address = form.contact_address.data.strip()

    config = email_service.update_config(
    config = email_config_service.update_config(
        config.brand_id, sender_address, sender_name, contact_address
    )


M byceps/blueprints/admin/shop/email/views.py => byceps/blueprints/admin/shop/email/views.py +2 -2
@@ 11,7 11,7 @@ from typing import Optional
from flask import abort, current_app, g

from .....services.brand import service as brand_service
from .....services.email import service as email_service
from .....services.email import config_service as email_config_service
from .....services.shop.order.email import (
    example_service as example_order_email_service,
)


@@ 33,7 33,7 @@ def view_for_shop(shop_id):

    brand = brand_service.get_brand(shop.brand_id)

    email_config = email_service.get_config(shop.brand_id)
    email_config = email_config_service.get_config(shop.brand_id)

    example_placed_order_message_text = _get_example_placed_order_message_text(
        shop.id

M byceps/blueprints/common/authentication/password/views.py => byceps/blueprints/common/authentication/password/views.py +5 -2
@@ 13,7 13,10 @@ from .....services.authentication.password import (
    reset_service as password_reset_service,
    service as password_service,
)
from .....services.email import service as email_service
from .....services.email import (
    config_service as email_config_service,
    service as email_service,
)
from .....services.email.transfer.models import NameAndAddress
from .....services.user import service as user_service
from .....services.verification_token import (


@@ 151,7 154,7 @@ def request_reset():

def _get_sender() -> NameAndAddress:
    if g.app_mode.is_site():
        email_config = email_service.get_config(g.brand_id)
        email_config = email_config_service.get_config(g.brand_id)
        return email_config.sender
    else:
        default_sender = current_app.config['MAIL_DEFAULT_SENDER']

M byceps/blueprints/site/ticketing/notification_service.py => byceps/blueprints/site/ticketing/notification_service.py +5 -2
@@ 9,7 9,10 @@ byceps.blueprints.site.ticketing.notification_service
from flask import g
from flask_babel import gettext

from ....services.email import service as email_service
from ....services.email import (
    config_service as email_config_service,
    service as email_service,
)
from ....services.party import service as party_service
from ....services.site import service as site_service
from ....services.ticketing.dbmodels.ticket import Ticket


@@ 153,7 156,7 @@ def _get_party_title():

def _enqueue_email(recipient: User, subject: str, body: str) -> None:
    site = site_service.get_site(g.site_id)
    email_config = email_service.get_config(site.brand_id)
    email_config = email_config_service.get_config(site.brand_id)
    sender = email_config.sender

    recipient_address = user_service.get_email_address(recipient.id)

A byceps/services/email/config_service.py => byceps/services/email/config_service.py +152 -0
@@ 0,0 1,152 @@
"""
byceps.services.email.config_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2014-2022 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from __future__ import annotations
from typing import Optional

from sqlalchemy.exc import IntegrityError

from ...database import db, upsert
from ...typing import BrandID

from .dbmodels import EmailConfig as DbEmailConfig
from .transfer.models import EmailConfig, NameAndAddress


class UnknownEmailConfigId(ValueError):
    pass


def create_config(
    brand_id: BrandID,
    sender_address: str,
    *,
    sender_name: Optional[str] = None,
    contact_address: Optional[str] = None,
) -> EmailConfig:
    """Create a configuration."""
    config = DbEmailConfig(
        brand_id,
        sender_address,
        sender_name=sender_name,
        contact_address=contact_address,
    )

    db.session.add(config)
    db.session.commit()

    return _db_entity_to_config(config)


def update_config(
    brand_id: BrandID,
    sender_address: str,
    sender_name: Optional[str],
    contact_address: Optional[str],
) -> EmailConfig:
    """Update a configuration."""
    config = _find_db_config(brand_id)

    if config is None:
        raise UnknownEmailConfigId(
            f'No e-mail config found for brand ID "{brand_id}"'
        )

    config.sender_address = sender_address
    config.sender_name = sender_name
    config.contact_address = contact_address

    db.session.commit()

    return _db_entity_to_config(config)


def delete_config(brand_id: BrandID) -> bool:
    """Delete a configuration.

    It is expected that no database records (sites) refer to the
    configuration anymore.

    Return `True` on success, or `False` if an error occurred.
    """
    get_config(brand_id)  # Verify ID exists.

    try:
        db.session \
            .query(DbEmailConfig) \
            .filter_by(brand_id=brand_id) \
            .delete()

        db.session.commit()
    except IntegrityError:
        db.session.rollback()
        return False

    return True


def _find_db_config(brand_id: BrandID) -> Optional[DbEmailConfig]:
    return db.session \
        .query(DbEmailConfig) \
        .filter_by(brand_id=brand_id) \
        .one_or_none()


def get_config(brand_id: BrandID) -> EmailConfig:
    """Return the configuration, or raise an error if none is configured
    for that brand.
    """
    config = _find_db_config(brand_id)

    if config is None:
        raise UnknownEmailConfigId(
            f'No e-mail config found for brand ID "{brand_id}"'
        )

    return _db_entity_to_config(config)


def set_config(
    brand_id: BrandID,
    sender_address: str,
    *,
    sender_name: Optional[str] = None,
    contact_address: Optional[str] = None,
) -> None:
    """Add or update configuration for that ID."""
    table = DbEmailConfig.__table__
    identifier = {
        'brand_id': brand_id,
        'sender_address': sender_address,
    }
    replacement = {
        'sender_name': sender_name,
        'contact_address': contact_address,
    }

    upsert(table, identifier, replacement)


def get_all_configs() -> list[EmailConfig]:
    """Return all configurations."""
    configs = db.session.query(DbEmailConfig).all()

    return [_db_entity_to_config(config) for config in configs]


def _db_entity_to_config(config: DbEmailConfig) -> EmailConfig:
    sender = NameAndAddress(
        name=config.sender_name,
        address=config.sender_address,
    )

    return EmailConfig(
        brand_id=config.brand_id,
        sender=sender,
        contact_address=config.contact_address,
    )

M byceps/services/email/service.py => byceps/services/email/service.py +1 -141
@@ 8,138 8,11 @@ byceps.services.email.service

from __future__ import annotations
from email.utils import parseaddr
from typing import Optional

from sqlalchemy.exc import IntegrityError

from ...database import db, upsert
from ... import email
from ...typing import BrandID
from ...util.jobqueue import enqueue

from .dbmodels import EmailConfig as DbEmailConfig
from .transfer.models import EmailConfig, Message, NameAndAddress


class UnknownEmailConfigId(ValueError):
    pass


def create_config(
    brand_id: BrandID,
    sender_address: str,
    *,
    sender_name: Optional[str] = None,
    contact_address: Optional[str] = None,
) -> EmailConfig:
    """Create a configuration."""
    config = DbEmailConfig(
        brand_id,
        sender_address,
        sender_name=sender_name,
        contact_address=contact_address,
    )

    db.session.add(config)
    db.session.commit()

    return _db_entity_to_config(config)


def update_config(
    brand_id: BrandID,
    sender_address: str,
    sender_name: Optional[str],
    contact_address: Optional[str],
) -> EmailConfig:
    """Update a configuration."""
    config = _find_db_config(brand_id)

    if config is None:
        raise UnknownEmailConfigId(
            f'No e-mail config found for brand ID "{brand_id}"'
        )

    config.sender_address = sender_address
    config.sender_name = sender_name
    config.contact_address = contact_address

    db.session.commit()

    return _db_entity_to_config(config)


def delete_config(brand_id: BrandID) -> bool:
    """Delete a configuration.

    It is expected that no database records (sites) refer to the
    configuration anymore.

    Return `True` on success, or `False` if an error occurred.
    """
    get_config(brand_id)  # Verify ID exists.

    try:
        db.session \
            .query(DbEmailConfig) \
            .filter_by(brand_id=brand_id) \
            .delete()

        db.session.commit()
    except IntegrityError:
        db.session.rollback()
        return False

    return True


def _find_db_config(brand_id: BrandID) -> Optional[DbEmailConfig]:
    return db.session \
        .query(DbEmailConfig) \
        .filter_by(brand_id=brand_id) \
        .one_or_none()


def get_config(brand_id: BrandID) -> EmailConfig:
    """Return the configuration, or raise an error if none is configured
    for that brand.
    """
    config = _find_db_config(brand_id)

    if config is None:
        raise UnknownEmailConfigId(
            f'No e-mail config found for brand ID "{brand_id}"'
        )

    return _db_entity_to_config(config)


def set_config(
    brand_id: BrandID,
    sender_address: str,
    *,
    sender_name: Optional[str] = None,
    contact_address: Optional[str] = None,
) -> None:
    """Add or update configuration for that ID."""
    table = DbEmailConfig.__table__
    identifier = {
        'brand_id': brand_id,
        'sender_address': sender_address,
    }
    replacement = {
        'sender_name': sender_name,
        'contact_address': contact_address,
    }

    upsert(table, identifier, replacement)


def get_all_configs() -> list[EmailConfig]:
    """Return all configurations."""
    configs = db.session.query(DbEmailConfig).all()

    return [_db_entity_to_config(config) for config in configs]
from .transfer.models import Message, NameAndAddress


def parse_address(address_str: str) -> NameAndAddress:


@@ 175,16 48,3 @@ def send_email(
) -> None:
    """Send e-mail."""
    email.send(sender, recipients, subject, body)


def _db_entity_to_config(config: DbEmailConfig) -> EmailConfig:
    sender = NameAndAddress(
        name=config.sender_name,
        address=config.sender_address,
    )

    return EmailConfig(
        brand_id=config.brand_id,
        sender=sender,
        contact_address=config.contact_address,
    )

M byceps/services/shop/order/email/service.py => byceps/services/shop/order/email/service.py +5 -2
@@ 14,7 14,10 @@ from typing import Iterator

from flask_babel import format_currency, format_date, gettext

from .....services.email import service as email_service
from .....services.email import (
    config_service as email_config_service,
    service as email_service,
)
from .....services.email.transfer.models import Message
from .....services.shop.order import service as order_service
from .....services.shop.order.transfer.order import Order, OrderID


@@ 230,7 233,7 @@ def _assemble_email_to_orderer(
    recipient_address: str,
) -> Message:
    """Assemble an email message with the rendered template as its body."""
    config = email_service.get_config(brand_id)
    config = email_config_service.get_config(brand_id)
    sender = config.sender
    recipients = [recipient_address]


M byceps/services/user/email_address_service.py => byceps/services/user/email_address_service.py +6 -3
@@ 19,7 19,10 @@ from ...events.user import (
from ...typing import UserID
from ...util.l10n import force_user_locale

from ..email import service as email_service
from ..email import (
    config_service as email_config_service,
    service as email_service,
)
from ..email.transfer.models import NameAndAddress
from ..site import service as site_service
from ..site.transfer.models import SiteID


@@ 41,7 44,7 @@ def send_email_address_confirmation_email_for_site(
) -> None:
    site = site_service.get_site(site_id)

    email_config = email_service.get_config(site.brand_id)
    email_config = email_config_service.get_config(site.brand_id)
    sender = email_config.sender

    send_email_address_confirmation_email(


@@ 189,7 192,7 @@ def send_email_address_change_email_for_site(
) -> None:
    site = site_service.get_site(site_id)

    email_config = email_service.get_config(site.brand_id)
    email_config = email_config_service.get_config(site.brand_id)
    sender = email_config.sender

    send_email_address_change_email(

M byceps/services/user_message/service.py => byceps/services/user_message/service.py +5 -2
@@ 17,7 17,10 @@ from ...typing import UserID
from ...util.l10n import force_user_locale
from ...util import templating

from ..email import service as email_service
from ..email import (
    config_service as email_config_service,
    service as email_service,
)
from ..email.transfer.models import Message
from ..site import service as site_service
from ..site.transfer.models import SiteID


@@ 51,7 54,7 @@ def create_message(
    sender = _get_user(sender_id)
    recipient = _get_user(recipient_id)
    site = site_service.get_site(site_id)
    email_config = email_service.get_config(site.brand_id)
    email_config = email_config_service.get_config(site.brand_id)

    recipients = [_get_user_address(recipient)]


M tests/integration/blueprints/admin/brand/test_views.py => tests/integration/blueprints/admin/brand/test_views.py +2 -2
@@ 4,7 4,7 @@
"""

import byceps.services.brand.service as brand_service
import byceps.services.email.service as email_service
import byceps.services.email.config_service as email_config_service


def test_index(brand_admin_client, brand):


@@ 43,7 43,7 @@ def test_create(brand_admin_client):
    assert brand.id == brand_id
    assert brand.title == title

    email_config = email_service.get_config(brand.id)
    email_config = email_config_service.get_config(brand.id)
    assert email_config.sender is not None
    assert email_config.sender.address == 'noreply@galant.example'
    assert email_config.sender.name == 'gaLANt'

M tests/integration/conftest.py => tests/integration/conftest.py +3 -3
@@ 17,7 17,7 @@ from byceps.services.board import board_service
from byceps.services.board.transfer.models import Board, BoardID
from byceps.services.brand import service as brand_service
from byceps.services.brand.transfer.models import Brand
from byceps.services.email import service as email_service
from byceps.services.email import config_service as email_config_service
from byceps.services.email.transfer.models import EmailConfig
from byceps.services.language import service as language_service
from byceps.services.party.transfer.models import Party


@@ 185,14 185,14 @@ def make_email_config(admin_app: Flask):
        if sender_address is None:
            sender_address = f'{generate_token()}@domain.example'

        email_service.set_config(
        email_config_service.set_config(
            brand_id,
            sender_address,
            sender_name=sender_name,
            contact_address=contact_address,
        )

        return email_service.get_config(brand_id)
        return email_config_service.get_config(brand_id)

    return _wrapper