mirror of
https://github.com/Tech-Workers-Coalition-Italia/mobilizon-reshare.git
synced 2025-02-16 11:40:54 +01:00
refactor coordinators (#171)
* introduced CommandConfig object * added dry_run for start command * added test to start_dry_run * added dry_run recap * fixed import * improved printing * fixed zulip debug info * improved recap dry-run print * moved coordinator classes into dedicated package * removed unused init * divided event coordinators from recap coordinators * added some docstrings
This commit is contained in:
Simone Robutti
GitHub
parent
2d328a30bf
commit
63a30bb483
@ -6,18 +6,22 @@ from mobilizon_reshare.event.event import MobilizonEvent
|
||||
from mobilizon_reshare.event.event_selection_strategies import select_event_to_publish
|
||||
from mobilizon_reshare.publishers import get_active_publishers
|
||||
from mobilizon_reshare.publishers.abstract import EventPublication
|
||||
from mobilizon_reshare.publishers.coordinator import (
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.notify import (
|
||||
PublicationFailureNotifiersCoordinator,
|
||||
PublisherCoordinatorReport,
|
||||
)
|
||||
from mobilizon_reshare.publishers.coordinator import PublisherCoordinator
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
PublisherCoordinatorReport,
|
||||
PublisherCoordinator,
|
||||
)
|
||||
from mobilizon_reshare.storage.query.read import (
|
||||
get_published_events,
|
||||
build_publications,
|
||||
events_without_publications,
|
||||
)
|
||||
from mobilizon_reshare.storage.query.write import save_publication_report
|
||||
from mobilizon_reshare.publishers.coordinator import DryRunPublisherCoordinator
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.dry_run import (
|
||||
DryRunPublisherCoordinator,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@ -7,17 +7,21 @@ from mobilizon_reshare.config.command import CommandConfig
|
||||
from mobilizon_reshare.event.event import EventPublicationStatus, MobilizonEvent
|
||||
from mobilizon_reshare.publishers import get_active_publishers
|
||||
from mobilizon_reshare.publishers.abstract import RecapPublication
|
||||
from mobilizon_reshare.publishers.coordinator import (
|
||||
RecapCoordinator,
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.notify import (
|
||||
PublicationFailureNotifiersCoordinator,
|
||||
BaseCoordinatorReport,
|
||||
)
|
||||
from mobilizon_reshare.publishers.coordinators.recap_publishing.recap import (
|
||||
RecapCoordinator,
|
||||
)
|
||||
from mobilizon_reshare.publishers.coordinators import BaseCoordinatorReport
|
||||
from mobilizon_reshare.publishers.platforms.platform_mapping import (
|
||||
get_publisher_class,
|
||||
get_formatter_class,
|
||||
)
|
||||
from mobilizon_reshare.storage.query.read import events_with_status
|
||||
from mobilizon_reshare.publishers.coordinator import DryRunRecapCoordinator
|
||||
from mobilizon_reshare.publishers.coordinators.recap_publishing.dry_run import (
|
||||
DryRunRecapCoordinator,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@ -3,7 +3,7 @@ from typing import Optional
|
||||
from uuid import UUID
|
||||
|
||||
from mobilizon_reshare.main.publish import publish_publications
|
||||
from mobilizon_reshare.publishers.coordinator import (
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
PublisherCoordinatorReport,
|
||||
)
|
||||
from mobilizon_reshare.storage.query.exceptions import EventNotFound
|
||||
|
||||
@ -3,7 +3,9 @@ import logging.config
|
||||
from mobilizon_reshare.config.command import CommandConfig
|
||||
from mobilizon_reshare.main.publish import select_and_publish
|
||||
from mobilizon_reshare.main.pull import pull
|
||||
from mobilizon_reshare.publishers.coordinator import PublisherCoordinatorReport
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
PublisherCoordinatorReport,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@ -1,299 +0,0 @@
|
||||
import dataclasses
|
||||
import logging
|
||||
from abc import abstractmethod, ABC
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, Sequence
|
||||
|
||||
from mobilizon_reshare.models.publication import PublicationStatus
|
||||
from mobilizon_reshare.publishers import get_active_notifiers
|
||||
from mobilizon_reshare.publishers.abstract import (
|
||||
EventPublication,
|
||||
AbstractPlatform,
|
||||
RecapPublication,
|
||||
)
|
||||
from mobilizon_reshare.publishers.exceptions import PublisherError
|
||||
from mobilizon_reshare.publishers.platforms.platform_mapping import get_notifier_class
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class BasePublicationReport:
|
||||
status: PublicationStatus
|
||||
reason: Optional[str]
|
||||
|
||||
@property
|
||||
def succesful(self):
|
||||
return self.status == PublicationStatus.COMPLETED
|
||||
|
||||
def get_failure_message(self):
|
||||
return (
|
||||
f"Publication failed with status: {self.status}.\n" f"Reason: {self.reason}"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RecapPublicationReport(BasePublicationReport):
|
||||
publication: RecapPublication
|
||||
published_content: Optional[str] = dataclasses.field(default=None)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EventPublicationReport(BasePublicationReport):
|
||||
publication: EventPublication
|
||||
published_content: Optional[str] = dataclasses.field(default=None)
|
||||
|
||||
def get_failure_message(self):
|
||||
if not self.reason:
|
||||
logger.error("Report of failure without reason.", exc_info=True)
|
||||
|
||||
return (
|
||||
f"Publication {self.publication.id} failed with status: {self.status}.\n"
|
||||
f"Reason: {self.reason}\n"
|
||||
f"Publisher: {self.publication.publisher.name}\n"
|
||||
f"Event: {self.publication.event.name}"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class BaseCoordinatorReport:
|
||||
reports: Sequence[BasePublicationReport]
|
||||
|
||||
@property
|
||||
def successful(self):
|
||||
return all(r.status == PublicationStatus.COMPLETED for r in self.reports)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RecapCoordinatorReport(BaseCoordinatorReport):
|
||||
reports: Sequence[RecapPublicationReport]
|
||||
|
||||
def __str__(self):
|
||||
platform_messages = []
|
||||
for report in self.reports:
|
||||
intro = f"Message for: {report.publication.publisher.name}"
|
||||
platform_messages.append(
|
||||
f"""{intro}
|
||||
{"*"*len(intro)}
|
||||
{report.published_content}
|
||||
{"-"*80}"""
|
||||
)
|
||||
return "\n".join(platform_messages)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PublisherCoordinatorReport(BaseCoordinatorReport):
|
||||
reports: Sequence[EventPublicationReport]
|
||||
publications: Sequence[EventPublication] = dataclasses.field(default_factory=list)
|
||||
|
||||
def __str__(self):
|
||||
platform_messages = []
|
||||
for report in self.reports:
|
||||
intro = f"Message for: {report.publication.publisher.name}"
|
||||
platform_messages.append(
|
||||
f"""{intro}
|
||||
{"*"*len(intro)}
|
||||
{report.published_content}
|
||||
{"-"*80}"""
|
||||
)
|
||||
return "\n".join(platform_messages)
|
||||
|
||||
|
||||
class PublisherCoordinator:
|
||||
def __init__(self, publications: List[EventPublication]):
|
||||
self.publications = publications
|
||||
|
||||
def run(self) -> PublisherCoordinatorReport:
|
||||
errors = self._validate()
|
||||
if errors:
|
||||
return PublisherCoordinatorReport(
|
||||
reports=errors, publications=self.publications
|
||||
)
|
||||
|
||||
return self._post()
|
||||
|
||||
def _post(self) -> PublisherCoordinatorReport:
|
||||
reports = []
|
||||
|
||||
for publication in self.publications:
|
||||
|
||||
try:
|
||||
logger.info(f"Publishing to {publication.publisher.name}")
|
||||
message = publication.formatter.get_message_from_event(
|
||||
publication.event
|
||||
)
|
||||
publication.publisher.send(message, publication.event)
|
||||
reports.append(
|
||||
EventPublicationReport(
|
||||
status=PublicationStatus.COMPLETED,
|
||||
publication=publication,
|
||||
reason=None,
|
||||
published_content=message,
|
||||
)
|
||||
)
|
||||
except PublisherError as e:
|
||||
logger.error(str(e))
|
||||
reports.append(
|
||||
EventPublicationReport(
|
||||
status=PublicationStatus.FAILED,
|
||||
reason=str(e),
|
||||
publication=publication,
|
||||
)
|
||||
)
|
||||
|
||||
return PublisherCoordinatorReport(
|
||||
publications=self.publications, reports=reports
|
||||
)
|
||||
|
||||
def _safe_run(self, reasons, f, *args, **kwargs):
|
||||
try:
|
||||
f(*args, **kwargs)
|
||||
return reasons
|
||||
except Exception as e:
|
||||
return reasons + [str(e)]
|
||||
|
||||
def _validate(self):
|
||||
errors = []
|
||||
|
||||
for publication in self.publications:
|
||||
reasons = []
|
||||
reasons = self._safe_run(
|
||||
reasons, publication.publisher.validate_credentials,
|
||||
)
|
||||
reasons = self._safe_run(
|
||||
reasons, publication.formatter.validate_event, publication.event
|
||||
)
|
||||
|
||||
if len(reasons) > 0:
|
||||
errors.append(
|
||||
EventPublicationReport(
|
||||
status=PublicationStatus.FAILED,
|
||||
reason=", ".join(reasons),
|
||||
publication=publication,
|
||||
)
|
||||
)
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
class DryRunPublisherCoordinator(PublisherCoordinator):
|
||||
def __init__(self, publications: List[EventPublication]):
|
||||
self.publications = publications
|
||||
|
||||
def run(self) -> PublisherCoordinatorReport:
|
||||
errors = self._validate()
|
||||
if errors:
|
||||
coord_report = PublisherCoordinatorReport(
|
||||
reports=errors, publications=self.publications
|
||||
)
|
||||
else:
|
||||
reports = [
|
||||
EventPublicationReport(
|
||||
status=PublicationStatus.COMPLETED,
|
||||
publication=publication,
|
||||
reason=None,
|
||||
published_content=publication.formatter.get_message_from_event(
|
||||
publication.event
|
||||
),
|
||||
)
|
||||
for publication in self.publications
|
||||
]
|
||||
coord_report = PublisherCoordinatorReport(
|
||||
publications=self.publications, reports=reports
|
||||
)
|
||||
|
||||
return coord_report
|
||||
|
||||
|
||||
class Sender:
|
||||
def __init__(self, message: str, platforms: List[AbstractPlatform] = None):
|
||||
self.message = message
|
||||
self.platforms = platforms
|
||||
|
||||
def send_to_all(self):
|
||||
for platform in self.platforms:
|
||||
try:
|
||||
platform.send(self.message)
|
||||
except Exception as e:
|
||||
logger.critical(f"Failed to send message:\n{self.message}")
|
||||
logger.exception(e)
|
||||
|
||||
|
||||
class AbstractNotifiersCoordinator(ABC):
|
||||
def __init__(
|
||||
self, report: EventPublicationReport, notifiers: List[AbstractPlatform] = None
|
||||
):
|
||||
self.platforms = notifiers or [
|
||||
get_notifier_class(notifier)() for notifier in get_active_notifiers()
|
||||
]
|
||||
self.report = report
|
||||
|
||||
@abstractmethod
|
||||
def notify_failure(self):
|
||||
pass
|
||||
|
||||
|
||||
class PublicationFailureNotifiersCoordinator(AbstractNotifiersCoordinator):
|
||||
"""
|
||||
Sends a notification of a failure report to the active platforms
|
||||
"""
|
||||
|
||||
def notify_failure(self):
|
||||
logger.info("Sending failure notifications")
|
||||
if self.report.status == PublicationStatus.FAILED:
|
||||
Sender(self.report.get_failure_message(), self.platforms).send_to_all()
|
||||
|
||||
|
||||
class PublicationFailureLoggerCoordinator(PublicationFailureNotifiersCoordinator):
|
||||
"""
|
||||
Logs a report to console
|
||||
"""
|
||||
|
||||
def notify_failure(self):
|
||||
if self.report.status == PublicationStatus.FAILED:
|
||||
logger.error(self.report.get_failure_message())
|
||||
|
||||
|
||||
class RecapCoordinator:
|
||||
def __init__(self, recap_publications: List[RecapPublication]):
|
||||
self.recap_publications = recap_publications
|
||||
|
||||
def _build_recap_content(self, recap_publication: RecapPublication):
|
||||
fragments = [recap_publication.formatter.get_recap_header()]
|
||||
for event in recap_publication.events:
|
||||
fragments.append(recap_publication.formatter.get_recap_fragment(event))
|
||||
return "\n\n".join(fragments)
|
||||
|
||||
def _send(self, content, recap_publication):
|
||||
recap_publication.publisher.send(content)
|
||||
|
||||
def run(self) -> RecapCoordinatorReport:
|
||||
reports = []
|
||||
for recap_publication in self.recap_publications:
|
||||
try:
|
||||
|
||||
message = self._build_recap_content(recap_publication)
|
||||
self._send(message, recap_publication)
|
||||
reports.append(
|
||||
RecapPublicationReport(
|
||||
status=PublicationStatus.COMPLETED,
|
||||
reason=None,
|
||||
published_content=message,
|
||||
publication=recap_publication,
|
||||
)
|
||||
)
|
||||
except PublisherError as e:
|
||||
reports.append(
|
||||
RecapPublicationReport(
|
||||
status=PublicationStatus.FAILED,
|
||||
reason=str(e),
|
||||
publication=recap_publication,
|
||||
)
|
||||
)
|
||||
|
||||
return RecapCoordinatorReport(reports=reports)
|
||||
|
||||
|
||||
class DryRunRecapCoordinator(RecapCoordinator):
|
||||
def _send(self, content, recap_publication):
|
||||
pass
|
||||
32
mobilizon_reshare/publishers/coordinators/__init__.py
Normal file
32
mobilizon_reshare/publishers/coordinators/__init__.py
Normal file
@ -0,0 +1,32 @@
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Sequence
|
||||
|
||||
from mobilizon_reshare.models.publication import PublicationStatus
|
||||
|
||||
|
||||
@dataclass
|
||||
class BasePublicationReport:
|
||||
status: PublicationStatus
|
||||
reason: Optional[str]
|
||||
|
||||
@property
|
||||
def succesful(self):
|
||||
return self.status == PublicationStatus.COMPLETED
|
||||
|
||||
def get_failure_message(self):
|
||||
return (
|
||||
f"Publication failed with status: {self.status}.\n" f"Reason: {self.reason}"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class BaseCoordinatorReport:
|
||||
reports: Sequence[BasePublicationReport]
|
||||
|
||||
@property
|
||||
def successful(self):
|
||||
return all(r.status == PublicationStatus.COMPLETED for r in self.reports)
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -0,0 +1,63 @@
|
||||
import dataclasses
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional
|
||||
|
||||
from mobilizon_reshare.models.publication import PublicationStatus
|
||||
from mobilizon_reshare.publishers.abstract import EventPublication
|
||||
from mobilizon_reshare.publishers.coordinators import BasePublicationReport
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EventPublicationReport(BasePublicationReport):
|
||||
publication: EventPublication
|
||||
published_content: Optional[str] = dataclasses.field(default=None)
|
||||
|
||||
def get_failure_message(self):
|
||||
if not self.reason:
|
||||
logger.error("Report of failure without reason.", exc_info=True)
|
||||
|
||||
return (
|
||||
f"Publication {self.publication.id} failed with status: {self.status}.\n"
|
||||
f"Reason: {self.reason}\n"
|
||||
f"Publisher: {self.publication.publisher.name}\n"
|
||||
f"Event: {self.publication.event.name}"
|
||||
)
|
||||
|
||||
|
||||
class BaseEventPublishingCoordinator:
|
||||
def __init__(self, publications: List[EventPublication]):
|
||||
self.publications = publications
|
||||
|
||||
def _safe_run(self, reasons, f, *args, **kwargs):
|
||||
try:
|
||||
f(*args, **kwargs)
|
||||
return reasons
|
||||
except Exception as e:
|
||||
return reasons + [str(e)]
|
||||
|
||||
def _validate(self):
|
||||
errors = []
|
||||
|
||||
for publication in self.publications:
|
||||
reasons = []
|
||||
reasons = self._safe_run(
|
||||
reasons, publication.publisher.validate_credentials,
|
||||
)
|
||||
reasons = self._safe_run(
|
||||
reasons, publication.formatter.validate_event, publication.event
|
||||
)
|
||||
|
||||
if len(reasons) > 0:
|
||||
errors.append(
|
||||
EventPublicationReport(
|
||||
status=PublicationStatus.FAILED,
|
||||
reason=", ".join(reasons),
|
||||
publication=publication,
|
||||
)
|
||||
)
|
||||
|
||||
return errors
|
||||
@ -0,0 +1,38 @@
|
||||
from mobilizon_reshare.models.publication import PublicationStatus
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing import (
|
||||
BaseEventPublishingCoordinator,
|
||||
)
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
PublisherCoordinatorReport,
|
||||
EventPublicationReport,
|
||||
)
|
||||
|
||||
|
||||
class DryRunPublisherCoordinator(BaseEventPublishingCoordinator):
|
||||
"""
|
||||
Coordinator to perform a dry-run on the event publication
|
||||
"""
|
||||
|
||||
def run(self) -> PublisherCoordinatorReport:
|
||||
errors = self._validate()
|
||||
if errors:
|
||||
coord_report = PublisherCoordinatorReport(
|
||||
reports=errors, publications=self.publications
|
||||
)
|
||||
else:
|
||||
reports = [
|
||||
EventPublicationReport(
|
||||
status=PublicationStatus.COMPLETED,
|
||||
publication=publication,
|
||||
reason=None,
|
||||
published_content=publication.formatter.get_message_from_event(
|
||||
publication.event
|
||||
),
|
||||
)
|
||||
for publication in self.publications
|
||||
]
|
||||
coord_report = PublisherCoordinatorReport(
|
||||
publications=self.publications, reports=reports
|
||||
)
|
||||
|
||||
return coord_report
|
||||
@ -0,0 +1,60 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import List
|
||||
|
||||
from mobilizon_reshare.models.publication import PublicationStatus
|
||||
from mobilizon_reshare.publishers import get_active_notifiers
|
||||
from mobilizon_reshare.publishers.abstract import AbstractPlatform
|
||||
from mobilizon_reshare.publishers.coordinators import logger
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
EventPublicationReport,
|
||||
)
|
||||
from mobilizon_reshare.publishers.platforms.platform_mapping import get_notifier_class
|
||||
|
||||
|
||||
class Sender:
|
||||
def __init__(self, message: str, platforms: List[AbstractPlatform] = None):
|
||||
self.message = message
|
||||
self.platforms = platforms
|
||||
|
||||
def send_to_all(self):
|
||||
for platform in self.platforms:
|
||||
try:
|
||||
platform.send(self.message)
|
||||
except Exception as e:
|
||||
logger.critical(f"Failed to send message:\n{self.message}")
|
||||
logger.exception(e)
|
||||
|
||||
|
||||
class AbstractNotifiersCoordinator(ABC):
|
||||
def __init__(
|
||||
self, report: EventPublicationReport, notifiers: List[AbstractPlatform] = None
|
||||
):
|
||||
self.platforms = notifiers or [
|
||||
get_notifier_class(notifier)() for notifier in get_active_notifiers()
|
||||
]
|
||||
self.report = report
|
||||
|
||||
@abstractmethod
|
||||
def notify_failure(self):
|
||||
pass
|
||||
|
||||
|
||||
class PublicationFailureNotifiersCoordinator(AbstractNotifiersCoordinator):
|
||||
"""
|
||||
Sends a notification of a failure report to the active platforms
|
||||
"""
|
||||
|
||||
def notify_failure(self):
|
||||
logger.info("Sending failure notifications")
|
||||
if self.report.status == PublicationStatus.FAILED:
|
||||
Sender(self.report.get_failure_message(), self.platforms).send_to_all()
|
||||
|
||||
|
||||
class PublicationFailureLoggerCoordinator(PublicationFailureNotifiersCoordinator):
|
||||
"""
|
||||
Logs a report to console
|
||||
"""
|
||||
|
||||
def notify_failure(self):
|
||||
if self.report.status == PublicationStatus.FAILED:
|
||||
logger.error(self.report.get_failure_message())
|
||||
@ -0,0 +1,87 @@
|
||||
import dataclasses
|
||||
from dataclasses import dataclass
|
||||
from typing import Sequence
|
||||
import logging
|
||||
from mobilizon_reshare.models.publication import PublicationStatus
|
||||
from mobilizon_reshare.publishers.abstract import EventPublication
|
||||
from mobilizon_reshare.publishers.coordinators import BaseCoordinatorReport
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing import (
|
||||
BaseEventPublishingCoordinator,
|
||||
EventPublicationReport,
|
||||
)
|
||||
from mobilizon_reshare.publishers.exceptions import PublisherError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PublisherCoordinatorReport(BaseCoordinatorReport):
|
||||
reports: Sequence[EventPublicationReport]
|
||||
publications: Sequence[EventPublication] = dataclasses.field(default_factory=list)
|
||||
|
||||
def __str__(self):
|
||||
platform_messages = []
|
||||
for report in self.reports:
|
||||
intro = f"Message for: {report.publication.publisher.name}"
|
||||
platform_messages.append(
|
||||
f"""{intro}
|
||||
{"*" * len(intro)}
|
||||
{report.published_content}
|
||||
{"-" * 80}"""
|
||||
)
|
||||
return "\n".join(platform_messages)
|
||||
|
||||
|
||||
class PublisherCoordinator(BaseEventPublishingCoordinator):
|
||||
"""
|
||||
Coordinator to publish an event on every active platform
|
||||
"""
|
||||
|
||||
def run(self) -> PublisherCoordinatorReport:
|
||||
errors = self._validate()
|
||||
if errors:
|
||||
return PublisherCoordinatorReport(
|
||||
reports=errors, publications=self.publications
|
||||
)
|
||||
|
||||
return self._publish()
|
||||
|
||||
def _publish(self) -> PublisherCoordinatorReport:
|
||||
reports = []
|
||||
|
||||
for publication in self.publications:
|
||||
|
||||
try:
|
||||
publication_report = self._publish_publication(publication)
|
||||
reports.append(publication_report)
|
||||
except PublisherError as e:
|
||||
logger.error(str(e))
|
||||
reports.append(
|
||||
EventPublicationReport(
|
||||
status=PublicationStatus.FAILED,
|
||||
reason=str(e),
|
||||
publication=publication,
|
||||
)
|
||||
)
|
||||
|
||||
return PublisherCoordinatorReport(
|
||||
publications=self.publications, reports=reports
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _publish_publication(publication):
|
||||
"""
|
||||
Publishes a single publication
|
||||
:param publication:
|
||||
:return:
|
||||
"""
|
||||
|
||||
logger.info("Publishing to %s", publication.publisher.name)
|
||||
message = publication.formatter.get_message_from_event(publication.event)
|
||||
publication.publisher.send(message, publication.event)
|
||||
return EventPublicationReport(
|
||||
status=PublicationStatus.COMPLETED,
|
||||
publication=publication,
|
||||
reason=None,
|
||||
published_content=message,
|
||||
)
|
||||
@ -0,0 +1,20 @@
|
||||
from mobilizon_reshare.publishers.coordinators.recap_publishing.recap import (
|
||||
RecapCoordinator,
|
||||
)
|
||||
|
||||
|
||||
class DryRunRecapCoordinator(RecapCoordinator):
|
||||
"""
|
||||
Coordinator to perform a dry-run on the event recap
|
||||
"""
|
||||
|
||||
def _send(self, content, recap_publication):
|
||||
"""
|
||||
Overrides the Recap Coordinator _send on the assumption that _send is just a side effect publishing
|
||||
on a given platform. The report generated by RecapCoordinator should be sufficient to perform
|
||||
the dry-run print at CLI level.
|
||||
:param content:
|
||||
:param recap_publication:
|
||||
:return:
|
||||
"""
|
||||
pass
|
||||
@ -0,0 +1,78 @@
|
||||
import dataclasses
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Sequence, List
|
||||
|
||||
from mobilizon_reshare.models.publication import PublicationStatus
|
||||
from mobilizon_reshare.publishers.abstract import RecapPublication
|
||||
from mobilizon_reshare.publishers.coordinators import (
|
||||
BasePublicationReport,
|
||||
BaseCoordinatorReport,
|
||||
)
|
||||
from mobilizon_reshare.publishers.exceptions import PublisherError
|
||||
|
||||
|
||||
@dataclass
|
||||
class RecapPublicationReport(BasePublicationReport):
|
||||
publication: RecapPublication
|
||||
published_content: Optional[str] = dataclasses.field(default=None)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RecapCoordinatorReport(BaseCoordinatorReport):
|
||||
reports: Sequence[RecapPublicationReport]
|
||||
|
||||
def __str__(self):
|
||||
platform_messages = []
|
||||
for report in self.reports:
|
||||
intro = f"Message for: {report.publication.publisher.name}"
|
||||
platform_messages.append(
|
||||
f"""{intro}
|
||||
{"*"*len(intro)}
|
||||
{report.published_content}
|
||||
{"-"*80}"""
|
||||
)
|
||||
return "\n".join(platform_messages)
|
||||
|
||||
|
||||
class RecapCoordinator:
|
||||
"""
|
||||
Coordinator to publish a recap on future events
|
||||
"""
|
||||
|
||||
def __init__(self, recap_publications: List[RecapPublication]):
|
||||
self.recap_publications = recap_publications
|
||||
|
||||
def _build_recap_content(self, recap_publication: RecapPublication):
|
||||
fragments = [recap_publication.formatter.get_recap_header()]
|
||||
for event in recap_publication.events:
|
||||
fragments.append(recap_publication.formatter.get_recap_fragment(event))
|
||||
return "\n\n".join(fragments)
|
||||
|
||||
def _send(self, content, recap_publication):
|
||||
recap_publication.publisher.send(content)
|
||||
|
||||
def run(self) -> RecapCoordinatorReport:
|
||||
reports = []
|
||||
for recap_publication in self.recap_publications:
|
||||
try:
|
||||
|
||||
message = self._build_recap_content(recap_publication)
|
||||
self._send(message, recap_publication)
|
||||
reports.append(
|
||||
RecapPublicationReport(
|
||||
status=PublicationStatus.COMPLETED,
|
||||
reason=None,
|
||||
published_content=message,
|
||||
publication=recap_publication,
|
||||
)
|
||||
)
|
||||
except PublisherError as e:
|
||||
reports.append(
|
||||
RecapPublicationReport(
|
||||
status=PublicationStatus.FAILED,
|
||||
reason=str(e),
|
||||
publication=recap_publication,
|
||||
)
|
||||
)
|
||||
|
||||
return RecapCoordinatorReport(reports=reports)
|
||||
@ -63,8 +63,7 @@ async def events_with_status(
|
||||
|
||||
|
||||
async def get_all_publications(
|
||||
from_date: Optional[Arrow] = None,
|
||||
to_date: Optional[Arrow] = None,
|
||||
from_date: Optional[Arrow] = None, to_date: Optional[Arrow] = None,
|
||||
) -> Iterable[Publication]:
|
||||
return await prefetch_publication_relations(
|
||||
_add_date_window(Publication.all(), "timestamp", from_date, to_date)
|
||||
@ -72,8 +71,7 @@ async def get_all_publications(
|
||||
|
||||
|
||||
async def get_all_events(
|
||||
from_date: Optional[Arrow] = None,
|
||||
to_date: Optional[Arrow] = None,
|
||||
from_date: Optional[Arrow] = None, to_date: Optional[Arrow] = None,
|
||||
) -> list[MobilizonEvent]:
|
||||
return [
|
||||
event_from_model(event)
|
||||
@ -133,8 +131,7 @@ async def publications_with_status(
|
||||
|
||||
|
||||
async def events_without_publications(
|
||||
from_date: Optional[Arrow] = None,
|
||||
to_date: Optional[Arrow] = None,
|
||||
from_date: Optional[Arrow] = None, to_date: Optional[Arrow] = None,
|
||||
) -> list[MobilizonEvent]:
|
||||
query = Event.filter(publications__id=None)
|
||||
events = await prefetch_event_relations(
|
||||
|
||||
@ -8,7 +8,9 @@ from mobilizon_reshare.event.event import MobilizonEvent
|
||||
from mobilizon_reshare.models.event import Event
|
||||
from mobilizon_reshare.models.publication import Publication
|
||||
from mobilizon_reshare.models.publisher import Publisher
|
||||
from mobilizon_reshare.publishers.coordinator import PublisherCoordinatorReport
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
PublisherCoordinatorReport,
|
||||
)
|
||||
from mobilizon_reshare.storage.query.converter import event_to_model
|
||||
from mobilizon_reshare.storage.query.read import (
|
||||
events_without_publications,
|
||||
@ -91,9 +93,7 @@ async def create_unpublished_events(
|
||||
|
||||
|
||||
@atomic()
|
||||
async def update_publishers(
|
||||
names: Iterable[str],
|
||||
) -> None:
|
||||
async def update_publishers(names: Iterable[str],) -> None:
|
||||
names = set(names)
|
||||
known_publisher_names = set(p.name for p in await Publisher.all())
|
||||
for name in names.difference(known_publisher_names):
|
||||
|
||||
@ -8,7 +8,7 @@ import mobilizon_reshare.publishers
|
||||
import mobilizon_reshare.storage.query.read
|
||||
from mobilizon_reshare.models.publisher import Publisher
|
||||
import mobilizon_reshare.main.recap
|
||||
from mobilizon_reshare.publishers import coordinator
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing import notify
|
||||
from tests import today
|
||||
from tests.conftest import event_1, event_0
|
||||
|
||||
@ -118,14 +118,10 @@ async def mock_publisher_config(monkeypatch, publisher_class, mock_formatter_cla
|
||||
mobilizon_reshare.main.recap, "get_active_publishers", _mock_active_pub
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
mobilizon_reshare.main.recap,
|
||||
"get_publisher_class",
|
||||
_mock_pub_class,
|
||||
mobilizon_reshare.main.recap, "get_publisher_class", _mock_pub_class,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
mobilizon_reshare.main.recap,
|
||||
"get_formatter_class",
|
||||
_mock_format_class,
|
||||
mobilizon_reshare.main.recap, "get_formatter_class", _mock_format_class,
|
||||
)
|
||||
return p
|
||||
|
||||
@ -142,9 +138,7 @@ async def mock_notifier_config(monkeypatch, publisher_class, mock_formatter_clas
|
||||
return mock_formatter_class
|
||||
|
||||
monkeypatch.setattr(
|
||||
coordinator,
|
||||
"get_notifier_class",
|
||||
_mock_notifier_class,
|
||||
notify, "get_notifier_class", _mock_notifier_class,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
mobilizon_reshare.publishers.platforms.platform_mapping,
|
||||
@ -152,7 +146,7 @@ async def mock_notifier_config(monkeypatch, publisher_class, mock_formatter_clas
|
||||
_mock_format_class,
|
||||
)
|
||||
|
||||
monkeypatch.setattr(coordinator, "get_active_notifiers", _mock_active_notifier)
|
||||
monkeypatch.setattr(notify, "get_active_notifiers", _mock_active_notifier)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@ -155,7 +155,7 @@ def initialize_db_tests(request) -> None:
|
||||
await Tortoise.init(config)
|
||||
try:
|
||||
await Tortoise._drop_databases()
|
||||
except: # pragma: nocoverage
|
||||
except: # noqa
|
||||
pass
|
||||
|
||||
await Tortoise.init(config, _create_db=True)
|
||||
@ -197,9 +197,7 @@ def event_model_generator():
|
||||
|
||||
@pytest.fixture()
|
||||
def publisher_model_generator():
|
||||
def _publisher_model_generator(
|
||||
idx=1,
|
||||
):
|
||||
def _publisher_model_generator(idx=1,):
|
||||
return Publisher(name=f"publisher_{idx}", account_ref=f"account_ref_{idx}")
|
||||
|
||||
return _publisher_model_generator
|
||||
@ -436,10 +434,7 @@ def mock_mobilizon_success_answer(mobilizon_answer, mobilizon_url):
|
||||
with responses.RequestsMock() as rsps:
|
||||
|
||||
rsps.add(
|
||||
responses.POST,
|
||||
mobilizon_url,
|
||||
json=mobilizon_answer,
|
||||
status=200,
|
||||
responses.POST, mobilizon_url, json=mobilizon_answer, status=200,
|
||||
)
|
||||
yield
|
||||
|
||||
@ -451,10 +446,7 @@ def mock_multiple_success_answer(multiple_answers, mobilizon_url):
|
||||
|
||||
for answer in multiple_answers:
|
||||
rsps.add(
|
||||
responses.POST,
|
||||
mobilizon_url,
|
||||
json=answer,
|
||||
status=200,
|
||||
responses.POST, mobilizon_url, json=answer, status=200,
|
||||
)
|
||||
|
||||
yield
|
||||
|
||||
@ -12,11 +12,15 @@ from mobilizon_reshare.models.publication import (
|
||||
)
|
||||
from mobilizon_reshare.models.publisher import Publisher
|
||||
from mobilizon_reshare.publishers.abstract import EventPublication, RecapPublication
|
||||
from mobilizon_reshare.publishers.coordinator import (
|
||||
PublisherCoordinatorReport,
|
||||
EventPublicationReport,
|
||||
PublisherCoordinator,
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.notify import (
|
||||
PublicationFailureNotifiersCoordinator,
|
||||
)
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
EventPublicationReport,
|
||||
PublisherCoordinatorReport,
|
||||
PublisherCoordinator,
|
||||
)
|
||||
from mobilizon_reshare.publishers.coordinators.recap_publishing.recap import (
|
||||
RecapCoordinator,
|
||||
)
|
||||
from mobilizon_reshare.storage.query.converter import (
|
||||
|
||||
@ -4,7 +4,9 @@ import responses
|
||||
|
||||
from mobilizon_reshare.config.config import get_settings
|
||||
from mobilizon_reshare.models.publication import PublicationStatus
|
||||
from mobilizon_reshare.publishers.coordinator import PublisherCoordinator
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
PublisherCoordinator,
|
||||
)
|
||||
from mobilizon_reshare.publishers.exceptions import (
|
||||
InvalidEvent,
|
||||
InvalidResponse,
|
||||
@ -88,9 +90,7 @@ def mocked_client_error_response():
|
||||
|
||||
@pytest.fixture
|
||||
@pytest.mark.asyncio
|
||||
async def setup_db(
|
||||
generate_models
|
||||
):
|
||||
async def setup_db(generate_models):
|
||||
settings = get_settings()
|
||||
settings["publisher"]["zulip"][
|
||||
"bot_email"
|
||||
|
||||
@ -5,9 +5,9 @@ import pytest
|
||||
from mobilizon_reshare.models.publication import PublicationStatus, Publication
|
||||
from mobilizon_reshare.models.publisher import Publisher
|
||||
from mobilizon_reshare.publishers.abstract import EventPublication
|
||||
from mobilizon_reshare.publishers.coordinator import (
|
||||
PublisherCoordinatorReport,
|
||||
from mobilizon_reshare.publishers.coordinators.event_publishing.publish import (
|
||||
EventPublicationReport,
|
||||
PublisherCoordinatorReport,
|
||||
)
|
||||
from mobilizon_reshare.publishers.platforms.telegram import (
|
||||
TelegramFormatter,
|
||||
@ -18,9 +18,7 @@ from mobilizon_reshare.storage.query.write import (
|
||||
update_publishers,
|
||||
create_unpublished_events,
|
||||
)
|
||||
from tests.storage import (
|
||||
complete_specification,
|
||||
)
|
||||
from tests.storage import complete_specification
|
||||
from tests.conftest import event_6, event_0, event_1, event_2, event_3, event_3_updated
|
||||
|
||||
two_publishers_specification = {"publisher": ["telegram", "twitter"]}
|
||||
@ -68,10 +66,7 @@ two_events_specification = {
|
||||
],
|
||||
)
|
||||
async def test_update_publishers(
|
||||
specification,
|
||||
names,
|
||||
expected_result,
|
||||
generate_models,
|
||||
specification, names, expected_result, generate_models,
|
||||
):
|
||||
await generate_models(specification)
|
||||
await update_publishers(names)
|
||||
@ -127,10 +122,7 @@ async def test_update_publishers(
|
||||
],
|
||||
)
|
||||
async def test_create_unpublished_events(
|
||||
specification,
|
||||
events_from_mobilizon,
|
||||
expected_result,
|
||||
generate_models,
|
||||
specification, events_from_mobilizon, expected_result, generate_models,
|
||||
):
|
||||
await generate_models(specification)
|
||||
|
||||
@ -171,11 +163,7 @@ async def test_create_unpublished_events(
|
||||
],
|
||||
)
|
||||
async def test_save_publication_report(
|
||||
specification,
|
||||
report,
|
||||
event,
|
||||
expected_result,
|
||||
generate_models,
|
||||
specification, report, event, expected_result, generate_models,
|
||||
):
|
||||
await generate_models(specification)
|
||||
known_publication_ids = set(p.id for p in await Publication.all())
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user