FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
rocketapi-python/rocketapi/threadsapi.py at main · rocketapi-io/rocketapi-python · GitHub
rocketapi-io
/
rocketapi-python
Public
Notifications
You must be signed in to change notification settings
Fork
4
Star
43
Code
Pull requests
0
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
rocketapi-python
/
rocketapi
/
threadsapi.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
185 lines (144 loc) · 6.5 KB
Breadcrumbs
rocketapi-python
/
rocketapi
/
threadsapi.py
Copy path
File metadata and controls
185 lines (144 loc) · 6.5 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
from
rocketapi
.
exceptions
import
NotFoundException
,
BadResponseException
from
rocketapi
.
rocketapi
import
RocketAPI
class
ThreadsAPI
(
RocketAPI
):
def
__init__
(
self
,
token
,
max_timeout
=
30
):
"""
Threads API client.
Args:
token (str): Your RocketAPI token (https://rocketapi.io/dashboard/)
max_timeout (int): Maximum timeout for requests. Please, don't use values lower than 15 seconds, it may cause problems with API.
For debugging purposes you can use the following variables:
last_response (dict): contains the last response from the API.
counter (int): contains the number of requests made in the current session.
For more information, see documentation: https://docs.rocketapi.io/api/
"""
self
.
last_response
=
None
self
.
counter
=
0
super
().
__init__
(
token
,
max_timeout
=
max_timeout
)
def
request
(
self
,
method
,
data
):
response
=
super
().
request
(
method
,
data
)
self
.
last_response
=
response
self
.
counter
+=
1
if
response
[
"status"
]
==
"done"
:
if
(
response
[
"response"
][
"status_code"
]
==
200
and
response
[
"response"
][
"content_type"
]
==
"application/json"
):
return
response
[
"response"
][
"body"
]
elif
response
[
"response"
][
"status_code"
]
==
404
:
raise
NotFoundException
(
"Instagram resource not found"
)
else
:
raise
BadResponseException
(
"Bad response from Threads"
)
raise
BadResponseException
(
"Bad response from RocketAPI"
)
def
search_users
(
self
,
query
):
"""
Search for a specific user in Threads
Args:
query (str): Username to search for
For more information, see documentation: https://docs.rocketapi.io/api/threads/search_users
"""
return
self
.
request
(
"threads/search_users"
, {
"query"
:
query
})
def
get_user_info
(
self
,
user_id
):
"""
Retrieve Threads user information by id.
Args:
user_id (int): User id
For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_info
"""
return
self
.
request
(
"threads/user/get_info"
, {
"id"
:
user_id
})
def
get_user_feed
(
self
,
user_id
,
max_id
=
None
):
"""
Retrieve Threads user feed by id.
Args:
user_id (int): User id
max_id (str): Use for pagination
You can use the `max_id` parameter to paginate through the media (take from the `next_max_id` field of the response).
For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_feed
"""
payload
=
{
"id"
:
user_id
}
if
max_id
is
not
None
:
payload
[
"max_id"
]
=
max_id
return
self
.
request
(
"threads/user/get_feed"
,
payload
)
def
get_user_replies
(
self
,
user_id
,
max_id
=
None
):
"""
Retrieve Threads user replies by id.
Args:
user_id (int): User id
max_id (str): Use for pagination
You can use the `max_id` parameter to paginate through the media (take from the `next_max_id` field of the response).
For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_replies
"""
payload
=
{
"id"
:
user_id
}
if
max_id
is
not
None
:
payload
[
"max_id"
]
=
max_id
return
self
.
request
(
"threads/user/get_replies"
,
payload
)
def
get_user_followers
(
self
,
user_id
,
max_id
=
None
):
"""
Retrieve Threads user followers by id.
Args:
user_id (int): User id
max_id (str): Use for pagination
You can use the `max_id` parameter to paginate through followers (take from the `next_max_id` field of the response).
For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_followers
"""
payload
=
{
"id"
:
user_id
}
if
max_id
is
not
None
:
payload
[
"max_id"
]
=
max_id
return
self
.
request
(
"threads/user/get_followers"
,
payload
)
def
search_user_followers
(
self
,
user_id
,
query
):
"""
Search Threads user followers by user id.
Args:
user_id (int): User id
query (str): Search query
For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_followers
"""
return
self
.
request
(
"threads/user/get_followers"
, {
"id"
:
user_id
,
"query"
:
query
}
)
def
get_user_following
(
self
,
user_id
,
max_id
=
None
):
"""
Retrieve Threads user following by id.
Args:
user_id (int): User id
max_id (str): Use for pagination
You can use the `max_id` parameter to paginate through followers (take from the `next_max_id` field of the response).
For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_following
"""
payload
=
{
"id"
:
user_id
}
if
max_id
is
not
None
:
payload
[
"max_id"
]
=
max_id
return
self
.
request
(
"threads/user/get_following"
,
payload
)
def
search_user_following
(
self
,
user_id
,
query
):
"""
Search Threads user following by user id.
Args:
user_id (int): User id
query (str): Search query
For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_following
"""
return
self
.
request
(
"threads/user/get_following"
, {
"id"
:
user_id
,
"query"
:
query
}
)
def
get_thread_replies
(
self
,
thread_id
,
max_id
=
None
):
"""
Retrieve thread replies by id.
Args:
thread_id (int): Thread id
max_id (str): Use for pagination
You can use the `max_id` parameter to paginate through the media (take from the `paging_tokens["downwards"]` field of the response).
For more information, see documentation: https://docs.rocketapi.io/api/threads/thread/get_replies
"""
payload
=
{
"id"
:
thread_id
}
if
max_id
is
not
None
:
payload
[
"max_id"
]
=
max_id
return
self
.
request
(
"threads/thread/get_replies"
,
payload
)
def
get_thread_likes
(
self
,
thread_id
):
"""
Retrieve thread likes by id.
Args:
thread_id (int): Thread id
For more information, see documentation: https://docs.rocketapi.io/api/threads/thread/get_likes
"""
payload
=
{
"id"
:
thread_id
}
return
self
.
request
(
"threads/thread/get_likes"
,
payload
)
Back
|
FazBrowse Home
|
New Git URL