Using Azure Storage Queues and Python to Automate Sending Tweets

October was ADHD Awareness Month and this year for October, I wanted to do something to bring awareness and encouragement to folks with ADHD and other neurodiversities.

Jump To

• Intro
• Storage Queues
• Creating Images
• Twitter Access
• Improving the Process
• Running the Script
• Was It Worth It
• Check it Out

Aside from speaking at Refactr.Tech and appearing and a few podcasts, I wanted to create something that:

1. Would have a wider reach. 
2. Would let me play with Python and Azure

If I look at where I have the largest reach, that's Twitter. So I decided to create an automated Twitter Campaign.

If I want to play with Python - Then I can send tweets programatically.

If I want to play with Azure... I can schedule the tweets with Azure Functions

You can see the results in Part 3

Getting Content

If you haven't seen the other parts in this series the TLDR was I asked for folks with neurodiversities (ADHD, Bipolar, Dyslexia, etc) to fill out a short-survey with the last part requesting the participant to send an anonymous message to folks that may have recently been diagnosed or are considering seeking a diagnosis.

Storage Queues

I needed a way to store the message so that they could be called.

Originally I was thinking to use a NoSQL Database like Cosmos DB. I could have added the schedule day and then have the script pull the message for the current day.

This would work but it felt like a lot to spin up a Database to store 30 documents.

My next thought was to do something lighter (like SQLite) and then I remembered a talk from PyTexas about using queues in Python. While I didn't want a script running for that long I started thinking about storing the text in a queue. This is when I remember there being Azure Storage Queues.

In short, a storage queue allows you to store millions of messages (either bytes or unicode). Its primary function is to pull data off the top of the queue and either delete the message or push it to the bottom.

This was perfect because I already had a list of messages. I just needed to add them to the queue. Then my azure function would just call from the queue processing the message.

from azure.storage.queues import QueueClient
import json

queue = QueueClient.from_connection_string(
    conn_str=<CONNECTION_STRING>, queue_name=<QUEUE_NAME>
)

for idx, message in enumerate(messages):
    queue.send_message({"index": idx, "text": message})

# to receive the message

_msg = json.loads(queue.receive_message()['content'])

Creating Images

Sending text to twitter didn't seem like the greatest option. There is a character limit and studies show tweets are more engaging when you add images. This gave me the idea of creating a base image and overlaying the text on the image.

I was able to build a simple base image using Canva. Then, I overlayed the text using Pillow. Working with text on images is somwhat complicated to get perfect so I opted to create 4 categories and a ratio for each one.

{
    "md": 0.010,
    "lg": 0.012,
    "xl": 0.014,
    "2xl": 0.020,
}

Then I used that ratio to autoscale the text until the bounding box was the desired size.

draw = ImageDraw.Draw(image)
font_size = 1
font = ImageFont.truetype("assets/Lato-BoldItalic.ttf", font_size)

while font.getbbox(text)[1] < image_size_ratio:
   font_size += 1
   font = ImageFont.truetype("assets/Lato-BoldItalic.ttf", font_size)

It wasn't perfect, but it was good_enough. Then I added that index and a hashtag onto the image, and I was ready to go.

Twitter Access

The next step was sending the tweet. Twitter has an API in which you can use to interact with the application. The most basic access is free. But in order to tweet images, I needed to request elevated access. Instead of using the API directly, I opted to use the python package Tweepy

First connect your twitter account using your CLIENT AND ACCESS KEYS AND SECRETS. Then you'll need to store the image as an object using the v1 API or (API).

import tweepy
from bytes import BytesIO #This (lets you save the image in memory and not to a file)

auth = tweepy.OAuth1UserHandler(
            consumer_key=self.consumer_key,
            consumer_secret=self.consumer_secret,
            access_token=self.access_token,
            access_token_secret=self.access_token_secret,
)

image = BytesIO()
img.save(image, format="PNG")
image.seek(0)
api = tweepy.API(auth)
media = api.media_upload(file=image, filename="my_file.png")

Then you use the v2 API (Client) to send the message embedding the media.

# with the message queued message as _msg
self.twitterv2.create_tweet(
        "text": f"{_msg['index']}: {_msg['text']} #31DaysOfNeurodivergence",
        "file": media.id
        "filename": f"{_msg['index']}{_msg['text'][:10]}",
    })

Improving the Process

There were some challenges in figuring out this process. My teammate Pamela Fox and I put together a little module called AZ Queue Tweeter that expanded on this process, allowing you to not only send tweets with images to storage queues, but also supports the full range of a personal account on twitter. Here is an example of Pamela sending a Twitter poll.

import json

qt.queue_message(
    json.dumps({
        "text": "Whats your fav Python web framework?",
        "poll_options": ["Flask", "Django", "FastAPI", "All of em!"],
        "poll_duration_minutes": 60*24}
    )
)

qt.send_next_message(message_transformer=lambda msg: json.loads(msg))

This also introduced the ability to manipulate the text prior to sending it. In my script, I use spacy to segment the messages into sentences keeping the text to 280 characters.

 Running the Script

So we've walked through how to make the tweet. How do we ensure that the tweets run everyday? This is where Azure functions come to play. I was able to create a simple timer function that runs once a day for the month of October.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "mytimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "0 0 12 * 10 *"
    }
  ]
}

You can use the CLI or VS Code.

Then pass a function that retrieves a message, creates the image, and sends the tweet.

def main(mytimer: func.TimerRequest) -> None:
    queue = QueueTweeter(storage_auth=sa, twitter_auth=ta)
    msg = queue.send_next_message(
        message_transformer=load_message, preview_mode=False, delete_after=False
    )

    # Message is in format "Index - Text"
    logging.info(f"{msg} triggered at {datetime.datetime.utcnow()}")

Was this worth the effort

When I think about the problem that I had (a lot of tweets getting sent over a large period of time) I don't think I had a lot of options that were "fully automated".

I did learn some new tools and some interesting solves (looking at you PILLOW), also my overall cost for the month was less than $0.10.

Solving the problem was actually incredibly valuable as I now intend to use the same tooling to automate some of my podcasts and other content. I can also quickly modify this to use Azure queues to send message to other channels (like Discord or Teams).

I've already converted this queue into a website using Flask + HTMX.

Check it Out

Check out the repo to see how I was able to build this campaign. Finally check out AZ Queue Tweeter to create your own campaigns!