pythreads 0.2.0

A Python wrapper of Meta's Threads API

PyThreads

PyThreads is a Python wrapper for Meta's Threads API. It is still in beta, but is well-tested and covers all the published endpoints documented by Meta.

Since it is in pre-release, the API is not yet guaranteed to be stable. Once there's been some opportunity to weed out any bugs and identify any DX inconveniences, a v1 with a stable API will be released. This project follows Semantic Versioning.

---

Table of Contents

• Installation
• Environment Variables
• Authentication & Authorization
• Making Requests
• API Methods
• Roadmap
• License

Installation

pip install pythreads

Environment Variables

You will need to create an app in the developer console which has the Threads Use Case enabled and add the the following variables to your environment. The redirect URI needs to be the URL in your application where you call the complete_authorization method.

THREADS_REDIRECT_URI=
THREADS_APP_ID=
THREADS_API_SECRET=

Authentication & Authorization

Authenticating with PyThreads is very simple:

1. Generate an authorization url and state key. Make the auth_url the href of a link or button and store the state_key (which is a randomly generated, opaque string used to mitigate CSRF attacks) somewhere you can reference later when Threads redirects the user back to your redirect URI.

   auth_url, state_key = Threads.authorization_url()
   

2. When the user clicks the link/button, they will be sent to Threads to authenticate and authorize your application access to their account. Upon authorization, they will be sent to your THREADS_REDIRECT_URI. At your THREADS_REDIRECT_URI endpoint, call complete_authorization with the full URL of the request made to your server (which contains the auth code) and the state_key, which was generated in the previous step:

   credentials = Threads.complete_authorization(requested_url, state_key)
   

3. This method automatically exchanges the auth code for a short-lived token and immediately exchanges the short-lived token for a long-lived token. The method returns Credentials object, which contains this long-lived token and details about the user. The Credentials object can be serialized and deserialized to/from JSON, making it easy for you to persist it in some data store or store it encrypted in the user's session.

   json = credentials.to_json()
   

   { 
     "user_id": "someid", 
     "scopes": ["threads_basic"], 
     "short_lived": false, 
     "access_token": "someaccesstoken", 
     "expiration": "2024-06-23T18:25:43.511Z"
   }
   

   Credentials.from_json(json)
   
   # or
   
   Credentials(
       user_id="someid", 
       scopes=["threads_basic"], 
       short_lived=false, 
       access_token="someaccesstoken", 
       expiration=datetime.datetime(2024, 6, 23, 18, 25, 43, 121680, tzinfo=datetime.timezone.utc)
   )
   

4. Long-lived tokens last 60 days, and are refreshable, as long as the token hasn't expired yet. Implement your own application logic to determine when it makes sense to refresh users' long-lived tokens, which you can do with:

   refreshed_credentials = Threads.refresh_long_lived_token(old_credentials)
   

   A Credentials object has convenience methods to make it easier for you to determine whether the token is still valid and how much longer it is valid for.

   credentials.expired()
   >>> False
   
   credentials.expires_in()
   >>> 7200 # seconds
   

   Of course, you can always check the expiration time directly. It is stored in UTC time:

   credentials.expiration
   >>> datetime(2024, 7, 11, 10, 50, 32, 870181, tzinfo=datetime.timezone.utc)
   

   If you call an API method using expired credentials, a ThreadsAccessTokenExpired exception will be raised.

Making Requests

Once you have a valid Credentials object, you can use an API object to call the Threads API. The API object uses an aiohttp.ClientSession to make async HTTP calls. You may either supply your own session or let the library create and manage one itself.

If you do not supply your own session, PyThreads will create one and take responsibility for closing it. You must use the API object as an async context manager if you want it to manage a session for you:

# Retrieve the user's credentials from whereever you're storing them:
credentials = Credentials.from_json(stored_json)

async with API(credentials=credentials) as api:
    await api.threads()

If you want to create and manage your own session (e.g. you're already using aiohttp elsewhere in your application and want to use a single session for all requests):

# Create an `aiohttp.ClientSession` at some point:
session = aiohttp.ClientSession()

# Retrieve the user's credentials from whereever you're storing them:
credentials = Credentials.from_json(stored_json)

api = API(credentials=credentials, session=session)
threads = await api.threads()

# If you supply your own session, you are responsible for closing it:
session.close()

API Methods

Better documentation is coming, but for now, you can browse the methods in api.py.

Most of the methods follow Meta's API closely, with required/optional arguments matching API required/optional parameters.

Making a text-only post

Making a text-only post is a two-step process. Create a container and then publish it:

async with API(credentials) as api:
    container_id = await api.create_container("A text-only post")
    result_id = await api.publish_container(container_id)
    # container_id == result_id

Making a post with a single media file

Making a post with a single media file is also a two-step process. Create a container with the media file and any post text and then publish it:

async with API(credentials) as api:
    # Create a video container. You must put media resources at a publicly-accessible URL where Threads can download it.
    a_video = Media(type=MediaType.VIDEO, url="https://mybucket.s3.amazonaws.com/video.mp4")
    container_id = await api.create_container(media=a_video)

    # Video containers need to complete processing before you can publish them
    await asyncio.sleep(15)

    # Check the status to see if it's finished processing
    status = await api.container_status(container_id)
    # >>> ContainerStatus(id='14781862679302648', status=<PublishingStatus.FINISHED: 'FINISHED'>, error=None)

    # Publish the video container
    result_id = await api.publish_container(container_id)
    # container_id == result_id

Making a carousel post

Making a post with a media carousel is a three-step process. Create a container for each media file in the carousel, then create a container for the carousel (attaching the media containers as children), and publish the carousel container:

async with API(self.credentials) as api:
    # Create an image container
    an_image = Media(type=MediaType.IMAGE, url="https://mybucket.s3.amazonaws.com/python.png")
    image_id = await api.create_container(media=an_image, is_carousel_item=True)

    # Create a video container
    a_video = Media(type=MediaType.VIDEO, url="https://mybucket.s3.amazonaws.com/video.mp4")
    video_id = await api.create_container(media=a_video, is_carousel_item=True)

    # Video containers need to complete processing before you can publish them
    await asyncio.sleep(15)

    # Check the status to see if the containers are finished processing
    status_1 = await api.container_status(image_id)
    status_2 = await api.container_status(video_id)
    # >>> ContainerStatus(id='14781862679302648', status=<PublishingStatus.FINISHED: 'FINISHED'>, error=None)
    # >>> ContainerStatus(id='14823646267930264', status=<PublishingStatus.FINISHED: 'FINISHED'>, error=None)

    # Create the carousel container, which wraps the media containers as children
    carousel_id = await api.create_carousel_container(containers=[status_1, status_2], text="Here's a carousel")

    await asyncio.sleep(15)

    # Check the carousel container status
    carousel_status = await api.container_status(carousel_id)
    # >>> ContainerStatus(id='15766826793021848', status=<PublishingStatus.FINISHED: 'FINISHED'>, error=None)

    # Publish the carousel container
    result_id = await api.publish_container(carousel_id)
    # carousel_id == result_id

A few key things to point out above:

1. Creating media containers requires you to put the image or video at a publicly-accessible URL. Meta retrieves the media file from that URL.

2. When creating a media container with a video, Threads requires you to wait for the video to be processed before you can either publish them or attach them to a carousel container. This can take seconds or minutes.

You are strongly encouraged to read Meta's documentation regarding the posting process:

• Post to Threads
• Understanding Container Status

Roadmap

• Improve documentation of API methods and publish the docs.
• Type the return values of the API methods. They currently all return Any.
• Add integration with S3 and R2 storage. The Threads API doesn't take media uploads directly. You have to upload files to a publicly accessible URL and pass the URL in the API response. This integration would handle the upload to cloud storage and passing of the URL to the Threads API for you.
• Explore adding JSON fixtures of expected responses to specs.

License

pythreads is distributed under the terms of the MIT license.