Skip to content

Commit 7f423b1

Browse files
author
Jim Moffitt
committed
JM: adding six examples for Direct Messages v2 m1.:
1 parent 0252ae4 commit 7f423b1

6 files changed

Lines changed: 660 additions & 0 deletions

File tree

‎Direct-Messages-lookup/get_events_by_conversation.py‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
import base64
2+
import hashlib
3+
import os
4+
import re
5+
import json
6+
import requests
7+
from requests_oauthlib import OAuth2Session
8+
9+
# This example is set up to retrieve Direct Message events by conversation ID. This supports both
10+
# one-to-one and group conversations.
11+
GET_DMS_EVENTS_URL = "https://api.twitter.com/2/dm_conversations/:dm_conversation_id/dm_events"
12+
13+
#-----------------------------------------------------------------------------------------------------------------------
14+
# These variables need to be updated to the setting that match how your Twitter App is set-up at
15+
# https://developer.twitter.com/en/portal/dashboard. These will not change from run-by-run.
16+
client_id = 'dE5vYmJaa3o5QWRlNEt1T01DZ206MTpjaQ'
17+
#This must match *exactly* the redirect URL specified in the Developer Portal.
18+
redirect_uri = "https://www.example.com"
19+
#-----------------------------------------------------------------------------------------------------------------------
20+
# This variable specifies the conversation to retrieve. A more ready-to-be used example would
21+
# have this passed in from some calling code.
22+
# What is the ID of the conversatikon to retrieve?
23+
dm_conversation_id = "1512210732774948865"
24+
#-----------------------------------------------------------------------------------------------------------------------
25+
26+
def handle_oauth():
27+
28+
# Set the scopes needed to be granted by the authenticating user.
29+
scopes = ["dm.read", "tweet.read", "users.read", "offline.access"]
30+
31+
# Create a code verifier
32+
code_verifier = base64.urlsafe_b64encode(os.urandom(30)).decode("utf-8")
33+
code_verifier = re.sub("[^a-zA-Z0-9]+", "", code_verifier)
34+
35+
# Create a code challenge
36+
code_challenge = hashlib.sha256(code_verifier.encode("utf-8")).digest()
37+
code_challenge = base64.urlsafe_b64encode(code_challenge).decode("utf-8")
38+
code_challenge = code_challenge.replace("=", "")
39+
40+
# Start and OAuth 2.0 session
41+
oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes)
42+
43+
# Create an authorize URL
44+
auth_url = "https://twitter.com/i/oauth2/authorize"
45+
authorization_url, state = oauth.authorization_url(
46+
auth_url, code_challenge=code_challenge, code_challenge_method="S256"
47+
)
48+
49+
# Visit the URL to authorize your App to make requests on behalf of a user
50+
print(
51+
"Visit the following URL to authorize your App on behalf of your Twitter handle in a browser:"
52+
)
53+
print(authorization_url)
54+
55+
# Paste in your authorize URL to complete the request
56+
authorization_response = input(
57+
"Paste in the full URL after you've authorized your App:\n"
58+
)
59+
60+
# Fetch your access token
61+
token_url = "https://api.twitter.com/2/oauth2/token"
62+
63+
# The following line of code will only work if you are using a type of App that is a public client
64+
auth = False
65+
66+
token = oauth.fetch_token(
67+
token_url=token_url,
68+
authorization_response=authorization_response,
69+
auth=auth,
70+
client_id=client_id,
71+
include_client_id=True,
72+
code_verifier=code_verifier,
73+
)
74+
75+
# Your access token
76+
access = token["access_token"]
77+
78+
return access
79+
80+
def get_events_by_conservation_id(dm_conversation_id):
81+
82+
access = handle_oauth()
83+
84+
headers = {
85+
"Authorization": "Bearer {}".format(access),
86+
"Content-Type": "application/json",
87+
"User-Agent": "TwitterDevSampleCode",
88+
"X-TFE-Experiment-environment": "staging1",
89+
"Dtab-Local": "/s/gizmoduck/test-users-temporary => /s/gizmoduck/gizmoduck"
90+
}
91+
92+
request_url = GET_DMS_EVENTS_URL.replace(':dm_conversation_id', str(dm_conversation_id))
93+
94+
response = requests.request("GET", request_url, headers=headers)
95+
96+
if response.status_code != 200:
97+
print("Request returned an error: {} {}".format(response.status_code, response.text))
98+
else:
99+
print(f"Response code: {response.status_code}")
100+
return response
101+
102+
def main():
103+
response = get_events_by_conservation_id(dm_conversation_id)
104+
print(json.dumps(json.loads(response.text), indent=4, sort_keys=True))
105+
106+
if __name__ == "__main__":
107+
main()

‎Direct-Messages-lookup/get_one_to_one_conversation_events.py‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
import base64
2+
import hashlib
3+
import os
4+
import re
5+
import json
6+
import requests
7+
from requests_oauthlib import OAuth2Session
8+
9+
# This example is set up to retrieve Direct Message conversation events associated with a one-to-one message.
10+
# Currently, the v2 DM endpoints support three conversation event types: MessageCreate, ParticipantsJoin, and
11+
# ParticipantsLeave.
12+
GET_DM_EVENTS_URL = "https://api.twitter.com/2/dm_conversations/with/:participant_id/dm_events"
13+
14+
#-----------------------------------------------------------------------------------------------------------------------
15+
# These variables need to be updated to the setting that match how your Twitter App is set-up at
16+
# https://developer.twitter.com/en/portal/dashboard. These will not change from run-by-run.
17+
client_id = 'dE5vYmJaa3o5QWRlNEt1T01DZ206MTpjaQ'
18+
#This must match *exactly* the redirect URL specified in the Developer Portal.
19+
redirect_uri = "https://www.example.com"
20+
#-----------------------------------------------------------------------------------------------------------------------
21+
# This variable indicates the participant of the one-to-one conversation. A more ready-to-be used example would
22+
# have this passed in from some calling code.
23+
# Who is this one-to-one conversation with?
24+
participant_id = "906948460078698496"
25+
#-----------------------------------------------------------------------------------------------------------------------
26+
27+
def handle_oauth():
28+
29+
# Set the scopes needed to be granted by the authenticating user.
30+
scopes = ["dm.read", "tweet.read", "users.read", "offline.access"]
31+
32+
# Create a code verifier
33+
code_verifier = base64.urlsafe_b64encode(os.urandom(30)).decode("utf-8")
34+
code_verifier = re.sub("[^a-zA-Z0-9]+", "", code_verifier)
35+
36+
# Create a code challenge
37+
code_challenge = hashlib.sha256(code_verifier.encode("utf-8")).digest()
38+
code_challenge = base64.urlsafe_b64encode(code_challenge).decode("utf-8")
39+
code_challenge = code_challenge.replace("=", "")
40+
41+
# Start and OAuth 2.0 session
42+
oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes)
43+
44+
# Create an authorize URL
45+
auth_url = "https://twitter.com/i/oauth2/authorize"
46+
authorization_url, state = oauth.authorization_url(
47+
auth_url, code_challenge=code_challenge, code_challenge_method="S256"
48+
)
49+
50+
# Visit the URL to authorize your App to make requests on behalf of a user
51+
print(
52+
"Visit the following URL to authorize your App on behalf of your Twitter handle in a browser:"
53+
)
54+
print(authorization_url)
55+
56+
# Paste in your authorize URL to complete the request
57+
authorization_response = input(
58+
"Paste in the full URL after you've authorized your App:\n"
59+
)
60+
61+
# Fetch your access token
62+
token_url = "https://api.twitter.com/2/oauth2/token"
63+
64+
# The following line of code will only work if you are using a type of App that is a public client
65+
auth = False
66+
67+
token = oauth.fetch_token(
68+
token_url=token_url,
69+
authorization_response=authorization_response,
70+
auth=auth,
71+
client_id=client_id,
72+
include_client_id=True,
73+
code_verifier=code_verifier,
74+
)
75+
76+
# Your access token
77+
access = token["access_token"]
78+
79+
return access
80+
81+
def get_one_to_one_conversation_events(participant_id):
82+
83+
access = handle_oauth()
84+
85+
headers = {
86+
"Authorization": "Bearer {}".format(access),
87+
"Content-Type": "application/json",
88+
"User-Agent": "TwitterDevSampleCode",
89+
"X-TFE-Experiment-environment": "staging1",
90+
"Dtab-Local": "/s/gizmoduck/test-users-temporary => /s/gizmoduck/gizmoduck"
91+
}
92+
93+
request_url = GET_DM_EVENTS_URL.replace(':participant_id', str(participant_id))
94+
95+
response = requests.request("GET", request_url, headers=headers)
96+
97+
if response.status_code != 200:
98+
print("Request returned an error: {} {}".format(response.status_code, response.text))
99+
else:
100+
print(f"Response code: {response.status_code}")
101+
return response
102+
103+
def main():
104+
response = get_one_to_one_conversation_events(participant_id)
105+
print(json.dumps(json.loads(response.text), indent=4, sort_keys=True))
106+
107+
if __name__ == "__main__":
108+
main()

‎Direct-Messages-lookup/get_user_conversation_events.py‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
import base64
2+
import hashlib
3+
import os
4+
import re
5+
import json
6+
import requests
7+
from requests_oauthlib import OAuth2Session
8+
9+
# This example is set up to retrieve Direct Message events of the authenticating user. This supports both
10+
# one-to-one and group conversations.
11+
GET_DM_EVENTS_URL = "https://api.twitter.com/2/dm_events"
12+
13+
#-----------------------------------------------------------------------------------------------------------------------
14+
# These variables need to be updated to the setting that match how your Twitter App is set-up at
15+
# https://developer.twitter.com/en/portal/dashboard. These will not change from run-by-run.
16+
client_id = "dE5vYmJaa3o5QWRlNEt1T01DZ206MTpjaQ"
17+
#This must match *exactly* the redirect URL specified in the Developer Portal.
18+
redirect_uri = "https://www.example.com"
19+
#-----------------------------------------------------------------------------------------------------------------------
20+
21+
def handle_oauth():
22+
23+
# Set the scopes needed to be granted by the authenticating user.
24+
scopes = ["dm.read", "tweet.read", "users.read", "offline.access"]
25+
26+
# Create a code verifier.
27+
code_verifier = base64.urlsafe_b64encode(os.urandom(30)).decode("utf-8")
28+
code_verifier = re.sub("[^a-zA-Z0-9]+", "", code_verifier)
29+
30+
# Create a code challenge.
31+
code_challenge = hashlib.sha256(code_verifier.encode("utf-8")).digest()
32+
code_challenge = base64.urlsafe_b64encode(code_challenge).decode("utf-8")
33+
code_challenge = code_challenge.replace("=", "")
34+
35+
# Start an OAuth 2.0 session.
36+
oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes)
37+
38+
# Create an authorize URL.
39+
auth_url = "https://twitter.com/i/oauth2/authorize"
40+
authorization_url, state = oauth.authorization_url(
41+
auth_url, code_challenge=code_challenge, code_challenge_method="S256"
42+
)
43+
44+
# Visit the URL to authorize your App to make requests on behalf of a user.
45+
print(
46+
"Visit the following URL to authorize your App on behalf of your Twitter handle in a browser:"
47+
)
48+
print(authorization_url)
49+
50+
# Paste in your authorize URL to complete the request.
51+
authorization_response = input(
52+
"Paste in the full URL after you've authorized your App:\n"
53+
)
54+
55+
# Fetch your access token.
56+
token_url = "https://api.twitter.com/2/oauth2/token"
57+
58+
# The following line of code will only work if you are using a type of App that is a public client.
59+
auth = False
60+
61+
token = oauth.fetch_token(
62+
token_url=token_url,
63+
authorization_response=authorization_response,
64+
auth=auth,
65+
client_id=client_id,
66+
include_client_id=True,
67+
code_verifier=code_verifier,
68+
)
69+
70+
# The access token.
71+
access = token["access_token"]
72+
73+
return access
74+
75+
def get_user_conversation_events():
76+
77+
access = handle_oauth()
78+
79+
headers = {
80+
"Authorization": "Bearer {}".format(access),
81+
"Content-Type": "application/json",
82+
"User-Agent": "TwitterDevSampleCode",
83+
"X-TFE-Experiment-environment": "staging1",
84+
"Dtab-Local": "/s/gizmoduck/test-users-temporary => /s/gizmoduck/gizmoduck"
85+
}
86+
87+
request_url = GET_DM_EVENTS_URL
88+
89+
response = requests.request("GET", request_url, headers=headers)
90+
91+
if response.status_code != 200:
92+
print("Request returned an error: {} {}".format(response.status_code, response.text))
93+
else:
94+
print(f"Response code: {response.status_code}")
95+
return response
96+
97+
def main():
98+
response = get_user_conversation_events()
99+
print(json.dumps(json.loads(response.text), indent=4, sort_keys=True))
100+
101+
if __name__ == "__main__":
102+
main()

0 commit comments

Comments
 (0)