Steal RSS abstractions from one of my other projects

config.example.json

@@ -4,4 +4,8 @@
   "logging": {
     "level": "info"
   }
+  "email": {
+    "api_key": "asdf1234",
+    "sender": "me@example.com"
+  }
 }

src/config/__init__.py

@@ -1,9 +1,9 @@
 import json
-from logging import INFO
 from dataclasses import dataclass
 from typing import Any
 
 from .logging import Logging
+from .email import Email
 from .parse import ConfigParseError, assert_key_of_type
 
 
@@ -12,21 +12,25 @@ class Config:
     host: str | None
     port: int | None
     logging: Logging
+    email: Email
 
     @classmethod
     def from_dict(cls, config: dict[str, Any]) -> 'Config':
         assert_key_of_type(config, 'logging', dict)
+        assert_key_of_type(config, 'email', dict)
         assert_key_of_type(config, 'host', str)
         assert_key_of_type(config, 'port', int)
         try:
             log_config = Logging.from_dict(config['logging'])
-            return Config(host=config['host'], port=config['port'], logging=log_config)
-
         except ConfigParseError as e:
             raise ConfigParseError(['logging', *e.keypath], e.issue) from e
 
+        try:
+            email_config = Email.from_dict(config['email'])
+        except ConfigParseError as e:
+            raise ConfigParseError(['email', *e.keypath], e.issue) from e
 
-DEFAULT_CONFIG = Config(logging=Logging(level=INFO), port=None, host=None)
+        return Config(host=config['host'], port=config['port'], email=email_config, logging=log_config)
 
 
 def parse_config(filename: str) -> Config:
@@ -35,4 +39,4 @@ def parse_config(filename: str) -> Config:
     return Config.from_dict(config)
 
 
-__all__ = ['Config', 'parse_config', 'DEFAULT_CONFIG']
+__all__ = ['Config', 'parse_config']

src/config/email.py

@@ -0,0 +1,18 @@
+from src.config.parse import assert_key_of_type
+from typing import Any
+from dataclasses import dataclass
+
+
+EmailAddress = str
+
+
+@dataclass
+class Email:
+    api_key: str
+    sender: EmailAddress
+
+    @classmethod
+    def from_dict(cls, config: dict[str, Any]) -> 'Email':
+        assert_key_of_type(config, 'api_key', str)
+        assert_key_of_type(config, 'sender', str)
+        return Email(api_key=config['api_key'], sender=config['sender'])

src/main.py

@@ -5,7 +5,7 @@ from flask import Flask
 import argparse
 import logging
 
-from src.config import parse_config, DEFAULT_CONFIG, Config
+from src.config import parse_config, Config
 
 logger = logging.getLogger(__name__)
 
@@ -23,9 +23,10 @@ def create_app(name: str, config: Config) -> Flask:
 
 def main():
     parser = argparse.ArgumentParser('Cereal', description='run the Cereal server')
-    parser.add_argument('--config', '-c', help='The path to the configuration file, expects JSON')
+    parser.add_argument('--config', '-c', help='The path to the configuration file, expects JSON', required=True)
     args = parser.parse_args()
-    config = parse_config(args.config) if args.config else DEFAULT_CONFIG
+
+    config = parse_config(args.config)
     app = create_app('Cereal', config)
     logger.info(str(config))
     serve(app, host='0.0.0.0' if config.host is None else config.host, port=config.port)

src/notifications/data.py

@@ -0,0 +1,38 @@
+from dataclasses import dataclass, field
+
+URL = str
+
+
+@dataclass
+class NotificationAction:
+    # A unique identifier for the action
+    action: str
+    # The title to show with the action
+    title: str
+    # Where to navigate to when clicked.
+    navigate: URL | None = None
+
+
+@dataclass
+class Notification:
+    """
+    Details to show in a notification
+
+    Much of the data here is intended to match the arguments to the [showNotification](https://developer.mozilla.org/en-US/docs/Web/API/ServiceWorkerRegistration/showNotification)
+    Web API, but this content is ideally displayable in other contexts (for example in Email or RSS)
+    """
+
+    # The main title content to show
+    title: str
+    # The content of the notification
+    content: str
+
+    # The UNIX timestamp in milliseconds since the epoch associated with this notification
+    timestamp: int
+    # A list of actions included with this notification
+    actions: list[NotificationAction] = field(default_factory=list)
+    # The location of an icon to show as the icon for this notification
+    icon: URL | None = None
+
+    # The location of an image associated with this notification
+    image: URL | None = None

src/notifications/email.py

@@ -0,0 +1,53 @@
+from typing import Any
+from email.utils import formatdate, make_msgid
+import abc
+from email.message import EmailMessage
+
+from ..config.email import Email as EmailConfig
+from .service import NotificationService
+from .data import Notification
+
+
+class EmailService(NotificationService):
+    def __init__(self, config: EmailConfig):
+        self._config = config
+
+    def notify(self, user: Any, notification: Notification):
+        email = self._prepare_email(user.email, notification)
+        self.send_email(email)
+
+    @abc.abstractmethod
+    def send_email(self, email: EmailMessage):
+        pass
+
+    def _render_template_text(self, notification: Notification) -> str:
+        # TODO
+        return ''
+
+    def _render_template_html(self, notification: Notification) -> str:
+        # TODO
+        return ''
+
+    def _prepare_email(self, receiver: str, notification: Notification) -> EmailMessage:
+        msg = EmailMessage()
+        msg['From'] = self._config.sender
+        msg['To'] = receiver
+        msg['Subject'] = notification.title
+
+        msg['Date'] = formatdate(localtime=True)
+        msg['Message-ID'] = make_msgid()
+
+        msg.set_content(self._render_template_text(notification))
+        msg.add_alternative(self._render_template_html(notification), subtype='html')
+
+        return msg
+
+
+class BirdEmailServiceImpl(EmailService):
+    def send_email(self, email: EmailMessage):
+        # TODO: Implement Bird Specific Email sending and Error handling
+        pass
+
+
+def get_service(config: EmailConfig) -> EmailService:
+    return BirdEmailServiceImpl(config)

src/notifications/rss.py

@@ -0,0 +1,205 @@
+from xml.etree.ElementTree import TreeBuilder, tostring as to_xml_string
+from typing import Any
+from email import utils
+from datetime import datetime
+from time import mktime
+from dataclasses import dataclass, field
+
+Email = str
+Language = str
+URL = str
+
+
+@dataclass
+class Image:
+    """https://www.rssboard.org/rss-specification#ltimagegtSubelementOfLtchannelgt"""
+
+    url: URL  # The URL of the image
+    title: str  # The channel title
+    link: URL  # The channel link
+    width: int | None = None  # in pixels max 144, default: 88.
+    height: int | None = None  # in pixels max 400, default: 31.
+    description: str | None = None
+
+
+@dataclass
+class Enclosure:
+    """https://www.rssboard.org/rss-specification#ltenclosuregtSubelementOfLtitemgt"""
+
+    url: URL
+    type: str  # MIME Type
+    length: int | None = None  # no. of bytes
+
+
+@dataclass
+class Item:
+    # One of Title or Description must be set
+    title: str | None = None
+    description: str | None = None
+    link: URL | None = None
+    author: Email | None = None
+    categories: list[str] = field(default_factory=list)
+    comments: URL | None = None
+    enclosures: list[Enclosure] = field(default_factory=list)
+    guid: str | None = None
+    pubDate: datetime | None = None
+
+
+@dataclass
+class Channel:
+    """https://www.rssboard.org/rss-specification"""
+
+    title: str
+    link: str
+    description: str
+    categories: list[str] = field(default_factory=list)
+    items: list[Item] = field(default_factory=list)
+    language: Language | None = None
+    copyright: str | None = None
+    managingEditor: Email | None = None
+    webMaster: Email | None = None
+    pubDate: datetime | None = None
+    lastBuildDate: datetime | None = None
+    generator: str | None = None  # A string indicating the program used to generate the channel.
+    docs: URL | None = None
+    ttl: int | None = None  # Time to cache in minutes
+    image: Image | None = None
+
+
+def convert_date(item: Any, key: str) -> datetime | None:
+    if key not in item:
+        return None
+
+    return datetime.fromtimestamp(mktime(item[key]))
+
+
+def maybe_convert_enclosure(enclosure) -> Enclosure | None:
+    href = enclosure.get('href')
+    mime = enclosure.get('type')
+    length = enclosure.get('length')
+    if any([x is None for x in [href, mime]]):
+        return None
+
+    kwargs = {
+        'url': href,
+        'type': mime,
+    }
+
+    if length is not None:
+        kwargs['length'] = int(length)
+
+    return Enclosure(**kwargs)
+
+
+def convert_item(item) -> Item:
+    tags = item['tags'] if 'tags' in item else []
+    enclosures: list[Any] = list(filter(lambda x: x is not None, [maybe_convert_enclosure(e) for e in item.enclosures]))
+
+    return Item(
+        title=item.get('title'),
+        description=item.get('summary'),
+        link=item.get('link'),
+        author=item.get('author'),
+        categories=[tag['term'] for tag in tags],
+        comments=item.get('comments'),
+        guid=item.get('id'),
+        pubDate=convert_date(item, 'published_parsed'),
+        enclosures=enclosures,
+    )
+
+
+def encode_inline(builder: TreeBuilder, tag: str, data: str | None, **kwargs):
+    if data is None:
+        return
+    builder.start(tag, kwargs)
+    builder.data(data)
+    builder.end(tag)
+
+
+def encode_date(date: datetime | None) -> str | None:
+    if date is None:
+        return
+
+    return utils.format_datetime(date)
+
+
+def encode_enclosure(builder: TreeBuilder, enclosure: Enclosure):
+    kwargs = {
+        'url': enclosure.url,
+        'type': enclosure.type,
+    }
+    if enclosure.length is not None:
+        kwargs['length'] = str(enclosure.length)
+
+    encode_inline(builder, 'enclosure', '', **kwargs)
+
+
+def encode_image(builder: TreeBuilder, image: Image):
+    builder.start('image', {})
+    encode_inline(builder, 'url', image.url)
+    encode_inline(builder, 'title', image.title)
+    encode_inline(builder, 'link', image.link)
+    encode_inline(builder, 'width', str(image.width) if image.width is not None else None)
+    encode_inline(builder, 'height', str(image.height) if image.height is not None else None)
+    encode_inline(builder, 'description', image.description)
+    builder.end('image')
+
+
+def encode_item(builder: TreeBuilder, item: Item):
+    builder.start('item', {})
+    encode_inline(builder, 'title', item.title)
+    encode_inline(builder, 'description', item.description)
+    encode_inline(builder, 'link', item.link)
+    encode_inline(builder, 'author', item.author)
+    encode_inline(builder, 'comments', item.comments)
+    encode_inline(builder, 'guid', item.guid)
+    encode_inline(builder, 'pubDate', encode_date(item.pubDate))
+
+    for cat in item.categories:
+        encode_inline(builder, 'category', cat)
+
+    for e in item.enclosures:
+        encode_enclosure(builder, e)
+
+    builder.end('item')
+
+
+def encode(channel: Channel) -> bytes:
+    """Given a Channel, emit RSS"""
+    # https://docs.python.org/3/library/xml.etree.elementtree.html#treebuilder-objects
+
+    builder = TreeBuilder()
+    headers = {
+        'version': '2.0',
+    }
+
+    builder.start('rss', headers)
+    builder.start('channel', {})
+
+    encode_inline(builder, 'title', channel.title)
+    encode_inline(builder, 'link', channel.link)
+    encode_inline(builder, 'language', channel.language)
+    encode_inline(builder, 'description', channel.description)
+    encode_inline(builder, 'copyright', channel.copyright)
+    encode_inline(builder, 'managingEditor', channel.managingEditor)
+    encode_inline(builder, 'pubDate', encode_date(channel.pubDate))
+    encode_inline(builder, 'lastBuildDate', encode_date(channel.lastBuildDate))
+    encode_inline(builder, 'webMaster', channel.webMaster)
+    encode_inline(builder, 'generator', channel.generator)
+    encode_inline(builder, 'docs', channel.docs)
+    encode_inline(builder, 'ttl', str(channel.ttl) if channel.ttl is not None else None)
+
+    if channel.image:
+        encode_image(builder, channel.image)
+
+    for cat in channel.categories:
+        encode_inline(builder, 'category', cat)
+
+    for i in channel.items:
+        encode_item(builder, i)
+
+    builder.end('channel')
+    builder.end('rss')
+
+    rss = builder.close()
+    return to_xml_string(rss)

src/notifications/service.py

@@ -0,0 +1,12 @@
+from typing import Any
+from .data import Notification
+import abc
+
+
+class NotificationService(abc.ABC):
+    """A service that sends notifications to the user"""
+
+    @abc.abstractmethod
+    def notify(self, user: Any, notification: Notification):
+        """Send a notification"""
+        pass