Echofeed was a service to take rss feeds and ’echo’ them to other services such as Mastodon. Recently they have been winding down the service for numerous but very understandable reasons. While I have no plans on offering an alternate service, I wrote my own simple version to continue having a similar flow of blog -> mastodon posts.

django-feedparser-low

Since I use django for most of my projects, my first search was to PyPI to see if there were any projects that combined django with feedparser (a popular library for processing rss/atom/json feeds). While there are a few listed, most seem to be a little older and I wanted something that more simply integrated with Django, and used the new tasks framework . My first draft of this is django-feedparser-low to handle the lower level basics (naming is hard!).

The goal with this, was to provide the basic models and task functions to handle scraping.

The basic models.py provides a Feed model to track subscriptions, and a basic Entry model. It’s likely that these may be slightly too simple once I get more production testing in, but this was a good starting point.

Using the tasks framework , I provide a tasks.py to handle scheduling the scrapes and the individual scrapes. Feedparser already has support to look at etag and modified times for feed, so the task tries to keep track of that, and exits early on a http 304

The most complicated part is likely in decorators.py where I have a test feed-filter decorator to combine Django’s post_save signal with a task object. The goal is that we can have a task, that is automatically called by our Entry model’s post_save if it passes some kind of conditional. This would allow us to more easily have tasks to process specific feeds. From our tasks.py example we can see it in use.

from django.tasks import task

from django_feedparser_low.decorators import feed_filter
from django_feedparser_low.models import Entry

@feed_filter(lambda entry: entry.feed.url == "https://example.com/index.xml")
@task
def process_entry(pk: str):
    # Only trigger this task, if the scraped Entry matches our filter
    entry: Entry = Entry.objects.get(pk=pk)
    print(entry)

Mastodon posting

This part is not currently open sourced, but with the above framework, now I can try to post to Mastodon.

I start with a simple model to link feed Entry with mastodon posts

class EchoPost(models.Model):
    owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.DB_CASCADE)
    entry = models.ForeignKey("django_feedparser_low.Entry", on_delete=models.DB_CASCADE)

    status_url = models.URLField(help_text="Remote Mastodon Post")
    status_id = models.CharField(help_text="Remote Mastodon Post")

I can then use my earlier feed_filter decorator to map my feed to my mastodon posting code with an extra check to avoid spamming mastodon with old posts.

@feed_filter(lambda i: i.feed.url == "https://paultraylor.net/index.xml")
@task
def process_entry(pk: str):
    entry: Entry = Entry.objects.get(pk=pk)
    # I don't want a new feed to spam my mastodon feed, so I only post entries newer than
    # when I first register the feed
    if entry.published < entry.feed.created_on:
        logger.debug("Skipping old post for now: %s %s", entry.published, entry.title)
        return
    # MastodonSession is a small requests wrapper that handles part of the rest api
    with MastodonSession(entry.feed.owner.username) as client:
        # need to convert an rss entry to mastodon status text
        # This includes the entry title, a link back, and possibly some tags
        status = entry_to_mastodon_status(entry)
        try:
            echo = models.EchoPost.objects.get(entry=entry)
        except models.EchoPost.DoesNotExist:
            logger.debug("Posting to mastodon %s", entry.url)
            # # https://docs.joinmastodon.org/methods/statuses/#create
            result = client.status_create(
                status=status,
                idempotency=entry.pk,
            )
            echo = models.EchoPost.objects.create(
                owner=entry.feed.owner,
                feed=entry.feed,
                entry=entry,
                status_url=result["url"],
                status_id=result["id"],
            )
            logger.info("Created echo %s", echo.status_url)
        else:
            # https://docs.joinmastodon.org/methods/statuses/#edit
            result = client.status_update(id=echo.status_id, status=status)
            logger.info("Updated echo %s", echo.status_url)

Future Work

It’s possible I could publish some of my mastodon posting code in the future, but I am not yet sure how that might look. Instead of relying just on polling, I also want to look at automatically processing a feed after publishing. While I will likely not use a full websub implementation, it should be easy to do something simpler. With this distraction out of the way, and more control over how I echo posts to Mastodon, maybe that’ll inspire me to write more to my blog.