Source code for election_day.collections.subscribers

from email.headerregistry import Address
from onegov.core.collection import Pagination
from onegov.core.templates import render_template
from onegov.election_day import _
from onegov.election_day.formats.imports.common import load_csv
from onegov.election_day.models import EmailSubscriber
from onegov.election_day.models import SmsSubscriber
from onegov.election_day.models import Subscriber
from sqlalchemy import func


from typing import Any
from typing import IO
from typing import TypeVar
from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from onegov.election_day.formats.imports.common import FileImportError
    from onegov.election_day.request import ElectionDayRequest
    from sqlalchemy.orm import Query
    from sqlalchemy.orm import Session
    from typing_extensions import Self
    from uuid import UUID


[docs] _S = TypeVar('_S', bound=Subscriber)
[docs] class SubscriberCollection(Pagination[_S]):
[docs] page: int
def __init__( self: 'SubscriberCollection[Subscriber]', session: 'Session', page: int = 0, term: str | None = None, active_only: bool | None = True ): super().__init__(page) self.session = session self.term = term self.active_only = active_only @property
[docs] def model_class(self) -> type[_S]: return Subscriber # type:ignore[return-value]
[docs] def __eq__(self, other: object) -> bool: return ( isinstance(other, self.__class__) and self.page == other.page and self.term == other.term and self.active_only == other.active_only )
[docs] def subset(self) -> 'Query[_S]': return self.query()
@property
[docs] def page_index(self) -> int: return self.page
[docs] def page_by_index(self, index: int) -> 'Self': return self.__class__(self.session, index)
[docs] def for_active_only(self, active_only: bool) -> 'Self': return self.__class__(self.session, 0, self.term, active_only)
[docs] def add( self, address: str, domain: str | None, domain_segment: str | None, locale: str, active: bool ) -> _S: subscriber = self.model_class( address=address, domain=domain, domain_segment=domain_segment, locale=locale, active=active ) self.session.add(subscriber) self.session.flush() return subscriber
[docs] def query(self, active_only: bool | None = None) -> 'Query[_S]': query = self.session.query(self.model_class) active_only = self.active_only if active_only is None else active_only if active_only: query = query.filter(self.model_class.active.is_(True)) if self.term: query = query.filter(self.model_class.address.contains(self.term)) self.batch_size = query.count() query = query.order_by(self.model_class.address) return query
[docs] def by_id(self, id: 'UUID') -> _S | None: """ Returns the subscriber by its id. """ query = self.query(active_only=False) query = query.filter(self.model_class.id == id) return query.first()
[docs] def by_address( self, address: str, domain: str | None, domain_segment: str | None, ) -> _S | None: """ Returns the (first) subscriber by its address. """ query = self.query(active_only=False) query = query.filter( self.model_class.address == address, self.model_class.domain == domain, self.model_class.domain_segment == domain_segment, ) return query.first()
[docs] def initiate_subscription( self, address: str, domain: str | None, domain_segment: str | None, request: 'ElectionDayRequest' ) -> _S: """ Initiate the subscription process. Might be used to change the locale by re-subscribing. """ subscriber = self.by_address(address, domain, domain_segment) if not subscriber: locale = request.locale assert locale is not None subscriber = self.add( address, domain, domain_segment, locale, False ) self.handle_subscription(subscriber, domain, domain_segment, request) return subscriber
[docs] def handle_subscription( self, subscriber: _S, domain: str | None, domain_segment: str | None, request: 'ElectionDayRequest' ) -> None: """ Send the subscriber a request to confirm the subscription. """ raise NotImplementedError()
[docs] def initiate_unsubscription( self, address: str, domain: str | None, domain_segment: str | None, request: 'ElectionDayRequest' ) -> None: """ Initiate the unsubscription process. """ subscriber = self.by_address(address, domain, domain_segment) if subscriber: self.handle_unsubscription(subscriber, request)
[docs] def handle_unsubscription( self, subscriber: _S, request: 'ElectionDayRequest' ) -> None: """ Send the subscriber a request to confirm the unsubscription. """ raise NotImplementedError()
[docs] def export(self) -> list[dict[str, Any]]: """ Returns all data connected to these subscribers. """ return [ { 'address': subscriber.address, 'domain': subscriber.domain, 'domain_segment': subscriber.domain_segment, 'locale': subscriber.locale, 'active': subscriber.active } for subscriber in self.query() ]
[docs] def cleanup( self, file: IO[bytes], mimetype: str, delete: bool ) -> tuple[list['FileImportError'], int]: """ Disables or deletes the subscribers in the given CSV. Ignores domain and domain segment, as this is inteded to cleanup bounced addresses. """ csv, error = load_csv(file, mimetype, expected_headers=['address']) if error: return [error], 0 assert csv is not None addresses = [l.address.lower() for l in csv.lines if l.address] query = self.session.query(self.model_class) query = query.filter( func.lower(self.model_class.address).in_(addresses) ) if delete: count = query.delete() else: query = query.filter(self.model_class.active.is_(True)) count = query.count() for subscriber in query: subscriber.active = False return [], count
[docs] class EmailSubscriberCollection(SubscriberCollection[EmailSubscriber]): @property
[docs] def model_class(self) -> type[EmailSubscriber]: return EmailSubscriber
[docs] def handle_subscription( self, subscriber: EmailSubscriber, domain: str | None, domain_segment: str | None, request: 'ElectionDayRequest' ) -> None: """ Send the (new) subscriber a request to confirm the subscription. """ from onegov.election_day.layouts import MailLayout # circular token = request.new_url_safe_token({ 'address': subscriber.address, 'domain': domain, 'domain_segment': domain_segment, 'locale': request.locale }) optin = request.link(request.app.principal, 'optin-email') optin = f'{optin}?opaque={token}' optout = request.link(request.app.principal, 'optout-email') optout = f'{optout}?opaque={token}' # even though this is technically a transactional e-mail we send # it as marketing, since the actual subscription is sent as # a marketing e-mail as well title = request.translate(_("Please confirm your email")) request.app.send_marketing_email( subject=title, receivers=(subscriber.address, ), reply_to=Address( display_name=request.app.principal.name or '', addr_spec=request.app.principal.reply_to or request.app.mail['marketing']['sender'] # type:ignore ), content=render_template( 'mail_confirm_subscription.pt', request, { 'title': title, 'model': None, 'optin': optin, 'optout': optout, 'layout': MailLayout(self, request), } ), headers={ 'List-Unsubscribe': f'<{optout}>', 'List-Unsubscribe-Post': 'List-Unsubscribe=One-Click' } )
[docs] def confirm_subscription( self, address: str, domain: str | None, domain_segment: str | None, locale: str, ) -> bool: """ Confirm the subscription. """ subscriber = self.by_address(address, domain, domain_segment) if subscriber: subscriber.active = True subscriber.locale = locale return True return False
[docs] def handle_unsubscription( self, subscriber: EmailSubscriber, request: 'ElectionDayRequest' ) -> None: """ Send the subscriber a request to confirm the unsubscription. """ from onegov.election_day.layouts import MailLayout # circular token = request.new_url_safe_token({ 'address': subscriber.address, 'domain': subscriber.domain, 'domain_segment': subscriber.domain_segment }) optout = request.link(request.app.principal, 'optout-email') optout = f'{optout}?opaque={token}' # even though this is technically a transactional e-mail we send # it as marketing, since the actual subscription is sent as # a marketing e-mail as well title = request.translate(_("Please confirm your unsubscription")) request.app.send_marketing_email( subject=title, receivers=(subscriber.address, ), reply_to=Address( display_name=request.app.principal.name or '', addr_spec=request.app.principal.reply_to or request.app.mail['marketing']['sender'] # type:ignore ), content=render_template( 'mail_confirm_unsubscription.pt', request, { 'title': title, 'model': None, 'optout': optout, 'layout': MailLayout(self, request), } ), headers={ 'List-Unsubscribe': f'<{optout}>', 'List-Unsubscribe-Post': 'List-Unsubscribe=One-Click' } )
[docs] def confirm_unsubscription( self, address: str, domain: str | None, domain_segment: str | None, ) -> bool: """ Confirm the unsubscription. """ subscriber = self.by_address(address, domain, domain_segment) if subscriber: subscriber.active = False return True return False
[docs] class SmsSubscriberCollection(SubscriberCollection[SmsSubscriber]): @property
[docs] def model_class(self) -> type[SmsSubscriber]: return SmsSubscriber
[docs] def handle_subscription( self, subscriber: SmsSubscriber, domain: str | None, domain_segment: str | None, request: 'ElectionDayRequest' ) -> None: """ Confirm the subscription by sending an SMS (if not already subscribed). There is no double-opt-in for SMS subscribers. """ if not subscriber.active or subscriber.locale != request.locale: assert request.locale is not None subscriber.locale = request.locale subscriber.active = True content = request.translate(_( "Successfully subscribed to the SMS service. You will" " receive a SMS every time new results are published." )) request.app.send_sms(subscriber.address, content)
[docs] def handle_unsubscription( self, subscriber: SmsSubscriber, request: 'ElectionDayRequest' ) -> None: """ Deactivate the subscriber. There is no double-opt-out for SMS subscribers. """ subscriber.active = False