byceps.services.email.service.parse_address() - Code Metrics - Inspection of "Use actual service, existing app to speed up user..." - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 1f94b4...a29a2e )

by Jochen

created 2021-09-26 00:15 UTC

byceps.services.email.service.parse_address() A

↳ Parent: byceps.services.email.service

Complexity

Conditions

Size

Total Lines	8
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	7.608

Importance

Changes

Metric	Value
cc	3
eloc	5
nop	1
dl	0
loc	8
ccs	1
cts	5
cp	0.2
crap	7.608
rs	10
c	0
b	0
f	0

"""
byceps.services.email.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

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]


def parse_address(address_str: str) -> NameAndAddress:
    """Parse a string into name and address parts."""
    name, address = parseaddr(address_str)

    if not name and not address:
        raise ValueError(f'Could not parse name and address value: "{address}"')

    return NameAndAddress(name, address)


def enqueue_message(message: Message) -> None:
    """Enqueue e-mail to be sent asynchronously."""
    enqueue_email(
        message.sender, message.recipients, message.subject, message.body
    )


def enqueue_email(
    sender: NameAndAddress,
    recipients: list[str],
    subject: str,
    body: str,
) -> None:
    """Enqueue e-mail to be sent asynchronously."""
    sender_str = sender.format()
    enqueue(send_email, sender_str, recipients, subject, body)


def send_email(
    sender: str, recipients: list[str], subject: str, body: str
) -> 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,
    )


1		"""
2		byceps.services.email.service
3		~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5		:Copyright: 2006-2021 Jochen Kupperschmidt
6		:License: Revised BSD (see `LICENSE` file for details)
7		"""
8
9	1	from __future__ import annotations
10	1	from email.utils import parseaddr
11	1	from typing import Optional
12
13	1	from sqlalchemy.exc import IntegrityError
14
15	1	from ...database import db, upsert
16	1	from ... import email
17	1	from ...typing import BrandID
18	1	from ...util.jobqueue import enqueue
19
20	1	from .dbmodels import EmailConfig as DbEmailConfig
21	1	from .transfer.models import EmailConfig, Message, NameAndAddress
22
23
24	1	class UnknownEmailConfigId(ValueError):
25	1	pass
26
27
28	1	def create_config(
29		brand_id: BrandID,
30		sender_address: str,
31		*,
32		sender_name: Optional[str] = None,
33		contact_address: Optional[str] = None,
34		) -> EmailConfig:
35		"""Create a configuration."""
36	1	config = DbEmailConfig(
37		brand_id,
38		sender_address,
39		sender_name=sender_name,
40		contact_address=contact_address,
41		)
42
43	1	db.session.add(config)
44	1	db.session.commit()
45
46	1	return _db_entity_to_config(config)
47
48
49	1	def update_config(
50		brand_id: BrandID,
51		sender_address: str,
52		sender_name: Optional[str],
53		contact_address: Optional[str],
54		) -> EmailConfig:
55		"""Update a configuration."""
56		config = _find_db_config(brand_id)
57
58		if config is None:
59		raise UnknownEmailConfigId(
60		f'No e-mail config found for brand ID "{brand_id}"'
61		)
62
63		config.sender_address = sender_address
64		config.sender_name = sender_name
65		config.contact_address = contact_address
66
67		db.session.commit()
68
69		return _db_entity_to_config(config)
70
71
72	1	def delete_config(brand_id: BrandID) -> bool:
73		"""Delete a configuration.
74
75		It is expected that no database records (sites) refer to the
76		configuration anymore.
77
78		Return `True` on success, or `False` if an error occurred.
79		"""
80	1	get_config(brand_id) # Verify ID exists.
81
82	1	try:
83	1	db.session \
84		.query(DbEmailConfig) \
85		.filter_by(brand_id=brand_id) \
86		.delete()
87
88	1	db.session.commit()
89		except IntegrityError:
90		db.session.rollback()
91		return False
92
93	1	return True
94
95
96	1	def _find_db_config(brand_id: BrandID) -> Optional[DbEmailConfig]:
97	1	return db.session \
98		.query(DbEmailConfig) \
99		.filter_by(brand_id=brand_id) \
100		.one_or_none()
101
102
103	1	def get_config(brand_id: BrandID) -> EmailConfig:
104		"""Return the configuration, or raise an error if none is configured
105		for that brand.
106		"""
107	1	config = _find_db_config(brand_id)
108
109	1	if config is None:
110		raise UnknownEmailConfigId(
111		f'No e-mail config found for brand ID "{brand_id}"'
112		)
113
114	1	return _db_entity_to_config(config)
115
116
117	1	def set_config(
118		brand_id: BrandID,
119		sender_address: str,
120		*,
121		sender_name: Optional[str] = None,
122		contact_address: Optional[str] = None,
123		) -> None:
124		"""Add or update configuration for that ID."""
125	1	table = DbEmailConfig.__table__
126	1	identifier = {
127		'brand_id': brand_id,
128		'sender_address': sender_address,
129		}
130	1	replacement = {
131		'sender_name': sender_name,
132		'contact_address': contact_address,
133		}
134
135	1	upsert(table, identifier, replacement)
136
137
138	1	def get_all_configs() -> list[EmailConfig]:
139		"""Return all configurations."""
140		configs = db.session.query(DbEmailConfig).all()
141
142		return [_db_entity_to_config(config) for config in configs]
143
144
145	1	def parse_address(address_str: str) -> NameAndAddress:
146		"""Parse a string into name and address parts."""
147		name, address = parseaddr(address_str)
148
149		if not name and not address:
150		raise ValueError(f'Could not parse name and address value: "{address}"')
151
152		return NameAndAddress(name, address)
153
154
155	1	def enqueue_message(message: Message) -> None:
156		"""Enqueue e-mail to be sent asynchronously."""
157	1	enqueue_email(
158		message.sender, message.recipients, message.subject, message.body
159		)
160
161
162	1	def enqueue_email(
163		sender: NameAndAddress,
164		recipients: list[str],
165		subject: str,
166		body: str,
167		) -> None:
168		"""Enqueue e-mail to be sent asynchronously."""
169	1	sender_str = sender.format()
170	1	enqueue(send_email, sender_str, recipients, subject, body)
171
172
173	1	def send_email(
174		sender: str, recipients: list[str], subject: str, body: str
175		) -> None:
176		"""Send e-mail."""
177	1	email.send(sender, recipients, subject, body)
178
179
180	1	def _db_entity_to_config(config: DbEmailConfig) -> EmailConfig:
181	1	sender = NameAndAddress(
182		name=config.sender_name,
183		address=config.sender_address,
184		)
185
186	1	return EmailConfig(
187		brand_id=config.brand_id,
188		sender=sender,
189		contact_address=config.contact_address,
190		)
191

byceps / byceps

Push — main ( 1f94b4...a29a2e )

byceps.services.email.service.parse_address() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like