Last active
November 20, 2022 18:46
-
-
Save lewoudar/08738c7029b9b9a2236af47142e97035 to your computer and use it in GitHub Desktop.
FastAPI server to to schedule tweet creation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # 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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment