Change Python Requests to use auth= parameter instead of headers= to authenticate the request

Author: aureliaspecker

Filtered-Stream/filtered_stream.py

@@ -4,16 +4,22 @@
 
 # To set your enviornment variables in your terminal run the following line:
 # export 'BEARER_TOKEN'='<your_bearer_token>'
+bearer_token = os.environ.get("BEARER_TOKEN")
 
 
-def create_headers(bearer_token):
-    headers = {"Authorization": "Bearer {}".format(bearer_token)}
-    return headers
+def bearer_oauth(r):
+    """
+    Method required by bearer token authentication.
+    """
 
+    r.headers["Authorization"] = f"Bearer {bearer_token}"
+    r.headers["User-Agent"] = "v2FilteredStreamPython"
+    return r
 
-def get_rules(headers, bearer_token):
+
+def get_rules():
     response = requests.get(
-        "https://api.twitter.com/2/tweets/search/stream/rules", headers=headers
+        "https://api.twitter.com/2/tweets/search/stream/rules", auth=bearer_oauth
     )
     if response.status_code != 200:
         raise Exception(
@@ -23,15 +29,15 @@ def get_rules(headers, bearer_token):
     return response.json()
 
 
-def delete_all_rules(headers, bearer_token, rules):
+def delete_all_rules(rules):
     if rules is None or "data" not in rules:
         return None
 
     ids = list(map(lambda rule: rule["id"], rules["data"]))
     payload = {"delete": {"ids": ids}}
     response = requests.post(
         "https://api.twitter.com/2/tweets/search/stream/rules",
-        headers=headers,
+        auth=bearer_oauth,
         json=payload
     )
     if response.status_code != 200:
@@ -43,7 +49,7 @@ def delete_all_rules(headers, bearer_token, rules):
     print(json.dumps(response.json()))
 
 
-def set_rules(headers, delete, bearer_token):
+def set_rules(delete):
     # You can adjust the rules if needed
     sample_rules = [
         {"value": "dog has:images", "tag": "dog pictures"},
@@ -52,7 +58,7 @@ def set_rules(headers, delete, bearer_token):
     payload = {"add": sample_rules}
     response = requests.post(
         "https://api.twitter.com/2/tweets/search/stream/rules",
-        headers=headers,
+        auth=bearer_oauth,
         json=payload,
     )
     if response.status_code != 201:
@@ -62,9 +68,9 @@ def set_rules(headers, delete, bearer_token):
     print(json.dumps(response.json()))
 
 
-def get_stream(headers, set, bearer_token):
+def get_stream(set):
     response = requests.get(
-        "https://api.twitter.com/2/tweets/search/stream", headers=headers, stream=True,
+        "https://api.twitter.com/2/tweets/search/stream", auth=bearer_oauth, stream=True,
     )
     print(response.status_code)
     if response.status_code != 200:
@@ -80,12 +86,10 @@ def get_stream(headers, set, bearer_token):
 
 
 def main():
-    bearer_token = os.environ.get("BEARER_TOKEN")
-    headers = create_headers(bearer_token)
-    rules = get_rules(headers, bearer_token)
-    delete = delete_all_rules(headers, bearer_token, rules)
-    set = set_rules(headers, delete, bearer_token)
-    get_stream(headers, set, bearer_token)
+    rules = get_rules()
+    delete = delete_all_rules(rules)
+    set = set_rules(delete)
+    get_stream(set)
 
 
 if __name__ == "__main__":

Follows-Lookup/followers_lookup.py

@@ -4,10 +4,7 @@
 
 # To set your environment variables in your terminal run the following line:
 # export 'BEARER_TOKEN'='<your_bearer_token>'
-
-
-def auth():
-    return os.environ.get("BEARER_TOKEN")
+bearer_token = os.environ.get("BEARER_TOKEN")
 
 
 def create_url():
@@ -20,13 +17,18 @@ def get_params():
     return {"user.fields": "created_at"}
 
 
-def create_headers(bearer_token):
-    headers = {"Authorization": "Bearer {}".format(bearer_token)}
-    return headers
+def bearer_oauth(r):
+    """
+    Method required by bearer token authentication.
+    """
+
+    r.headers["Authorization"] = f"Bearer {bearer_token}"
+    r.headers["User-Agent"] = "v2FollowersLookupPython"
+    return r
 
 
-def connect_to_endpoint(url, headers, params):
-    response = requests.request("GET", url, headers=headers, params=params)
+def connect_to_endpoint(url, params):
+    response = requests.request("GET", url, auth=bearer_oauth, params=params)
     print(response.status_code)
     if response.status_code != 200:
         raise Exception(
@@ -38,11 +40,9 @@ def connect_to_endpoint(url, headers, params):
 
 
 def main():
-    bearer_token = auth()
     url = create_url()
-    headers = create_headers(bearer_token)
     params = get_params()
-    json_response = connect_to_endpoint(url, headers, params)
+    json_response = connect_to_endpoint(url, params)
     print(json.dumps(json_response, indent=4, sort_keys=True))
 
 

Follows-Lookup/following_lookup.py

@@ -4,10 +4,7 @@
 
 # To set your environment variables in your terminal run the following line:
 # export 'BEARER_TOKEN'='<your_bearer_token>'
-
-
-def auth():
-    return os.environ.get("BEARER_TOKEN")
+bearer_token = os.environ.get("BEARER_TOKEN")
 
 
 def create_url():
@@ -20,13 +17,18 @@ def get_params():
     return {"user.fields": "created_at"}
 
 
-def create_headers(bearer_token):
-    headers = {"Authorization": "Bearer {}".format(bearer_token)}
-    return headers
+def bearer_oauth(r):
+    """
+    Method required by bearer token authentication.
+    """
+
+    r.headers["Authorization"] = f"Bearer {bearer_token}"
+    r.headers["User-Agent"] = "v2FollowingLookupPython"
+    return r
 
 
-def connect_to_endpoint(url, headers, params):
-    response = requests.request("GET", url, headers=headers, params=params)
+def connect_to_endpoint(url, params):
+    response = requests.request("GET", url, auth=bearer_oauth, params=params)
     print(response.status_code)
     if response.status_code != 200:
         raise Exception(
@@ -38,11 +40,9 @@ def connect_to_endpoint(url, headers, params):
 
 
 def main():
-    bearer_token = auth()
     url = create_url()
-    headers = create_headers(bearer_token)
     params = get_params()
-    json_response = connect_to_endpoint(url, headers, params)
+    json_response = connect_to_endpoint(url, params)
     print(json.dumps(json_response, indent=4, sort_keys=True))
 
 

Full-Archive-Search/full-archive-search.py

@@ -13,22 +13,26 @@
 query_params = {'query': '(from:twitterdev -is:retweet) OR #twitterdev','tweet.fields': 'author_id'}
 
 
-def create_headers(bearer_token):
-    headers = {"Authorization": "Bearer {}".format(bearer_token)}
-    return headers
+def bearer_oauth(r):
+    """
+    Method required by bearer token authentication.
+    """
 
+    r.headers["Authorization"] = f"Bearer {bearer_token}"
+    r.headers["User-Agent"] = "v2FullArchiveSearchPython"
+    return r
 
-def connect_to_endpoint(url, headers, params):
-    response = requests.request("GET", search_url, headers=headers, params=params)
+
+def connect_to_endpoint(url, params):
+    response = requests.request("GET", search_url, auth=bearer_oauth, params=params)
     print(response.status_code)
     if response.status_code != 200:
         raise Exception(response.status_code, response.text)
     return response.json()
 
 
 def main():
-    headers = create_headers(bearer_token)
-    json_response = connect_to_endpoint(search_url, headers, query_params)
+    json_response = connect_to_endpoint(search_url, query_params)
     print(json.dumps(json_response, indent=4, sort_keys=True))
 
 

Full-Archive-Tweet-Counts/full_archive_tweet_counts.py

@@ -12,22 +12,26 @@
 query_params = {'query': 'from:twitterdev','granularity': 'day', 'start_time': '2021-01-01T00:00:00Z'}
 
 
-def create_headers(bearer_token):
-    headers = {"Authorization": "Bearer {}".format(bearer_token)}
-    return headers
+def bearer_oauth(r):
+    """
+    Method required by bearer token authentication.
+    """
 
+    r.headers["Authorization"] = f"Bearer {bearer_token}"
+    r.headers["User-Agent"] = "v2FullArchiveTweetCountsPython"
+    return r
 
-def connect_to_endpoint(url, headers, params):
-    response = requests.request("GET", search_url, headers=headers, params=params)
+
+def connect_to_endpoint(url, params):
+    response = requests.request("GET", search_url, auth=bearer_oauth, params=params)
     print(response.status_code)
     if response.status_code != 200:
         raise Exception(response.status_code, response.text)
     return response.json()
 
 
 def main():
-    headers = create_headers(bearer_token)
-    json_response = connect_to_endpoint(search_url, headers, query_params)
+    json_response = connect_to_endpoint(search_url, query_params)
     print(json.dumps(json_response, indent=4, sort_keys=True))
 
 

Likes-Lookup/liked_tweets.py

@@ -4,10 +4,7 @@
 
 # To set your enviornment variables in your terminal run the following line:
 # export 'BEARER_TOKEN'='<your_bearer_token>'
-
-
-def auth():
-    return os.environ.get("BEARER_TOKEN")
+bearer_token = os.environ.get("BEARER_TOKEN")
 
 
 def create_url():
@@ -28,14 +25,19 @@ def create_url():
     return url, tweet_fields
 
 
-def create_headers(bearer_token):
-    headers = {"Authorization": "Bearer {}".format(bearer_token)}
-    return headers
+def bearer_oauth(r):
+    """
+    Method required by bearer token authentication.
+    """
+
+    r.headers["Authorization"] = f"Bearer {bearer_token}"
+    r.headers["User-Agent"] = "v2LikedTweetsPython"
+    return r
 
 
-def connect_to_endpoint(url, headers, tweet_fields):
+def connect_to_endpoint(url, tweet_fields):
     response = requests.request(
-        "GET", url, headers=headers, params=tweet_fields)
+        "GET", url, auth=bearer_oauth, params=tweet_fields)
     print(response.status_code)
     if response.status_code != 200:
         raise Exception(
@@ -47,10 +49,8 @@ def connect_to_endpoint(url, headers, tweet_fields):
 
 
 def main():
-    bearer_token = auth()
     url, tweet_fields = create_url()
-    headers = create_headers(bearer_token)
-    json_response = connect_to_endpoint(url, headers, tweet_fields)
+    json_response = connect_to_endpoint(url, tweet_fields)
     print(json.dumps(json_response, indent=4, sort_keys=True))
 
 

Likes-Lookup/liking_users.py

@@ -4,10 +4,7 @@
 
 # To set your enviornment variables in your terminal run the following line:
 # export 'BEARER_TOKEN'='<your_bearer_token>'
-
-
-def auth():
-    return os.environ.get("BEARER_TOKEN")
+bearer_token = os.environ.get("BEARER_TOKEN")
 
 
 def create_url():
@@ -25,13 +22,18 @@ def create_url():
     return url, user_fields
 
 
-def create_headers(bearer_token):
-    headers = {"Authorization": "Bearer {}".format(bearer_token)}
-    return headers
+def bearer_oauth(r):
+    """
+    Method required by bearer token authentication.
+    """
+
+    r.headers["Authorization"] = f"Bearer {bearer_token}"
+    r.headers["User-Agent"] = "v2LikingUsersPython"
+    return r
 
 
-def connect_to_endpoint(url, headers, user_fields):
-    response = requests.request("GET", url, headers=headers, params=user_fields)
+def connect_to_endpoint(url, user_fields):
+    response = requests.request("GET", url, auth=bearer_oauth, params=user_fields)
     print(response.status_code)
     if response.status_code != 200:
         raise Exception(
@@ -43,10 +45,8 @@ def connect_to_endpoint(url, headers, user_fields):
 
 
 def main():
-    bearer_token = auth()
     url, tweet_fields = create_url()
-    headers = create_headers(bearer_token)
-    json_response = connect_to_endpoint(url, headers, tweet_fields)
+    json_response = connect_to_endpoint(url, tweet_fields)
     print(json.dumps(json_response, indent=4, sort_keys=True))
 
 

Recent-Search/recent_search.py

@@ -13,22 +13,26 @@
 query_params = {'query': '(from:twitterdev -is:retweet) OR #twitterdev','tweet.fields': 'author_id'}
 
 
-def create_headers(bearer_token):
-    headers = {"Authorization": "Bearer {}".format(bearer_token)}
-    return headers
+def bearer_oauth(r):
+    """
+    Method required by bearer token authentication.
+    """
 
+    r.headers["Authorization"] = f"Bearer {bearer_token}"
+    r.headers["User-Agent"] = "v2RecentSearchPython"
+    return r
 
-def connect_to_endpoint(url, headers, params):
-    response = requests.request("GET", search_url, headers=headers, params=params)
+
+def connect_to_endpoint(url, params):
+    response = requests.get(search_url, auth=bearer_oauth, params=params)
     print(response.status_code)
     if response.status_code != 200:
         raise Exception(response.status_code, response.text)
     return response.json()
 
 
 def main():
-    headers = create_headers(bearer_token)
-    json_response = connect_to_endpoint(search_url, headers, query_params)
+    json_response = connect_to_endpoint(search_url, query_params)
     print(json.dumps(json_response, indent=4, sort_keys=True))