PythonでTwitter API v2のエンドポイントを叩くライブラリを書いてみた #Python

注意事項

本記事では、Twitter APIを使用するための登録手順は説明しておりませんので、
ご了承ください。
また、v2になるため、Bearer Tokenが必要ですが、払い出しの手順も説明しておりません。

きっかけ

Tweepyでv2がまだ対応していなかったので、
そのライブラリを作ったらいいのかもしれないと思った。
暇だった。

対象のエンドポイント

Search Spaces
Spaces lookup

Search Spaces

概要

検索文字列と検索対象のSpacesの状態（ライブ中 or 予定）を設定して、
Spacesを検索する。

Endpoint URL

https://api.twitter.com/2/spaces/search

ドキュメント

Spaces lookup

概要

スペースのIDやユーザーIDを設定して、Spacesを検索する。

Endpoint URL

※:idは実際のspacesのIDを設定する

ドキュメント

実際に書いたコード

spacy.py

import requests
import json

class spacy:
    def __init__(self):
        self.search_params = {'query': '', 'state': ''}#, 'space.fields': '', 'expansions': ''}
        self.search_spaceids_params = {'ids': ''}#, 'space.fields': 'title,created_at', 'expansions': 'creator_id'}
        self.search_creatorid_params = {'user_ids': ''}#, 'space.fields': '', 'expansions': ''}
        self.headers = {'Authorization': ''}
        self.expansions = {'expansions': ''}
        self.spacefield = {'space.fields': ''}
        self.params = {}

        self.id = ''

    def create_headers(self, bearer_token):
        self.headers = {
            "Authorization": "Bearer {}".format(bearer_token)#,
        }

    def connect_to_search_space_endpoint(self):

        if self.search_params['query'] == '':
            raise Exception('[query] is required parameter.')

        if self.search_params['state'] == '':
            raise Exception('[state] is required parameter.')

        self.search_params.update(**self.expansions, **self.spacefield)

        search_url = "https://api.twitter.com/2/spaces/search"
        print(self.search_params)
        response = requests.request("GET", search_url, headers=self.headers, params=self.search_params)
        print(response.status_code)
        if response.status_code != 200:
            raise Exception(response.status_code, response.text)

        print(json.dumps(response.json(), indent=4, sort_keys=True))

    def connect_to_space_lookup_with_spaceids(self):

        if self.search_spaceids_params['ids'] == '':
            raise Exception('[ids] is required parameter.')

        self.params.update(**self.search_spaceids_params,**self.expansions, **self.spacefield)

        search_url = "https://api.twitter.com/2/spaces"
        response = requests.request("GET", search_url, headers=self.headers, params=self.params)
        print(response.status_code)
        if response.status_code != 200:
            raise Exception(response.status_code, response.text)

        print(json.dumps(response.json(), indent=4, sort_keys=True))

    def connect_to_space_lookup_with_spaceid(self):
        if self.space_id == '':
            raise Exception('[id] is required parameter')

        self.params.update(**self.spacefield, **self.expansions)

        search_url = "https://api.twitter.com/2/spaces/" + self.space_id
        response = requests.request("GET", search_url, headers=self.headers, params=self.params)
        print(self.params)
        print(response.status_code)

        if response.status_code != 200:
            raise Exception(response.status_code, response.text)

        print(json.dumps(response.json(), indent=4, sort_keys=True))

    def connect_to_space_lookup_with_creatorids(self):
        if self.search_creatorid_params == '':
            raise Exception('[user_ids] is required parameter')

        self.params.update(**self.search_creatorid_params, **self.spacefield, **self.expansions)

        search_url = "https://api.twitter.com/2/spaces/by/creator_ids"
        response = requests.request("GET", search_url, headers=self.headers, params=self.params)
        print(self.params)
        print(response.status_code)

        if response.status_code != 200:
            raise Exception(response.status_code, response.text)

        print(json.dumps(response.json(), indent=4, sort_keys=True))

    def create_params(self, *, user_ids='', ids='', id='', query='', state='', **kwargs):
        
        # search_spaces
        self.search_params['query'] = query
        self.search_params['state'] = state
        
        # space_lookup
        self.search_creatorid_params['user_ids'] = user_ids

        # space_lookup
        self.search_spaceids_params['ids'] = ids

        # space_lookup
        self.space_id = id
        
        if kwargs.get('expansions'):
            self.expansions['expansions'] = kwargs['expansions']
        
        if kwargs.get('fields'):
            self.spacefield['space.fields'] = kwargs['fields']

使い方

Pythonインタプリタを起動した後

Search Spacesの場合

import spacy
変数 = spacy.spacy()
変数.create_params(query='検索文字列', state='live or scheduled', expansions='creator_id')
変数.create_headers('Bearer Token')
変数.connect_to_search_space_endpoint()

Spaces lookupのユーザID検索の場合

import spacy
変数 = spacy.spacy()
変数.create_params(user_ids='ユーザID', expansions='creator_id')
変数.create_headers('Bearer Token')
変数.connect_to_space_lookup_with_creatorids()

Spaces lookupのスペースID（単体）検索の場合

import spacy
変数 = spacy.spacy()
変数.create_params(id='スペースID', expansions='creator_id')
変数.create_headers('Bearer Token')
変数.connect_to_space_lookup_with_spaceid()

Spaces lookupのスペースID（複数）検索の場合

import spacy
変数 = spacy.spacy()
変数.create_params(ids='スペースID1,スペースID2', expansions='creator_id')
変数.create_headers('Bearer Token')
変数.connect_to_space_lookup_with_spaceids()

編集後記

初めてライブラリを作ってみました。
エンドポイントを叩く関数が長くなってしまったので、短くしたかったのですが、
いい案が思いつかず長くなってしまった。
もしかしたら、今後アップデートして、jsonを返す形にするかなとか思ったり、
返却されたjsonをそれぞれ整形する関数を作ったりするかもしれないです。気分で。
もっとこうした方が良いや不具合あるぞとか、ありましたら、是非コメントください。
よろしくお願いします。

2022.04.02
githubにpushしました。
GitHub - madaragon/spacy