FastAPI server to to schedule tweet creation

twitter_server_for_postponed_tweet_creation.py

# you will need to install the following libraries
# - FastAPI
# - tweepy (version 4.X)
# - twitter-text-parser
# - apscheduler (version 3.X)
import logging
import os
from datetime import datetime
from typing import Optional, List

import tweepy
import twitter_text
from fastapi import FastAPI, HTTPException, Response
from apscheduler.schedulers.background import BackgroundScheduler
from pydantic import BaseModel, validator

logger = logging.getLogger(__name__)
scheduler = BackgroundScheduler()


def start_scheduler():
    scheduler.configure({'apscheduler.timezone': 'UTC'})
    scheduler.start()


def stop_scheduler():
    scheduler.shutdown()


app = FastAPI(on_startup=[start_scheduler], on_shutdown=[stop_scheduler])


class Media(BaseModel):
    media_ids: Optional[List[str]]
    tagged_user_ids: Optional[List[str]]


class TweetInput(BaseModel):
    creation_date: datetime
    text: Optional[str]
    media: Optional[Media]

    @validator('creation_date')
    def check_creation_date(cls, value: datetime) -> datetime:
        if value <= datetime.utcnow():
            logger.error('tweet creation date is less than or equal to current time')
            raise ValueError('tweet creation date must be in the future')

        return value

    @validator('text')
    def check_text_length(cls, value: Optional[str]) -> Optional[str]:
        if value is None:
            return value

        result = twitter_text.parse_tweet(value)
        if not result.valid:
            logger.error('tweet text exceeded 280 characters according to Twitter rules')
            raise ValueError('text is more than 280 characters according to Twitter rules')

        return value

    def validate_tweet(self) -> None:
        if self.text is None and self.media is None:
            logger.error('no text or media information was provided in tweet input data')
            # the goal here is to have consistent 422 error messages
            detail = [
                {
                    'loc': ['__root__'],
                    'msg': 'At least text or media information should be provided',
                    'type': 'value_error.missing'
                }
            ]
            raise HTTPException(status_code=422, detail=detail)


def create_tweet(data: dict) -> None:
    payload = {'text': data.get('text')}
    media = data.get('media')
    if media is not None:
        payload['media_ids'] = media.get('media_ids')
        payload['media_tagged_user_ids'] = media.get('tagged_user_ids')

    client = tweepy.Client(os.getenv('ACCESS_TOKEN'))
    client.create_tweet(user_auth=False, **payload)


@app.post('/tweet')
def post_tweet(data: TweetInput):
    data.validate_tweet()
    logger.info('schedule tweet creation at %s', data.creation_date.isoformat())
    scheduler.add_job(create_tweet, 'date', args=(data.dict(exclude={'creation_date'}),), run_date=data.creation_date)
    return Response(status_code=200)