DEV Community: Kevin Tewouda

Create a Twitter bot in python - part 2

Kevin Tewouda — Tue, 27 Dec 2022 12:20:19 +0000

In this second part, we will see how to send an email with collected tweets and retweet relevant tweets instantly. If you haven't read the first part, you can find it here.

Weekly sending of search results by email

We will first see how to send an email with an attachment of the different tweets resulting from the search for positive messages about the movie Wakanda Forever every Monday at 9 am, resulting from past week.We will use the following python libraries:

apscheduler: to schedule repetitive tasks.
emails: to send emails.

Here is the resulting code that we will comment on next:

Notes:

There are some environment variables to configure before sending the email. Email login and password are only necessary if using TLS, which will be the case most of the time.
To test email sending locally, you can use the mailhog service.
Lines 34 to 47, to store the tweets in the file, I use the JSON lines format. It is convenient to save a lot of data without eating up all the memory.
In lines 49 to 60, we define email information (you can modify them as you wish), the attachment, and the recipient. I print an error message when the email is not sent but you can configure logging instead. By the way, to try to debug the error, you should configure logging to display messages from the emails library at debug level.
Line 66, we define the scheduler, and we use the Blocking version which is appropriate to our use case. But depending on the type of project you are building, you could use the Threading or Asyncio version. I let you look at the documentation for more details.
Line 70, we schedule our job using the crontab syntax. If you want to check the syntax of your crontab before applying it, you can use this website.
Line 74, our scheduler will run forever.

You can host your application for free on platforms like fly.io.

Retweet Wakanda tribute

For the second application, we want to retweet instantly tweets paying tribute to our favorite movie. Unfortunately, we can't use the bearer token here anymore. We need an access token with specific rights. If we look at the endpoint documentation, we see that we need the permissions tweet:read, tweet:write and users:read. The workflow to gain an access token is defined here. Nevertheless, I will show you how to do it with tweepy. You will need a redirect url. This url will be used to send a code necessary to gain the access token. On this url, two pieces of information will be passed as query parameters:

state: a random security string. More information about oauth2 vocabulary can be found here.
code: a security value returned by the Twitter API.

Here is an example of a FastAPI server to set up quickly a redirect url.

import logging
from fastapi import FastAPI, Response

app = FastAPI()
logger = logging.getLogger(__name__)

@app.get('/')
def get_code(code: str, state: str):
    logger.info('code: %s, state: %s', code, state)
    return Response(status_code=200)

Once you have a redirect url set up, you need to go back to your developer portal, on the detail page of your project, you have a section User authentication settings, click here. You can pass the first section App permissions which concerns oauth1 that we don't use. In the second section Type of App, choose Web app, Automated App or bot. In the section App info, fill in the redirection url corresponding to your server. You will also have to fill in a personal url (you can put anything as long as it is linked to you) and other optional information if you want. Once this is done, you will get a client_id and a client_secret needed for oauth2 authentication. Keep them in a safe place. We will use in the next scripts the CLIENT_ID and CLIENT_SECRET environment variables which must contain these values. As for the workflow to obtain an access token with tweepy, it is as follows:

1 - Use the tweepy.OAuth2UserHandler class by filling in all the necessary information.

import os
import tweepy

oauth2_user_handler = tweepy.OAuth2UserHandler(
    client_id=os.getenv('CLIENT_ID'),
    client_secret=os.getenv('CLIENT_SECRET'),
    redirect_uri='votre url de redirection ici',
    # we list the permissions we want here
    scope=['tweet.read', 'tweet.write', 'users.read', 'offline.access'],
)
# it will generate the authorization url that we must launch in our web browser.
print(oauth2_user_handler.get_authorization_url())

You will notice for the scope argument that I added the offline.access permission. It will help to refresh the access token without manual intervention, I will explain it later. Once you have the authorization url, copy it into the address bar of your browser and follow it (click on Enter). You will have to authorize your bot to have access to your account, once this is done you will be redirected to the url that you have defined as your redirection url.

2 - Copy the redirection url in the browser that should contain the code assigned to you, and use a method of OAuth2UserHandler that will retrieve the access token.

access_token = oauth2_user_handler.fetch_token(
    'redirection url here'
)

3 - Once this is done you can use the tweepy client as usual, except that instead of the bearer token, the access token will be used.

client = tweepy.Client('access token')

Now you can create and retweet tweets as you wish:

import os
import tweepy

client = tweepy.Client(os.getenv('ACCESS_TOKEN'))
tweet_response = client.create_tweet(text='Hello from bot!', user_auth=False)
response = client.retweet(tweet_response.data['id'], user_auth=False)
print(response)

The argument user_auth=False is important otherwise tweepy will try an oauth1 authentication and the request will fail. It's a bit strange as an API, but it's the legacy of the old API. For your information, an access token is valid for two hours. And now some of you must be wondering if they will have to repeat the operation manually with the browser... That doesn't sound very automated. Remember the offline.access permission we used at the beginning to get the authorization url? It will be used to refresh our token before it expires. In fact, when we retrieved the access token, our permission also allowed us to retrieve a refresh token. It is this last token that is used for the refresh. It is kept internally by tweepy. To refresh the access token with tweepy, here is how we proceed:

import tweepy

token_info = oauth2_user_handler.refresh_token(
    'https://api.twitter.com/2/oauth2/token'
)
client = tweepy.Client(token['access_token'])

We have to pass the url to refresh a token. If you wonder where I found this url, it is available on this page under the Step 5… section. If you have a refresh_token that you got in another way than by tweepy, you can pass it with the refresh_token argument. In token_info we have a set of information including the access token and the new refresh token (again saved by tweepy for later use). So we can instantiate a new client with the new access token without having to make a manual intervention. And to automate everything, we can use apscheduler that we know well now. An example of code that you can write:

import os

import tweepy
from apscheduler.schedulers.blocking import BlockingScheduler


scheduler = BlockingScheduler()
oauth2_user_handler = tweepy.OAuth2UserHandler(
    client_id=os.getenv('CLIENT_ID'),
    client_secret=os.getenv('CLIENT_SECRET'),
    redirect_uri='your redirection url here',
    scope=['tweet.read', 'tweet.write', 'users.read', 'offline.access'],
)

def refresh_token():
    token_data = oauth2_user_handler.refresh_token(
        'https://api.twitter.com/2/oauth2/token'
    )
    # save the token where you want
    os.environ['ACCESS_TOKEN'] = token_data['access_token']

# think to add this job the first time you get an access token
scheduler.add_job(refresh_token, 'interval', hours=1, minutes=55)
scheduler.start()

Here we use the interval trigger to refresh our token every one hour and fifty-five minutes since the token only lasts two hours. 😁

Finally, here is the code to automatically retweet every tweet paying tribute to the movie Wakanda Forever.

import os

import tweepy


client = tweepy.Client(os.getenv('ACCESS_TOKEN'))


class WakandaTributeRetweet(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        print(tweet.id, tweet.text)
        client.retweet(tweet.id, user_auth=False)
    def on_errors(self, errors):
        print(errors)
    def on_connection_error(self):
        self.disconnect()

bot = WakandaTributeRetweet(os.getenv('BEARER_TOKEN'))
bot.filter()

Note: You need to still have the rules we created in the first part for the above script to work.

Bonus: schedule the creation of a tweet at a later date

Surprisingly enough, there is no way to schedule tweet creation as we can do on the web application. But we can do it ourselves with all the tools we already know. 😉

We will need a server to get tweet creation requests and schedule the operations, and a client to send information about the tweet. Let's start with the server. Again I will use FastAPI but you can choose whatever you want.

Notes:

We use the background scheduler this time which is appropriate for a web server.
We use some pydantic validators to check our payload. We check that the creation_date is correct, the tweet text is correct according to Twitter rules, and for that, we use the twitter-text-parser library.
We also add a custom method to check that the payload contains a text or media information.

Now for the client, we will write a command line interface using the click library. I have a tutorial if you want to learn more about it.

from datetime import datetime
from typing import Tuple

import click
import requests


@click.command('tweet')
@click.option('-t', '--text', help='optional text of the tweet')
@click.option(
    '-m',
    '--media-ids',
    multiple=True,
    help='optional media ids for tweet, can be multiple'
)
@click.option(
    '-T',
    '--tagged-user-ids',
    multiple=True,
    help='optional user ids tagged in the tweet, can be multiple'
)
@click.option(
    '-c', '--creation-date',
    type=click.DateTime(),
    required=True,
    help='creation date of the tweet, it must be in UTC timezone'
)
def create_tweet(text: str, media_ids: Tuple[str], tagged_user_ids: Tuple[str], creation_date: datetime):
    """A simple command to create tweet at a later date"""
    payload = {'text': text, 'creation_date': creation_date.isoformat()}
    if media_ids or tagged_user_ids:
        payload['media'] = {'media_ids': media_ids, 'tagged_user_ids': tagged_user_ids}
    # change the url with the url of your server
    print(payload)
    response = requests.post('http://localhost:8000/tweet', json=payload)
    if response.status_code == 200:
        click.secho('tweet creation was scheduled correctly', fg='green')
    else:
        click.secho('Error: ', nl=False, err=True, fg='red')
        click.secho(response.text, err=True, fg='red')

if __name__ == '__main__':
    create_tweet()

You can save this script in a file and run it with the python command. An example usage can look like this: python my_script.py --text "hello from bot" -c "2022-11-20 16:48:00".

This is the end of the tutorial, hope you enjoy reading it as I enjoy writing it. Take care of yourself and see you next time! 😁

This article was originally published on Medium.

If you like my article and want to continue learning with me, don't hesitate to follow me here and subscribe to my newsletter on substack 😉

Create a Twitter bot in python - part 1

Kevin Tewouda — Fri, 09 Dec 2022 07:59:23 +0000

Twitter is one of the most used social networks on the planet. As a result, it is full of many resources and also many annoyances. To help you get the best experience from the social network, various applications have emerged like threadreaderapp or blockpartyapp. These applications use the Twitter API that has existed since 2012.
In this article, we will see the basics to design one of these applications by creating two simple bots that will do the following actions:

Passive listening of messages paying tribute to the Marvel movie Wakanda Forever. And yes, I'm a big fan of Marvel movies. 😆
An instant listening of the same messages with this time an automatic retweet. This is something quite classical for bots.

In this first part of the article, we will see the important notions to know to start with the Twitter API and the endpoints that will interest us to perform our actions. Once this big part is over, the second part will be more fun with the implementation of bots. 😉

Prerequisites

1 - Have a Twitter account (yes, it's essential!). If you don't want to use your personal account, you can create an account just for your bot.

2 - Create a developer account that you will associate with your Twitter account. To do this, go to https://developer.twitter.com. You click on Sign up, then you will have to follow the instructions and choose the Bot option for the use you will make of the API. Once you have created your account, you will be assigned a Bearer Token. Keep it safe, as you will need it to make your requests later.

3 - Have a basic knowledge of python, know how to install it, and initialize a project. If you don't have any skills in this language, there are plenty of resources on the internet like this one.

Twitter API documentation

The documentation of the Twitter API is very well done and available here. It is best to start with the Getting Started and Fundamentals sections. We will work with version 2 of the API which is the most recent and easiest to learn. After that, there is the API endpoint reference that you have to keep somewhere under your pillow. 😉

A bit of terminology

Before manipulating the Twitter API, I want to make a quick point about the different objects we can manipulate and the format of the responses. The objects are the following:

The different endpoints of the Twitter API will return the first two types of objects, namely Tweet and User, for the rest, you will have to use the notion of expansion.

Let's take an example of the route that lists tweets based on their id. I will use the requests library which is very well known for making HTTP calls in the following examples. The API endpoint used is documented here.

import os
from pprint import pprint
import requests

params = {'ids': '1588915242490560512'}
# The bearer token is stored in an environment variable called BEARER_TOKEN
headers = {'Authorization': f"Bearer {os.getenv('BEARER_TOKEN')}"}
r = requests.get('https://api.twitter.com/2/tweets', params=params, headers=headers)
pprint(r.json())
# Sample response
{
    'data': [
        {
            'edit_history_tweet_ids': ['1588915242490560512'],
            'id': '1588915242490560512',
            'text': 'in what is becoming a tradition in odd point release...'
        }
    ]
}

Note: if you wonder how to get a tweet id, click on a tweet from a web browser, look at the url, and after the status part, the numerical value that follows is its id. 😁 ✌️

You will notice that the data is in the data part of the response and that each object represents a tweet. Here there is only one object as I have only selected one in the query field. If you have observed the details of a Tweet object you will notice that not all the fields are present in the response, which is intended to avoid having too large responses. To display more fields, you'll have to use an additional query field tweet.fields. It will be the same with other types of objects, so if you want to get more fields from a user, for example, you will have to configure the user.fields field.
To come back to our example, let's say that we will display the date of creation of the tweet and the author's id.

import os
from pprint import pprint

import requests
params = {
    'ids': '1588915242490560512',
    # we add the extra fields we want
    'tweet.fields': 'created_at,author_id'
}
headers = {'Authorization': f"Bearer {os.getenv('BEARER_TOKEN')}"}
r = requests.get('https://api.twitter.com/2/tweets', params=params, headers=headers)
pprint(r.json(), indent=4)
# Sample response
{
    'data': [
        {
            'author_id': '1037022474762768384',
            'created_at': '2022-11-05T15:24:49.000Z',
            'edit_history_tweet_ids': ['1588915242490560512'],
            'id': '1588915242490560512',
            'text': 'in what is becoming a tradition in odd point release...',
        }
    ]
}

Bingo! We have our extra fields! But it would be nice to have more information about the author, not just his id. How to do that? This is where expansions come in! Expansions allow you to add additional objects in the response, here it's a User we want to include, but it could be a Media or even a Tweet related to the one we searched (parent tweet for example if we are in a conversation). The list of possible expansions is displayed in the documentation of each API endpoint, for the one we use you will find it here.

Our example will look like this:

import os
from pprint import pprint

import requests
params = {
    'ids': '1588915242490560512',
    # we add the extra fields we want
    'tweet.fields': 'created_at,author_id',
    # we include user information in the response
    'expansions': 'author_id'
}
headers = {'Authorization': f"Bearer {os.getenv('BEARER_TOKEN')}"}
r = requests.get('https://api.twitter.com/2/tweets', params=params, headers=headers)
pprint(r.json(), indent=4)
# Sample response
{
    'data': [
        {
            'author_id': '1037022474762768384',
            'created_at': '2022-11-05T15:24:49.000Z',
            'edit_history_tweet_ids': ['1588915242490560512'],
            'id': '1588915242490560512',
            'text': 'in what is becoming a tradition in odd point release...',
        }
    ],
    'includes': {
        'users': [
            {
                'id': '1037022474762768384',
                'name': 'htmx.org',
                'username': 'htmx_org'
            }
        ]
    }
}

Et voilà! You will notice a new includes key added in the answer. It includes all the expansions we requested. In our case, it is the author's information. Again, this is default information, if you want to see more information from the user, you'll have to add a user.fields query parameter with the desired fields. I leave it as an exercise. 😁

Note: Most of the responses returned by the Twitter API endpoints have this structure with four root keys:

data: the requested main object is returned in this key. It can be an object or a list of objects.
includes: which contains objects resulting from expansions.
meta: an optional key that will contain pagination information when there are many objects to return.
errors: an optional key if there is a problem with the request. It will contain the list of errors.

Note: Due to a large number of users of the Twitter API, it goes without saying that there are limitations on the number of calls you can make. More information is on this page. Note that the documentation for each endpoint also specifies the number of calls you can make on that route during a 15-minute window.

Observe a theme using keywords

Now let's get to the heart of the matter, we'll see how to observe specific events that happen on Twitter and perform an appropriate action. There are two endpoints that will interest us.

Search recent tweets.

The documentation of this endpoint can be found here. It can be used in two ways:

Obtaining historical data: In this mode, we search for tweets from the last seven days.
Polling: In this mode, we search for recent tweets in relation to our last search. If it's not clear, don't be afraid, you'll understand with an example :)

Build a search filter

The first step will be to create a filter that will allow us to search for specific tweets. The documentation on how to create a filter can be found here and a more advanced tutorial here. I will however mention some information.

1 - Depending on the type of access you have, probably Essential if you are just starting out, you will be limited to creating a filter of 512 characters in length.

2 - You can use several operators to perform your search, some of the most important are:

AND which is naturally used by separating the search keywords with a space. For example, if I want a tweet that includes the words banana and potato (I'm not hungry!) I would build a banana potato filter.
OR if I want a tweet that includes banana or potato, I would write (banana OR potato). Note that parentheses are necessary, for example, if you want to search for exact sentences like ("Twitter API" OR #v2).
- the negation of a search, for example, if I search for tweets that have potato but not banana, I will have a filter potato -banana. So a sentence that contains both words will not pass the search.
from: to search for tweets from a specific account, for example from:marvel.
to: to search for tweets in response to a specific account, for example to:marvel.
conversation_id: In discussion threads, tweets descending from the root tweet will have a conversation_id field to try to reconstruct the whole discussion. This can be used to filter tweets. Example: conversation_id:1334987486343299072.
has: allows you to search for tweets with metadata, for example, a tweet search containing media or links: (has:media OR has:links).
is: allows you to search for different types of tweets such as retweets (is:retweet), quotes, or retweets with a message (is:quote) or other contextual information such as whether the user is verified (you know the little blue mark that may soon become payable :p) is:verified.

3 - The complete list of filters is available here.

For our example, we will search positive messages about the last Marvel movie (at the moment of writing) Wakanda Forever. The filter will be the following: ("black panther" OR #wakandaforever) (magnificent OR amazing OR excellent OR awesome OR great) -is:retweet. Normally, you should understand it but let's unpack it, so we look for tweets:

Having the term black panther (for your information the search is case-insensitive) or the hashtag #wakandaforever ("black panther" OR #wakandaforever)
and having one of the following words: magnificent, amazing, excellent, awesome, great (magnificent OR amazing OR excellent OR awesome OR great)
and is not a retweet -is:retweet. This last filter is very important and even recommended in the official documentation because many tweets are often retweets, it avoids adding unnecessary noise in our replies.

Get historical

As a reminder, the route we are interested in allows us to search for tweets over the last 7 days. We could continue to work with the requests library, but it would be tedious to analyze the results, manage the pagination, etc. That's why we are going to use a specialized library for the Twitter API named tweepy. We will use its tweepy.Client class to manipulate the endpoints of the Twitter v2 API. To know the mapping between the methods of this class and the API endpoints, please refer to this page. In our case, the method we are interested in is search_recent_tweets.

import os

import tweepy

client = tweepy.Client(os.getenv('BEARER_TOKEN'))
response = client.search_recent_tweets(
    '("black panther" OR #wakandaforever) (magnificent OR amazing OR excellent OR awesome OR great) -is:retweet',
    max_results=100,
    tweet_fields=['created_at']
)
if response.data is not None:
    for tweet in response.data:
        print(tweet.id, tweet.created_at, tweet.text)
# pagination information is listed here
print(response.meta)

Notes:

The usage is simple, here we have specified the filter which is the only mandatory argument. Then we specified the number of elements we wanted to display and the additional fields. In general, the method arguments are the same as the API endpoint can take. To know all the possible arguments to pass to this method, refer to its documentation.
The returned response object is a namedtuple which contains four keys data, includes, meta and errors. The same as when you make the request yourself requests for example :)
In response.data, we have the list of requested objects. As a rule, if you know the list of fields of the object, then you can use the field names as properties as done in the example above. However, they are all documented in the tweepy documentation :)

Now imagine the case where you have more than 100 results, how do you iterate over all the results? This is where we use the information provided by response.meta. We will have a next_token information that we will have to use in our method. We could complete the previous code with this piece of code:

next_token = response.meta.get('next_token')

while next_token is not None:
    response = client.search_recent_tweets(
        '("black panther" OR #wakandaforever) (magnificent OR amazing OR excellent OR awesome OR great) -is:retweet',
        max_results=100,
        tweet_fields=['created_at'],
        next_token=next_token
    )
    if response.data is not None:
        for tweet in response.data:
            print(tweet.id, tweet.created_at, tweet.text)
    next_token = response.meta.get('next_token')

And there we go! But tweepy has a Paginator class making our lives easier. Here's an example:

import os

import tweepy
client = tweepy.Client(os.getenv('BEARER_TOKEN'))
for response in tweepy.Paginator(
        client.search_recent_tweets,
        '("black panther" OR #wakandaforever) (magnificent OR amazing OR excellent OR awesome OR great) -is:retweet',
        max_results=100
):
    if response.data is not None:
        for tweet in response.data:
            print(tweet.id, tweet.text)

Easy-peasy! You can limit the number of response iterations returned with the limit argument. This way of using Paginator is interesting if you want to read possible expansions contained in response.includes but if you are only interested in the data part, then the flatten method should be useful.

import os

import tweepy

client = tweepy.Client(os.getenv('BEARER_TOKEN'))

for tweet in tweepy.Paginator(
        client.search_recent_tweets,
        '("black panther" OR #wakandaforever) (magnificent OR amazing OR excellent OR awesome OR great) -is:retweet',
        max_results=100
).flatten():
    print(tweet.id, tweet.text)

Here we iterate directly on tweets. Also, we can pass a limit of tweets to retrieve with the limit argument to pass to the flatten method.

Polling

The second way to use the recent tweets endpoint is polling, it's a bit of a real-time mode. Here the idea is to search for all the tweets that match the filter, not in the past, but from a specific tweet. This requires knowing a (recent) tweet that already meets our criteria and iterating from there. For example, if we have a tweet with an id 10000 that matches our search, and we want to know in real-time all the tweets from there that satisfy our filter, we will write a code like the following:

import os

import tweepy

client = tweepy.Client(os.getenv('BEARER_TOKEN'))
response = client.search_recent_tweets(
    '("black panther" OR #wakandaforever) (magnificent OR amazing OR excellent OR awesome OR great) -is:retweet',
    max_results=100,
    tweet_fields=['created_at'],
    since_id=10000  # we can use a string or an integer
)
if response.data is not None:
    for tweet in response.data:
        print(tweet.id, tweet.created_at, tweet.text)

Note: This last example will not work because we are not in the last 7 days window.

If tweets more recent than id 10000 and matching our criteria exist, they will be displayed. To continue iterating on more recent tweets, we will use the data included in response.meta. Let's assume that the latter contains this information:

{
  "newest_id": "12000",
  "oldest_id": "10005",
  "result_count": 7
}

newest_id corresponds to the most recent tweet in the result list, and we will use this value to continue iterating results by replacing the value of since_id with 12000.
However, if you get more results that cannot be returned at once, you will have a next_token key in meta.

{
  "newest_id": "12000",
  "oldest_id": "10005",
  "next_token": "fnsih9chihsnkjbvkjbsc",
  "result_count": 10
}

In this case, if you use since_id=12000, you will lose all the results that were not displayed at once. The solution is to keep since_id at 10000 and add a next_token argument. Iterate over the results until there is no more next_token in meta before resuming with a since_id of 12000. This is what a code that looks for the latest Wakanda Forever tribute could look like:

import os

import tweepy

client = tweepy.Client(os.getenv('BEARER_TOKEN'))
# replace the value here by a relevant tweet id
first_since_id = 10000
search = '("black panther" OR #wakandaforever) (magnificent OR amazing OR excellent OR awesome OR great) -is:retweet'
response = client.search_recent_tweets(
    search,
    max_results=100,
    tweet_fields=['created_at'],
    since_id=first_since_id
)
if response.data is None:
    since_id = first_since_id
else:
    since_id = response.meta.get('newest_id')
    for tweet in response.data:
        print(tweet.id, tweet.created_at, tweet.text)
while True:
    # you can sleep the code here to not exceed the amount of tweets you can read :)
    next_token = response.meta.get('next_token')
    while next_token is not None:
        paginated_response = client.search_recent_tweets(
            search,
            max_results=100,
            tweet_fields=['created_at'],
            next_token=next_token,
            since_id=since_id
        )
        for tweet in paginated_response.data:
            print(tweet.id, tweet.created_at, tweet.text)
        next_token = paginated_response.meta.get('next_token')
    # if the first request returns nothing, we continue with the first since_id
    since_id = response.meta.get('newest_id') or first_since_id
    response = client.search_recent_tweets(
        search,
        max_results=100,
        tweet_fields=['created_at'],
        since_id=since_id
    )
    if response.data is None:
        continue
    for tweet in response.data:
        print(tweet.id, tweet.text)

It's tedious, isn't it? Fortunately, there is another much simpler endpoint to track tweets in real-time and that's what we'll see right now!

Filtered stream

The second method to search for tweets in real-time is the filtered stream. Its documentation can be found here. It includes 3 endpoints to use:

One to add or remove filters.
One to list filters.
One to search tweets.

Unlike the previous endpoint, you can add several filtering rules, the number depends on the type of access you have.
For Essential access, we have 5 rules of 512 characters each. This is probably the case for you if you are new to the Twitter API.
For Elevated access, we have 25 rules of 512 characters each.
For Academic access, we have 1000 rules of 1024 characters each.

The idea here is simple, we add one or many filtering rules, and we search for recent tweets that are related to these filters. If only one rule matches a tweet, it will be returned. To manipulate these endpoints with tweepy, we will use the tweepy.StreamingClient class.

To add, read and delete filters, we can write:

import os

import tweepy

client = tweepy.StreamingClient(os.getenv('BEARER_TOKEN'))
rules = [
    # we add our rules here
    tweepy.StreamRule(
        '("black panther" OR #wakandaforever) (magnificent OR amazing OR excellent OR awesome OR great) -is:retweet',
        tag='black panther tribute'
    )
]
client.add_rules(rules)
# we list our rules
response = client.get_rules()
for rule in response.data:
    print(rule)
# we delete one or more routes by passing their id
client.delete_rules(['158939726852798054'])

Notes:

You can pass the argument dry_run=True to the methods add_rules and delete_rules to test the syntax of the rule to make sure it is correct without actually executing it on the server side.
When defining several rules, it is recommended to associate a tag to remember what the filter does. Indeed, a filter can be quite complex to read :)

Once we have created our rules, all we have to do is call the endpoint to list the tweets. Normally the method to use is listen except that if we call it as is, we won't see anything because by default StreamingClient does nothing with the tweets it retrieves 😆. So you have to inherit from the class and override some of these methods.

import os

import tweepy

class IDPrinter(tweepy.StreamingClient):
    # we can get a response object
    def on_response(self, response):
        # It has the structure: StreamResponse(tweet, includes, errors, matching_rules)
        # So for each tweet, we have all the matching_rules
        print(response)
    # or we can just read the tweet
    def on_tweet(self, tweet):
        print(tweet.id, tweet.text)
    def on_errors(self, errors):
        print(errors)
    def on_connection_error(self):
        # what to do in case of network error
        self.disconnect()
    def on_request_error(self, status_code):
        # what to do when the HTTP response status code is >= 400
        pass

printer = IDPrinter(os.getenv('BEARER_TOKEN'))
printer.filter()

Notes:

By overloading on_response, we have as a parameter a StreamingResponse object that contains the tweet and the set of rules that matched this tweet.
You can pass the argument threaded=True to filter to avoid blocking the entire python script and get the created thread to close it later.
There is an asynchronous version of this streaming class AsyncStreamingClient that allows you to handle coroutines instead of threads. I won't talk about it here as it is an advanced topic. For the brave ones, you can read this article. There is another method of the StreamingClient class, namely sample which is linked to this endpoint. It does not take into account the filters created and returns 1% of the new tweets registered on the Twitter platform. You can't get more real-time than that 😅. This would allow you, for example, with natural language processing algorithms to detect Twitter trends like the ones we see displayed on the right of the web interface. Be careful though when using this endpoint, it can quickly end up your limit of tweets to read per month. You should have a clear objective before using it and only do it in a short period of time.

Ok, if you made it this far, congratulations! That was a big chunk but necessary to understand how the Twitter API works. In the second part of the article (because this one is getting very long) we will actually see how:

Send an email with tweets found in a recent search.
Retweet instantly all Wakanda Forever tribute tweets.
And a little bonus that I won't mention here 😉.

Take care of yourself and see you next time! 😁

This article was originally published on Medium.

If you like my article and want to continue learning with me, don't hesitate to follow me here and subscribe to my newsletter on substack 😉

Auditing your python environment

Kevin Tewouda — Thu, 18 Aug 2022 14:37:08 +0000

We have more and more hackers trying to steal information from users using malicious packages. Therefore, it is important to check frequently the dependencies of your project to know if vulnerabilities have been found. In this short tutorial, I will introduce you to two tools to help you audit your dependencies.

safety

The first one is safety. It is maintained by the pyup team which uses their custom safety
database. It comes in two flavors:

An open source version that we can freely use. It is updated once a month.
A paid version that is updated frequently (more than once a month).

installation

To install the package, type the following command in your terminal:

$ pip install safety

or if you are using a more advanced python package installer like poetry:

$ poetry add -D safety

If you don't want to install it directly in your project, and you are using GitHub, there is an official GitHub action you can use in your CI.

 - uses: pyupio/safety@v1
     # if you subscribed to a paid plan, you can insert your key like the following snippet
     # it assumes that you have a secret key called SAFETY_API_KEY and define under
     # Settings -> Secrets -> Actions in GitHub admin.
     with:
       api-key: ${{ secrets.SAFETY_API_KEY }}

usage

To test safety, we will install flask with version 0.5.

$ pip install flask==0.5
# or
$ poetry add flask==0.5

The usage is pretty straightforward. To scan vulnerabilities in your current python environment, you type:

$ safety check

You will see an output like the following:

...
 REPORT 

  Safety v2.1.1 is scanning for Vulnerabilities...
  Scanning dependencies in your environment:

  -> /home/kevin/.cache/pypoetry/virtualenvs/orm-L9juRWWT-py3.8/lib/python3.8/site-packages

  Using non-commercial database
  Found and scanned 64 packages
  Timestamp 2022-08-17 22:52:00
  3 vulnerabilities found
  0 vulnerabilities ignored

+=======================================================================================================================+
 VULNERABILITIES FOUND 
+=======================================================================================================================+

-> Vulnerability found in flask version 0.5
   Vulnerability ID: 38654
   Affected spec: <0.12.3
   ADVISORY: Flask 0.12.3 includes a fix for CVE-2019-1010083: Unexpected memory usage. The impact is denial of service. 
   The attack vector is crafted encoded JSON data.
   NOTE: this may overlap CVE-2018-1000656.https://github.com/pallets/flask/pull/2695/commits/0e1e9a04aaf29ab78f721cfc79ac2a691f6e3929
   CVE-2019-1010083
   For more information, please visit https://pyup.io/vulnerabilities/CVE-2019-1010083/38654/


 Scan was completed. 3 vulnerabilities were found. 
 ...
+=======================================================================================================================+
   REMEDIATIONS

  3 vulnerabilities were found in 1 package. For detailed remediation & fix recommendations, upgrade to a commercial license. 
...

If you want a more detailed output, you can run the previous command with the flag --full-report:

$ safety check --full-report

Note: as you can see in the output, it is only when you subscribed to a paid plan that you have suggestions on how to fix this vulnerability issue. This is not the case with the second tool I will introduce.
If you just want to see the vulnerable packages without further information, you can run:

$ safety check --bare
flask

You can also scan packages through a requirements file.

$ safety check -r requirements.txt

Or scan a single package.

$ echo "package-to-check==0.1" | safety check --stdin

By default, the output is in text format, but you can change it in JSON if you want with the --output option:

$ safety check --output json

pip-audit

The second tool I want to introduce to you is pip-audit. It is maintained by folks at Trails of Bit with some Google support. It uses the Pypa Advisory Database via the PyPI JSON API as a source of vulnerability reports.

installation

You can install it using pip or poetry like the following:

$ pip install pip-audit
# or
$ poetry add -D pip-audit

If you are using GitHub, there is a GitHub action you can use:

jobs:
  pip-audit:
    steps:
      - uses: trailofbits/gh-action-pip-audit@v0.0.4
        # if you use a requirements file, you can add the following
        with:
          inputs: requirements.txt

More information about how to configure it can be found here.

You can also use a pre-commit hook, although I will not recommend it since it will always trigger a network request and reduce the developer experience.

- repo: https://github.com/trailofbits/pip-audit
    rev: v2.4.3
    hooks:
      - id: pip-audit
        args: [ "-r", "requirements.txt" ]

ci:
  # Leave pip-audit to only run locally and not in CI
  # pre-commit.ci does not allow network calls
  skip: [ pip-audit ]

usage

We will still consider having the flask package with version 0.5 in our environment. To run pip-audit in our current python environment, we can type:

$ pip-audit

We will have an output like the following:

Found 2 known vulnerabilities in 1 package
Name  Version ID             Fix Versions
----- ------- -------------- ------------
flask 0.5     PYSEC-2019-179 1.0
flask 0.5     PYSEC-2018-66  0.12.3

For a more detailed output use the --desc flag:

$ pip-audit --desc

If you want to fix the issue, you can use the --fix flag. The command will install the minimal version fixing the issue.
Bear in mind that this workflow only works if you are using pip as your package dependency manager. If you are using another tool like pipenv or poetry, you probably want to see the fixed version without installing it with pip otherwise you will break your dependencies workflow. To achieve that, you must append
the --dry-run flag to the --fix one.

$ pip-audit --fix --dry-run
INFO:pip_audit._cli:Dry run: would have upgraded Flask to 1.0
Found 2 known vulnerabilities in 1 package and fixed 0 vulnerabilities in 0 packages
Name  Version ID             Fix Versions
----- ------- -------------- ------------
flask 0.5     PYSEC-2019-179 1.0
flask 0.5     PYSEC-2018-66  0.12.3

Now we see that the minimum version to solve the flask issue is 1.0. If you are using poetry, you need to change the version in pyproject.toml file with something like ^1.0 and run the following command:

$ poetry update flask

Now we can sleep well 😁.

pip-audit can also scan a requirements file:

$ pip-audit -r ./requirements.txt

If you are not in a virtual environment and use the pyproject.toml file to list
your dependencies, you can run:

# this assumes pyproject.toml is in the current directory
$ pip-audit .

By default, the output is in columns format, you can change it with the -f option. There are many supported formats like markdown, json, etc... To know more about the different formats supported you can type pip-audit --help in your
terminal. Example of the use of the json format:

$ pip-audit -f json | python -m json.tool

which audit tool to use in your project?

The answer to this question depends on your needs and constraints. Here are some considerations to take into account when making your decision.

safety has a company behind it that dedicates its time finding vulnerabilities in python packages ahead of official CVE.
To make the best use of safety i.e. have the best up-to-date vulnerability database, you should use a paid plan. You need to be sure you or your company has the budget for that.
pip-audit is more open source than safety since the management of the database is transparent for all users.
pip-audit provides a hint to fix a found vulnerability, safety doesn't do that unless you subscribed to a paid plan.
pip-audit may be integrated into pip in the future.

This is all for this tutorial, hope you enjoyed it. Take care of yourself and see you next time! 😁

If you like my article and want to continue learning with me, don’t hesitate to follow me here and subscribe to my newsletter on substack 😉

Alternatives to SQLAlchemy for your project - Prisma case

Kevin Tewouda — Mon, 08 Aug 2022 17:34:00 +0000

Although I use a lot SQLAlchemy in my daily work, I'm not satisfied by its api that I found a little difficult compared to the Django ORM with which I started my python journey! Django ORM is probably the best ORM IMHO in the python ecosystem. Unfortunately it is tight to the Django project and cannot be used elsewhere. So I decided to look for SQLAlchemy alternatives and starting with this article, I will present you the ones I liked the most.

For data engineers / data scientists this library can be particularly useful for you because it has a cleaner api to fetch data from databases.

Introduction

Prisma is a recent TypeScript ORM taking a different approach from its predecessors and focusing on type safety thanks to TypeScript.

But don't worry in this tutorial we will not talk about TypeScript, I will present you the unofficial python client created for this project. To present it briefly, I will borrow his words:

Prisma Client Python is a next-generation ORM built on top of Prisma that has been designed from the ground up for ease of use and correctness.

Installation

To install it you can use pip or poetry with the following command:

$ pip install prisma
# or
$ poetry add prisma

The prisma python client comes with a CLI that actually embeds the official prisma CLI with some additional commands. To be sure it is installed, type the following in your shell:

$ prisma -h

◭  Prisma is a modern DB toolkit to query, migrate and model your database (https://prisma.io)

Usage

  $ prisma [command]

Commands

            init   Setup Prisma for your app
        generate   Generate artifacts (e.g. Prisma Client)
              db   Manage your database schema and lifecycle
         migrate   Migrate your database
          studio   Browse your data with Prisma Studio
          format   Format your schema
...

Note: you may notice that it downloads some binaries when you first invoke this command. This is normal it fetches the node prisma cli and engines used by prisma. 😁

Prisma schema file

In Prima, it all starts with a schema.prisma file where you define your business models and relations between them. After that you will invoke a prisma CLI command to generate the client. This is different from ORMs like Django ORM and SQLAlchemy which use the Active Record
pattern where we define model classes with data and methods to interact with the database. Here Prisma (at least the python library) is in charge of defining the api to manipulate the database as well as the different data models that will be involved in the operations.

So let's start with this schema (saved it in a file called schema.prisma):

datasource db {
  provider = "sqlite"
  url      = "file:db.sqlite"
}

generator client {
  provider  = "prisma-client-py"
}

model User {
  id         String   @id @default(uuid())
  firstname  String
  lastname   String
  is_admin   Boolean  @default(false)
  email      String   @unique
  created_at DateTime @default(now())
  updated_at DateTime @updatedAt
  posts      Post[] // not represented in the database

  @@map("user")
}

model Post {
  id         String   @id @default(uuid())
  title      String
  content    String
  published  Boolean  @default(false)
  created_at DateTime @default(now())
  updated_at DateTime @updatedAt
  // user field will not be represented in the database
  // To link a post to its author we precise which field of this table
  // is a foreign key (user_id) and on which field of the other table
  // it points (id)
  user       User     @relation(fields: [user_id], references: [id])
  user_id    String

  @@map("post")
}

Ok, let's break down what we have.

datasource

In this section, we define how to connect to a database. We provide:

a provider: in our case it is sqlite but prisma supports MySQL, MariaDB, PostgreSQL, SQL Server, CockroachDB and MongoDB.
an url: the connection string with all the information necessary to connect. For sqlite it is quite simple, we pass a filename path prefixed with file:. For other databases, refer to prisma documentation.

Note: At the moment of writing this article CockroachDB is not supported by the python client, but it should not take long.

Also, you can use environment variables to not reveal secrets in your prisma schema file. For example, you can replace the url value in the previous example with env("DATABASE_URL"). The prisma client will know that it needs to fetch database connection string from an environment variable called DATABASE_URL.

generator

In this section, we specify the script that will generate a python client we can use for our queries.
For our python library, the value is prisma-client-py but you can create a custom generator if you want.
Warning! It is an advanced topic. 🙂

model

And we finally have the model definition section. In our example, we defined two models User and Post. These two models will represent two tables in the database db.sqlite and the attributes defined in each of them will
represent table columns.
To know all the types available in prisma, refer to this
section of the official documentation. There are not many of them.

Some notes:

To define a primary field, we use the attribute @id and we ensure that the value is filled with an uuid using the attribute @default combined with the function uuid()
In the User model, the is_admin boolean field is assigned a default value of false. This means that every user created will not be an admin. The other value possible for this boolean field is true of course.
In the User model, the email field is assigned a @unique attribute which means that we can't have the same email twice in the user table.
In User and Post models, the created_at date time field is assigned a default attribute with function now() which means that each time we create a record, the field will be automatically filled with the current timestamp.
In User and Post models, the updated_at date time field is assigned an @updatedAt attribute which means each time we update a record, the field will be automatically filled with the current timestamp.
We defined a 1-n relation between Post and User models which is represented by the user and posts fields. These fields are not represented in the database but at prisma level to easily manage relations.
The @@map model attribute is used to define the name of the table in the database. If we don't define it, by default prisma will use the model name.

Basic Usage

Now that we have a schema file we can create the database and generate the client with this command:

$ prisma db push

Normally, you should have an output similar to the following:

Prisma schema loaded from schema.prisma
Datasource "db": SQLite database "db.sqlite" at "file:db.sqlite"

SQLite database db.sqlite created at file:db.sqlite
...
✔ Generated Prisma Client Python (v0.6.6) to ...

Create a python file with the following content. This is the minimum to use the python client.

import asyncio
from prisma import Prisma


async def main() -> None:
    db = Prisma()
    await db.connect()
    await db.user.find_first()

    await db.disconnect()


if __name__ == '__main__':
    asyncio.run(main())

If you run it, it should return nothing since there is no record in the database but at least everything is fine :)

Well, if you don't want the async interface, for example you want to adopt it in a synchronous web framework, or you are a data engineer / data scientist who is not used to work in this paradigm, you can generate the client with a synchronous interface. First, you must modify the generator section of the schema file to insert the interface information. It will look like the following:

// I just put the interesting part, not the whole file
generator client {
  provider  = "prisma-client-py"
  interface = "sync"
}

Now re-generate the client with:

$ prisma db push

And you should be able to run the following script:

from prisma import Prisma


def main() -> None:
    db = Prisma()
    db.connect()
    db.user.find_first()

    db.disconnect()


if __name__ == '__main__':
    main()

We are good now! In the following examples I will use asynchronous examples, but all you have to do to make it work
with your synchronous client is to remove async / await keywords. 😉

Ok, let's create some users and posts.

import asyncio

from prisma import Prisma


async def main() -> None:
    db = Prisma()
    await db.connect()
    user_1 = await db.user.create(
        data={
            'firstname': 'Kevin',
            'lastname': 'Bogard',
            'email': 'kevin@bogard.com',
            'posts': {
                'create': [
                    {
                        'title': 'Prisma is awesome',
                        'content': 'This is the truth!'
                    },
                    {
                        'title': 'Do you know the Bogard family?',
                        'content': "If not, you probably don't know King of Fighters!"
                    }
                ]
            }
        }
    )
    user_2 = await db.user.create(
        data={
            'firstname': 'Rolland',
            'lastname': 'Beaugosse',
            'email': 'rolland@beaugosse.fr',
            'posts': {
                'create': [
                    {
                        'title': 'Prisma is awesome',
                        'content': 'you should use it!'
                    },
                    {
                        'title': 'What other ORM do you use?',
                        'content': 'Tell me in the comments'
                    }
                ]
            }
        }
    )
    print(user_1)
    print(user_2.json(indent=2))

    await db.disconnect()


if __name__ == '__main__':
    asyncio.run(main())

You notice that we can create a user and its related posts in one query if we want. The query is straightforward thanks to prisma which manages the relations in a implicit way for us.

The result of this query is the user model created, it is a pydantic model, so
you can use its methods to introspect the data. if you don't know pydantic, it is a powerful library for data validation. I have a tutorial on some of its features.

Hidden powers of pydantic. Discover some key features of pydantic… | by Kevin Tewouda | Medium

Kevin Tewouda ・ Dec 27, 2022 ・
lewoudar.Medium

If you are curious you can look the definition of the generated models in the module prisma.models.

Note: if you don't target sqlite, you can create multiple objects in a batch like this

# not possible in sqlite
users = await db.user.create_many(
    data=[
        {'firstname': 'Kevin', 'lastname': 'Bogard', 'email': 'kevin@bogard.com'},
        {'firstname': 'Rolland', 'lastname': 'Beaugosse', 'email': 'rolland@beaugosse.com'},
    ]
)

Now let's see how to query objects with prisma and the python client.

# I don't put all the imports, you have them in the previous example :)
from datetime import datetime, timedelta

# returns the first user of the database
user = await db.user.find_first()
print(user.json(indent=2))

# returns the user where email is rolland@beaugosse.com and include post information
user = await db.user.find_unique(
    where={'email': 'rolland@beaugosse.fr'},
    include={'posts': True}
)
print(user.json(indent=2))

# lists posts from the first user
posts = await db.post.find_many(
    where={
        'user': {
            'is': {
                'email': 'kevin@bogard.com',
            }
        }
    }
)
for post in posts:
    print(post.json(indent=2))

# find posts
# where title contains "Prisma" or creation date is less than 1 hour
# order by creation date descendant
# and include user information
posts = await db.post.find_many(
    where={
        'OR': [
            {
                'title': {
                    'contains': 'Prisma'
                }
            },
            {
                'created_at': {
                    'gt': datetime.utcnow() - timedelta(hours=1)
                }
            }
        ]
    },
    include={'user': True},
    order={'created_at': 'desc'}
)
for post in posts:
    print(post.json(indent=2))

The queries made with prisma are really simple to follow and powerful. To know more about what you can do, check the prisma client reference and the official documentation.

Introspection

What if you already have a database in place and want to take advantage of prisma? Prisma got your back with a command which introspects the database and generate the models in your schema file. Concretely, you have to define a base file with the datasource and generator sections like this:

datasource db {
  provider = "sqlite"
  // replace the url with correct one in your case
  url      = "file:db.sqlite"
}

generator client {
  provider  = "prisma-client-py"
}

and run the command:

$ prisma db pull

Prisma will automatically populate the models with the metadata retrieved from the database. Of course, it is not
perfect, and you will probably have to adjust some information but it is a good start. More information on prisma introspection can be found in this section of the official documentation.

migrations

Another great feature of prisma is its migration system. This allows for a smoother database experience allowing us to make incremental changes in our web application for example. When you create / update / delete models, you can create a migration file that you will apply later on your production database. For example, since we have created two models so far, we can create a migration file with the following command:

$ prisma migrate dev --name "create user and post models"

Note: you should notice that the CLI asks you the permission to continue because it will erase all the data in your local
database. This is normal since it is our first migration and it is only done on our local database.

Now we have a folder migrations with sub-folders containing your different migration files. Once we have done all
the necessary tests and are sure about our changes, we can apply it on our production database with the following command:

# be sure the url in the datasource section of your prisma file
# has the correct value
$ prisma migrate deploy

Note: unlike other migration systems like Django migrations or alembic, you don't have a system to upgrade or downgrade the database.
You always update the database with a new migration. So if you want to roll back a migration, you must create a new migration file.

Now re-create some users and posts (you can use our first example). And let's say we want now to count the number of times a post is viewed, so we add this in our model:

...
model Post {
  id         String   @id @default(uuid())
  title      String
  content    String
  published  Boolean  @default(false)
  // we have a new field to count the number of times
  // a post is viewed
  views      Int      @default(0)
  ...
}

You can create a new migration with prisma migrate:

$ prisma migrate dev --name "add views field in post model"

Note: the database is not erased this time because we already have a migration in place. 😉

More information about migrations can be found in this section of the official documentation.

relations

Probably the best feature for me in prisma is that we can easily query deeply nested and linked models thanks to how it handles relations in the background for us. In this regard, it supports the following relations:

1-n (one-to-many) relation
1-1 (one-to-one) relation
n-m (many-to-many) relation
self-relations where a field model references itself

if these terms do not speak to you (the first three especially), I invite you to read this article.

We have already seen the one-to-many relation where a user has 0 to many posts and a post has only one user. Now let's see how to define one-to-one and many-to-many relations.

one-to-one relation

Let's say we want to extend the user information in a profile table, we will add the following model in the schema and also update the User model.

model User {
 ...
 // we add an optional profile field
  profile    Profile?

  @@map("user")
}

model Profile {
  id      String @id @default(uuid())
  bio     String
  user    User   @relation(fields: [user_id], references: [id])
  // the @unique attribute is what make the difference between a
  // 1-n relation and a 1-1 relation
  user_id String @unique

  @@map("profile")
}

It is not really different from how we define the user<->post relation except for two things:

the profile field in the User model has a ? sign next to the type because we want it to be optional. This is because we already have some users in the database, so if this field is required, it will be possible to update the database. Of course, if it is a new project, you can make it mandatory if you want. I just want to say that it is not related to the definition of a one-to-one relation.
in the Profile model, the user_id field is marked as @unique. If you look the Post model and how we define the same field, you will notice that it is the only difference, this attribute is what creates the one-to-one relation between User and Profile.

Now you can upgrade the database with the command:

$ prisma migrate dev --name "add profile model"

And starts to play with it:

user = await db.user.find_first()
profile = await db.profile.create(
    data={
        'bio': "I'm cool",
        'user': {
            # we link the profile to its user
            'connect': {
                'id': user.id
            }
        }
    }
)
print(profile.json(indent=2))

# we get the same profile
profile = await db.profile.find_first(
    where={
        'user': {
            'is': {
                'email': user.email
            }
        }
    },
    include={'user': True}
)
print(profile.json(indent=2))

many-to-many relation

Now, let's say we want to attach some categories to a post, we can update the schema to something like this:

model Post {
  ...
  categories Category[] @relation(references: [id])

  @@map("post")
}

model Category {
  id    String @id @default(uuid())
  name  String
  posts Post[] @relation(references: [id])

  @@map("category")
}

And migrate the database with the following command:

$ prisma migrate dev --name "add category model"

If you look at the migration file, you will notice that prisma creates a glue table to handle this relation. Now we can do the following stuff thanks to prisma.

# we create a user, post and category together!
user = await db.user.create(
    data={
        'firstname': 'foo',
        'lastname': 'bar',
        'email': 'foo@bar.com',
        'posts': {
            'create': [
                {
                    'title': 'Prisma is so good',
                    'content': 'oh yeah!',
                    'categories': {
                        'create': [
                            {'name': 'prisma'},
                            {'name': 'orm'}
                        ]
                    }
                }
            ]
        }
    }
)
print(user.json(indent=2))

# we fetch the categories
categories = await db.category.find_many(
    where={
        'posts': {
            'every': {
                'title': {
                    'contains': 'Prisma'
                }
            }
        }
    },
    include={'posts': True}
)
for category in categories:
    print(category.json(indent=2))

Sometimes, you want to add some attributes in an n-m relation, in that case you have to manually create the intermediate table. For more information, see this section of the official documentation.

I won't explain the self-relation either because this tutorial is already long! Feel free to learn more in this section of the documentation.

cli

I want to summarize some commands you will often use when working with prisma.

prisma format: Like the name suggests, it formats the schema file and also validates its content.
prisma db pull: Fetch models from the database and generate the client.
prisma generate: Generates the client, useful in a pull strategy where you want to modify models without pushing the changes in the database.
prisma db push: When you are testing an application, this is the command you should use to reset the database and generate the client.
prisma migrate dev --name <name>: To use after db push when you are sure of your changes. It will create a migration file, apply it on your local environment and generate the client.
prisma migrate deploy: Apply a migration file on a production environment.
prisma py version: Show debugging information about the python client, prisma version supported, binaries, etc...

Summary

Ok, at the end of this tutorial, I will list the advantages and drawbacks I see with prisma.

advantages

Simple and neat api to CRUD data.
Auto-completion feature to write queries faster on editors supporting the Language Server Protocol and pyright like VS Code. Unfortunately, Pycharm has limited support for TypedDict and it doesn't work for now.
Migration system.
Introspection feature to generate models and clients by reading metadata on the database (this is really cool!).
Support many relational databases, even a relatively new one I just discovered while learning prisma, CockroachDB. It also supports MongoDB which is a NoSQL database.

drawbacks

Even if it supports many relational databases, there is one notable absence in the list: Oracle. It can be restrictive if you are working for a bank or a large corporation that is used to working with this database. There is an ongoing issue if you want to follow this topic.
We can't compare two fields of the same table, something we can do with the F expressions in Django. There is an ongoing issue if you want to follow this topic. The workaround is to use raw queries.
There are some inconsistencies in database support where some fields are implemented in some databases but not others like the JSON field which is not implemented on sqlite (there is an ongoing issue here) and even where it is implemented, the way to query differs from one database to another. This is not what I expect from an ORM. It should be database agnostic.
We don’t have mixins support for models to avoid repeating fields between models like datetime fields (created_at, updated_at).

Anyway, prisma is a relatively new project, and it is very pleasant to use. I encourage you to give it a try and support the python client (at least a star on GitHub).
For me who doesn't like SQL, being able to retrieve models from the database and make queries with prisma is a breath of fresh air. 🤣

This is all for this tutorial, stay tuned for the next ORM we will explore together! 😉

If you like my article and want to continue learning with me, don’t hesitate to follow me here and subscribe to my newsletter on substack 😉

Brief history of data classes in python

Kevin Tewouda — Fri, 22 Jul 2022 07:54:54 +0000

In this tutorial, we will review the different ways we create data classes in python from the oldest way to the newer. Hopefully at the end, you will be convinced to use pydantic dataclasses as your default way to create data
classes.

Default python class creation

The oldest way to define a class in python is via the special __init__ method. For our example we will work with a Point class taking x and y coordinates as input.

class Point:
    def __init__(self, x: int, y: int):
        self.x = x
        self.y = y


p1 = Point(1, 2)
print(p1.x)  # will print 1
print(p1)  # will print something like <__main__.Point object at 0x7fc5d4283c90>

p2 = Point(1, 2)
print(p1 == p2)  # will print False

Like you can see, there is some boilerplate code because we define the same variables as method arguments and class attributes (x and y). When we try to print our created object to see how it looks like, we see a strange default representation made by python.
Even worse, we can instantiate an object like the following Point(1, 'foo') and python will not complaint at all.
Ok you can say that tools like mypy or pyright will help you catch the bug but not everybody wants to use them, so we have to find another way.
Also, the default class implementation does not have comparison methods implemented, therefore p == p2 returns False even if the attributes have the same values between the two objects. 🥲
Here is how we can fix these three issues with the following code.

class Point:
    def __init__(self, x: int, y: int):
        for item in [x, y]:
            if not isinstance(item, int):
                raise ValueError(f'{item} is not an integer')
        self.x = x
        self.y = y

    def __repr__(self):
        return f'Point(x={self.x}, y={self.y})'

    def __eq__(self, other):
        if other.__class__ is self.__class__:
            return (self.x, self.y) == (other.x, other.y)
        return NotImplemented

    # we take the opportunity to write the opposite method
    def __ne__(self, other):
        result = self.__eq__(other)
        if result is NotImplemented:
            return NotImplemented
        return not result


p1 = Point(1, 2)
print(p1)  # will print Point(x=1, y=2)

p2 = Point(1, 2)
print(p1 == p2)  # will print True
Point(1, 'foo')  # will raise ValueError

Now we have a working class with a pretty representation, but look the amount of code we have to write... And we haven't even implemented all the comparison methods.

Namedtuples

I will make a small digression on
the namedtuples because one car argue that it is a way to declare a class quickly, and yes it is the case but not without some caveats...
Consider the following example:

from collections import namedtuple

Point = namedtuple('Point', ['x', 'y'])
p1 = Point(1, 2)
print(p1)  # will print Point(x=1, y=2)

p2 = Point(1, 'foo')  # we can still mess with variable type

print(p1 == (1, 2))  # will print True :(

Ok, with namedtuples, we have a pretty representation by default, but we suffer for the same lack of type verification like in the default way of creating classes and since it inherits the tuple class, the comparison with a tuple in the last instruction is correct which is not always what we will want.

attrs

Taking into account the problems related to class creation quoted above, a well known pythonista decides to bring a solution with a library called attrs. Let's see how we can re-write our Point class.

from attrs import define, field, validators


@define
class Point:
    x: int = field(validator=[validators.instance_of(int)])
    y: int = field(validator=[validators.instance_of(int)])


p1 = Point(1, 2)
print(p1)  # will print Point(x=1, y=2)

p2 = Point(1, 2)
print(p1 == p2)  # will print True

print(p1 == (1, 2))  # will print False

p = Point(1, 'foo')  # will raise a TypeError

Ok, we clearly see a difference with the handwritten class we wrote above. We have:

A pretty default representation
Type verification using the field function and validators.
A default comparison implementation taking in account the type of the compared objects. This is why the test with a tuple returns False.

It is a well-thought library, and it can be customized in different ways like defining slots, frozen classes, keyword-only arguments and more.

dataclasses

In python3.7, the language introduces dataclasses defined in PEP 557 with the will to simplify the writing of classes. In fact, this new standard library is heavily inspired by attrs. Let's see what our famous Point class look like:

from dataclasses import dataclass


@dataclass
class Point:
    x: int
    y: int


p1 = Point(1, 2)

p2 = Point(1, 2)
print(p1 == p2)  # will print True

print(p1 == (1, 2))  # will print False

p = Point(1, 'foo')  # will not raise a TypeError :(

We have almost the same advantages as the attrs definition except that the argument type is not verified at initialization time.
The only way to achieve this verification is to do the following:

from dataclasses import dataclass


@dataclass
class Point:
    x: int
    y: int

    def __post_init__(self):
        if not isinstance(self.x, int):
            raise TypeError('x is not an integer')
        if not isinstance(self.y, int):
            raise TypeError('y is not an integer')


Point(1, 'foo')  # will raise a TypeError

Yeah, it sucks a little, but it was a will of the CPython maintainers to have a simplified version of attrs without validation and other joys.

pydantic dataclasses

Finally, we will talk about pydantic, a data validation library made famous by
a relatively young web framework, FastAPI. If you don't know it, I highly recommend to check its api, it is another well-though piece of software. I also wrote a blog post presenting some of its advantages.

The feature that interest us in this article is that of the dataclasses. Again we will look at our Point class implementation. 😁

from pydantic.dataclasses import dataclass


@dataclass
class Point:
    x: int
    y: int


p1 = Point(1, 2)
print(p1)  # will print Point(x=1, y=2)

p2 = Point(1, 2)
print(p1 == p2)  # will print True

print(p1 == (1, 2))  # will print False

p = Point(1, 'foo')  # will raise a pydantic.ValidationError

We have all the advantages that we had with the attrs implementation and the written code is even smaller!
pydantic leverages type annotations to validate data, and we can still use the api provided by the standard library dataclasses like field, asdict, etc... because pydantic.dataclasses is just a wrapper around the standard one. For proof, look at the following example:

import dataclasses
import pydantic.dataclasses


@dataclasses.dataclass
class Point:
    x: int
    y: int


# no error raised
print(Point('foo', 2))

Point = pydantic.dataclasses.dataclass(Point)

# error raised
print(Point('foo', 2))

Here we wrap a normal dataclass into a pydantic one, and we have the same verifications and features!
Pydantic is a fantastic library (yes I am a little biased) and I can only recommend you to check out its documentation.

This is all for this tutorial, hope you enjoyed it. Take care of yourself and see you next time! 😁

If you like my article and want to continue learning with me, don’t hesitate to follow me here and subscribe to my newsletter on substack 😉

websockets-cli: a friendly companion to your web project

Kevin Tewouda — Mon, 18 Jul 2022 06:55:55 +0000

In this article, I will present my last open source project I have been working on for the last few months, websockets-cli. It is a Command Line Interface (CLI) to interact with a websocket server.

Why?

There are two main reasons for this project:

Each time I work on a web project involving websockets, I found myself wanting a simple CLI tool to test what I have coded. I always ended up writing a script to solve this issue because I could not find an appropriate library.
It is a fun project to do 🙃

Of course, there are graphical solutions like Postman to help you with websockets, but I'm sure I'm not the only one who prefer to work with the terminal for such testing.

Installation

To install it, you will need python3.7 or newer and pip installed. After that, you can type:

$ pip install websockets-cli

You can also use another package manager like poetry. I have
a tutorial on it if you don't know this tool.

$ poetry add -D websockets-cli

If you want some kind of global installation, you can use the pipx library.

$ pipx install websockets-cli

Settings

There are many settings you can configure for your needs. There are all defined in this section of the official documentation. You can configure them either using the:

pyproject.toml file
environment variables
a custom env file named .ws.env in your current working directory or home directory.

Note:

the pyproject.toml file has precedence over environment variables.

Usage

The CLI is straightforward to use. To know all available commands, just type ws in the terminal:

$ ws
Usage: ws [OPTIONS] COMMAND [ARGS]...

  A convenient websocket cli.

  Example usage:

  # listens incoming messages from endpoint ws://localhost:8000/path
  $ ws listen ws://localhost:8000/path

  # sends text "hello world" in a text frame
  $ ws text wss://ws.postman-echo.com/raw "hello world"

  # sends the content from json file "hello.json" in a binary frame
  $ ws byte wss://ws.postman-echo.com/raw file@hello.json

Options:
  --version   Show the version and exit.
  -h, --help  Show this message and exit.

Commands:
  byte                Sends binary message to URL endpoint.
  echo-server         Runs an echo websocket server.
  install-completion  Install completion script for bash, zsh and fish...
  listen              Listens messages on a given URL.
  ping                Pings a websocket server located at URL.
  pong                Sends a pong to websocket server located at URL.
  session             Opens an interactive session to communicate with...
  tail                An emulator of the tail unix command that output...
  text                Sends text message on URL endpoint.

Install-completion

The first command to use is install-completion which will add completion support for commands and options. It will work for bash, fish and zsh. Windows support (Powershell) is planned.

$ ws install-completion
# when the command succeeded, you should see the following message
Successfully installed completion script!

Echo-server

If you want to self-host an echo websocket server like Postman does, you can run the echo-server command. Here is an example:

$ ws echo-server -H ::1 -p 8000
Running server on ::1:8000 💫
# To stop the server, you can just tap Ctrl+C
^CProgram was interrupted by Ctrl+C, good bye! 👋

Listen

Often, you will want to see incoming messages from a websocket endpoint. It is really easy with websockets-cli:

# assuming you have an endpoint localhost:8000 sending messages
$ ws listen :8000
# you will have an output like the following
──────────────────── TEXT message on 2022-05-25 07:10:07 ────────────────────────────────────
{"hello": "world"}
──────────────────── BINARY message on 2022-05-25 07:10:07 ──────────────────────────────────────
b'{"hello": "world"}'

Session

Probably the command you will use the most. It allows a complete interaction with a websocket server. You can send ping, pong, close, text and binary frames.

$ ws session wss://ws.postman-echo.com/raw
Welcome to the interactive websocket session! 🌟
For more information about commands, type the help command.
When you see <> around a word, it means this argument is optional.
To know more about a particular command type help <command>.
To close the session, you can type Ctrl+D or the quit command.

> help
The session program lets you interact with a websocket endpoint with the
following commands:

• ping <message>: Sends a ping with an optional message.
• pong <message>: Sends a pong with an optional message.
• text message: Sends text message.
• byte message: Sends byte message.
• close <code> <reason>: Closes the websocket connection with an optional code
and message.
• quit: equivalent to close 1000.

> help close
Closes the session given a code and an optional reason.

Example usage:

If no code is given, 1000 is considered as default meaning a normal closure.
Thus, it is equivalent to the quit command.

┌────────────────────────────────────────────────────────────────────────────┐
│ > close                                                                    │
└────────────────────────────────────────────────────────────────────────────┘

Closes the connection with a code 1001 and no message.

┌────────────────────────────────────────────────────────────────────────────┐
│ > close 1001                                                               │
└────────────────────────────────────────────────────────────────────────────┘

Closes the connection with a code 1003 and a message "received unknown data".

The message length must not be greater than 123 bytes.

┌────────────────────────────────────────────────────────────────────────────┐
│ > close 1003 'received unknown data'                                       │
└────────────────────────────────────────────────────────────────────────────┘

To know more about close codes, please refer to the RFC.
> quit
Bye! 👋

Thanks to prompt-toolkit, the session command has features like:

Syntax highlighting
Auto-completion of subcommands
Ability to read history of the previous commands typed during the current session
Autosuggestion of commands based on your history within the current session.

I will stop here for the presentation, if you want to know more about the project, you can read the official documentation.
I hope you will find it interesting to use if you play a lot with websockets. 🙂

If you like my article and want to continue learning with me, don’t hesitate to follow me here and subscribe to my newsletter 😉

DEV Community: Kevin Tewouda

Create a Twitter bot in python - part 2

Weekly sending of search results by email

Retweet Wakanda tribute

Bonus: schedule the creation of a tweet at a later date

Create a Twitter bot in python - part 1

Prerequisites

Twitter API documentation

A bit of terminology

Observe a theme using keywords

Search recent tweets.

Build a search filter

Get historical

Polling

Filtered stream

Auditing your python environment

safety

installation

usage

pip-audit

installation

usage

which audit tool to use in your project?

Alternatives to SQLAlchemy for your project - Prisma case

Introduction

Installation

Prisma schema file

datasource

generator

model

Basic Usage

Hidden powers of pydantic. Discover some key features of pydantic… | by Kevin Tewouda | Medium

Kevin Tewouda ・ Dec 27, 2022 ・ lewoudar.Medium

Introspection

migrations

relations

one-to-one relation

many-to-many relation

cli

Summary

advantages

drawbacks

Brief history of data classes in python

Default python class creation

Namedtuples

attrs

dataclasses

pydantic dataclasses

websockets-cli: a friendly companion to your web project

Why?

Installation

Settings

Usage

Install-completion

Echo-server

Listen

Session

Kevin Tewouda ・ Dec 27, 2022 ・
lewoudar.Medium