aminoacid.client 

  1from os import PathLike
  2from time import time
  3from typing import BinaryIO, List, Optional, Union
  4from uuid import uuid4
  5
  6from . import exceptions
  7from .abc import (
  8    AminoBaseClass,
  9    Embed,
 10    Member,
 11    Message,
 12    Session,
 13    Thread,
 14    User,
 15    Community,
 16    Blog,
 17)
 18
 19
 20class ApiClient(AminoBaseClass):
 21    """ApiClient skeleton to reduce repeating API calls in code and to clean up code"""
 22
 23    async def login(self, email: str, password: str) -> User:
 24        """Authenticates with the given email and password
 25
 26        Parameters
 27        ----------
 28        email : str
 29            Email to login with
 30        password : str
 31            Password to login with
 32
 33        Returns
 34        -------
 35        User
 36            The user that was authenticated
 37        """
 38        response = await (
 39            await self._http.request(
 40                "POST",
 41                "/g/s/auth/login",
 42                json={
 43                    "email": email,
 44                    "v": 2,
 45                    "secret": f"0 {password}",
 46                    "deviceID": self._http.device,
 47                    "clientType": 100,
 48                    "action": "normal",
 49                    "timestamp": int(time() * 1000),
 50                },
 51            )
 52        ).json()
 53
 54        if response.get("api:statuscode") != 0:
 55            return exceptions.handle_exception(response.get("api:statuscode"), response)
 56
 57        self.profile = User(**(response["userProfile"]), client=self)
 58        self._http.session = Session(response.get("sid"))
 59
 60        return self.profile
 61
 62    async def fetch_thread(self, threadId: str, ndcId: Optional[str] = "") -> Thread:
 63        """Fetches a given `Thread`
 64
 65        Parameters
 66        ----------
 67        threadId : str
 68            the ID of the thread
 69        ndcId : Optional[str], optional
 70            the community the thread is in, if not given (or 0) it will look for the thread in global, by default ""
 71
 72        Returns
 73        -------
 74        Thread
 75            The `Thread` object that was requested
 76        """
 77        response = await (
 78            await self._http.request(
 79                "GET",
 80                f"/x{ndcId}/s/chat/thread/{threadId}"
 81                if ndcId
 82                else f"/g/s/chat/thread/{threadId}",
 83            )
 84        ).json()
 85
 86        if response.get("api:statuscode") != 0:
 87            return exceptions.handle_exception(response.get("api:statuscode"), response)
 88        return Thread(**(response["thread"]), client=self)
 89
 90    async def fetch_message(
 91        self, messageId: str, threadId: str, ndcId: Optional[str] = ""
 92    ) -> Message:
 93        """Fetches a given `Message` from a `Thread`
 94
 95        Parameters
 96        ----------
 97        messageId : str
 98            the ID of the `Message` to fetch
 99        threadId : str
100            the ID of the `Thread` to fetch from
101        ndcId : Optional[str], optional
102            the community the thread is in, if not given (or 0) it will look for the thread in global, by default, by default ""
103
104        Returns
105        -------
106        Message
107            The `Message` object that was requested
108        """
109        response = await (
110            await self._http.request(
111                "GET",
112                f"/x{ndcId}/s/chat/thread/{threadId}/message/{messageId}"
113                if ndcId
114                else f"/g/s/chat/thread/{threadId}/message/{messageId}",
115            )
116        ).json()
117
118        if response.get("api:statuscode") != 0:
119            return exceptions.handle_exception(response.get("api:statuscode"), response)
120        return Thread(**(response["message"]), client=self)
121
122    async def fetch_user(self, userId: str) -> User:
123        """Fetches a user from the API
124
125        Parameters
126        ----------
127        userId : str
128            the userId to search for
129
130        Returns
131        -------
132        User
133            The `User` object that was requested
134        """
135        response = await (
136            await self._http.request("GET", f"/g/s/user-profile/{userId}")
137        ).json()
138
139        if response.get("api:statuscode") != 0:
140            return exceptions.handle_exception(response.get("api:statuscode"), response)
141        return User(**(response["userProfile"]), client=self)
142
143    async def fetch_member(self, userId: str, ndcId: str) -> Member:
144        """Fetches a member for a given community
145
146        Parameters
147        ----------
148        userId : str
149            The userId to search for
150        ndcId : str
151            The community the user is a member in
152
153        Returns
154        -------
155        Member
156            The `Member` object requested for the given community
157        """
158        response = await (
159            await self._http.request("GET", f"/x{ndcId}/s/user-profile/{userId}")
160        ).json()
161
162        if response.get("api:statuscode") != 0:
163            return exceptions.handle_exception(response.get("api:statuscode"), response)
164
165        return Member(**(response["userProfile"]), client=self)
166
167    async def send_message(
168        self,
169        threadId: str,
170        content: str,
171        *,
172        ndcId: Optional[str] = "",
173        embed: Optional[Embed] = Embed(None, None, None, None, None, None),
174        **kwargs,
175    ) -> Message:
176        """Sends a message to a given Thread.
177
178        Parameters
179        ----------
180        threadId : str
181            the thread to send the message to
182        content : str
183            the content of the message to send, must be within 2000 characters
184        ndcId : Optional[str], optional
185            The community the Thread is in, if not given (or 0) it will look for a global chat, by default ""
186        embed : Optional[Embed], optional
187            Embed to send alongside the message, by default empty embed
188
189        Returns
190        -------
191        Message
192            Returns the `Message` object of the sent message
193        """
194        response = await (
195            await self._http.request(
196                "POST",
197                f"/x{ndcId}/s/chat/thread/{threadId}/message"
198                if ndcId
199                else f"/g/s/chat/thread/{threadId}",
200                json={
201                    "type": 0,
202                    "content": content,
203                    "clientRefId": int(time() % 86400),
204                    "timestamp": int(time() * 1000),
205                    "attachedObject": embed.__dict__(),
206                    **kwargs,
207                },
208            )
209        ).json()
210
211        if response.get("api:statuscode") != 0:
212            return exceptions.handle_exception(response.get("api:statuscode"), response)
213        return Message(**(response["message"]), client=self)
214
215    async def start_dm(self, userId: str, *, ndcId: Optional[str] = "") -> Thread:
216        """Start direct messaging a user or return the Thread if a DM already exists
217
218        Parameters
219        ----------
220        userId : str
221            the user you want to start a DM with
222        ndcId : Optional[str], optional
223            the community the member is in, if sending to a user, this  will be empty
224
225        Returns
226        -------
227        Thread
228            Thread of the DMs
229        """
230        response = await (
231            await self._http.request(
232                "POST",
233                f"/x{ndcId}/s/chat/thread/" if ndcId else f"/g/s/chat/thread/",
234                json={
235                    "title": None,
236                    "content": None,
237                    "initialMessageContent": None,
238                    "timestamp": int(time() * 1000),
239                    "inviteeUids": [userId],
240                    "type": 0,
241                },
242            )
243        ).json()
244
245        if response.get("api:statuscode") != 0:
246            return exceptions.handle_exception(response.get("api:statuscode"), response)
247
248        return Thread(**(response["thread"]), client=self)
249
250    async def message_user(self, userId: str, **kwargs) -> Message:
251        """Send a message to a user's DMs, this starts the DMs (via `start_dm()`) if they don't exist already
252
253        Parameters
254        ----------
255        userId : str
256            userId to send a message to
257
258        Returns
259        -------
260        Message
261            Return an object representing the sent message
262        """
263        return await (
264            await self.start_dm(userId=userId, ndcId=kwargs.pop("ndcId", ""))
265        ).send(**kwargs)
266
267    async def upload_image(self, image: Union[bytes, BinaryIO, PathLike]) -> str:
268        """Upload an image to the amino servers
269
270        Parameters
271        ----------
272        image : Union[bytes, BinaryIO, PathLike]
273            Either the read out image, an IO object representing the Image, or the image path
274
275        Returns
276        -------
277        str
278            The direct link of the image
279        """
280        if isinstance(image, (BinaryIO, PathLike)):
281            kwargs = {"file": image}
282        else:
283            kwargs = {"data": image}
284        response = await (
285            await self._http.request("POST", "/g/s/media/upload", **kwargs)
286        ).json()
287
288        if response.get("api:statuscode") != 0:
289            return exceptions.handle_exception(response.get("api:statuscode"), response)
290
291        return response["mediaValue"]
292
293    async def set_device(self, ndcId: str = "") -> Optional[dict]:
294        """Set the device pushToken to receive notifications on the websocket
295
296        Parameters
297        ----------
298        ndcId : str, optional
299            Optionally a community to set the token in, by default ""
300
301        Returns
302        -------
303        Optional[dict]
304            A dictionary containing devOptions, returns None if it doesn't exist
305        """
306        response = await (
307            await self._http.request(
308                "POST",
309                f"/x{ndcId}/s/device" if ndcId else "/g/s/device",
310                json={
311                    "deviceID": self._http.device,
312                    "bundleID": "com.narvii.amino.master",
313                    "clientType": 100,
314                    "timezone": 60,
315                    "systemPushEnabled": True,
316                    "locale": "en_DE",
317                    "deviceToken": self._http.token,
318                    "deviceTokenType": 1,
319                    "timestamp": int(time() * 1000),
320                },
321            )
322        ).json()
323        if response.get("api:statuscode") != 0:
324            return exceptions.handle_exception(response.get("api:statuscode"), response)
325
326        return response["devOptions"]
327
328    async def fetch_communities(
329        self, start: int = 0, size: int = 25
330    ) -> List[Community]:
331        """Fetch a list of communities that the bot is in, this can be used to set tokens for each community
332
333        Parameters
334        ----------
335        start : int, optional
336            start index, by default 0
337        size : int, optional
338            amount of communities to fetch, by default 25
339
340        Returns
341        -------
342        List[Community]
343            List of `Community` objects describing the communities
344        """
345        response = await (
346            await self._http.request(
347                "GET", "/g/s/community/joined", params={"start": start, "size": size}
348            )
349        ).json()
350
351        if response.get("api:statuscode") != 0:
352            return exceptions.handle_exception(response.get("api:statuscode"), response)
353        return [
354            Community(**community, client=self)
355            for community in response["communityList"]
356        ]
357
358    async def fetch_blogs(
359        self, ndcId: str, start: int = 0, size: int = 25, *, userId: Optional[str] = ""
360    ) -> List[Blog]:
361        """Fetches blogs by user or community if no user is supplied
362
363        Parameters
364        ----------
365        ndcId : Optional[str], optional
366            the community id, by default ""
367        start : int, optional
368            offset to start at, by default 0
369        size : int, optional
370            amount of blogs to get, by default 25
371        userId : Optional[str], optional
372            the id of the user, by default ""
373        """
374        params = {
375            "start": start,
376            "size": size,
377            "url": f"/x{ndcId}/s/blog" if userId else f"/x{ndcId}/s/feed/blog-all",
378            **({"type": "user", "q": userId} if userId else {}),
379        }
380        response = await (
381            await self._http.request("GET", params.pop("url"), params=params)
382        ).json()
383
384        if response.get("api:statuscode") != 0:
385            return exceptions.handle_exception(response.get("api:statuscode"), response)
386        return [Blog(**blog, client=self) for blog in response["blogList"]]
387
388    async def tip_blog(self, ndcId: str, blogId: str, amount: int):
389        """Sends coins to a blog
390
391        Parameters
392        ----------
393        ndcId : str
394            community id that the blog is in
395        blogId : str
396            id of the blog
397        amount : int
398            the amount of coins to send
399        """
400        response = await (
401            await self._http.request(
402                "POST",
403                f"/x{ndcId}/s/blog/{blogId}/tipping",
404                json={
405                    "coins": amount,
406                    "tippingContext": {"transactionId": str(uuid4())},
407                    "timestamp": int(time() * 1000),
408                },
409            )
410        ).json()
411        if response.get("api:statuscode") != 0:
412            return exceptions.handle_exception(response.get("api:statuscode"), response)
class ApiClient(aminoacid.abc.AminoBaseClass): 
 21class ApiClient(AminoBaseClass):
 22    """ApiClient skeleton to reduce repeating API calls in code and to clean up code"""
 23
 24    async def login(self, email: str, password: str) -> User:
 25        """Authenticates with the given email and password
 26
 27        Parameters
 28        ----------
 29        email : str
 30            Email to login with
 31        password : str
 32            Password to login with
 33
 34        Returns
 35        -------
 36        User
 37            The user that was authenticated
 38        """
 39        response = await (
 40            await self._http.request(
 41                "POST",
 42                "/g/s/auth/login",
 43                json={
 44                    "email": email,
 45                    "v": 2,
 46                    "secret": f"0 {password}",
 47                    "deviceID": self._http.device,
 48                    "clientType": 100,
 49                    "action": "normal",
 50                    "timestamp": int(time() * 1000),
 51                },
 52            )
 53        ).json()
 54
 55        if response.get("api:statuscode") != 0:
 56            return exceptions.handle_exception(response.get("api:statuscode"), response)
 57
 58        self.profile = User(**(response["userProfile"]), client=self)
 59        self._http.session = Session(response.get("sid"))
 60
 61        return self.profile
 62
 63    async def fetch_thread(self, threadId: str, ndcId: Optional[str] = "") -> Thread:
 64        """Fetches a given `Thread`
 65
 66        Parameters
 67        ----------
 68        threadId : str
 69            the ID of the thread
 70        ndcId : Optional[str], optional
 71            the community the thread is in, if not given (or 0) it will look for the thread in global, by default ""
 72
 73        Returns
 74        -------
 75        Thread
 76            The `Thread` object that was requested
 77        """
 78        response = await (
 79            await self._http.request(
 80                "GET",
 81                f"/x{ndcId}/s/chat/thread/{threadId}"
 82                if ndcId
 83                else f"/g/s/chat/thread/{threadId}",
 84            )
 85        ).json()
 86
 87        if response.get("api:statuscode") != 0:
 88            return exceptions.handle_exception(response.get("api:statuscode"), response)
 89        return Thread(**(response["thread"]), client=self)
 90
 91    async def fetch_message(
 92        self, messageId: str, threadId: str, ndcId: Optional[str] = ""
 93    ) -> Message:
 94        """Fetches a given `Message` from a `Thread`
 95
 96        Parameters
 97        ----------
 98        messageId : str
 99            the ID of the `Message` to fetch
100        threadId : str
101            the ID of the `Thread` to fetch from
102        ndcId : Optional[str], optional
103            the community the thread is in, if not given (or 0) it will look for the thread in global, by default, by default ""
104
105        Returns
106        -------
107        Message
108            The `Message` object that was requested
109        """
110        response = await (
111            await self._http.request(
112                "GET",
113                f"/x{ndcId}/s/chat/thread/{threadId}/message/{messageId}"
114                if ndcId
115                else f"/g/s/chat/thread/{threadId}/message/{messageId}",
116            )
117        ).json()
118
119        if response.get("api:statuscode") != 0:
120            return exceptions.handle_exception(response.get("api:statuscode"), response)
121        return Thread(**(response["message"]), client=self)
122
123    async def fetch_user(self, userId: str) -> User:
124        """Fetches a user from the API
125
126        Parameters
127        ----------
128        userId : str
129            the userId to search for
130
131        Returns
132        -------
133        User
134            The `User` object that was requested
135        """
136        response = await (
137            await self._http.request("GET", f"/g/s/user-profile/{userId}")
138        ).json()
139
140        if response.get("api:statuscode") != 0:
141            return exceptions.handle_exception(response.get("api:statuscode"), response)
142        return User(**(response["userProfile"]), client=self)
143
144    async def fetch_member(self, userId: str, ndcId: str) -> Member:
145        """Fetches a member for a given community
146
147        Parameters
148        ----------
149        userId : str
150            The userId to search for
151        ndcId : str
152            The community the user is a member in
153
154        Returns
155        -------
156        Member
157            The `Member` object requested for the given community
158        """
159        response = await (
160            await self._http.request("GET", f"/x{ndcId}/s/user-profile/{userId}")
161        ).json()
162
163        if response.get("api:statuscode") != 0:
164            return exceptions.handle_exception(response.get("api:statuscode"), response)
165
166        return Member(**(response["userProfile"]), client=self)
167
168    async def send_message(
169        self,
170        threadId: str,
171        content: str,
172        *,
173        ndcId: Optional[str] = "",
174        embed: Optional[Embed] = Embed(None, None, None, None, None, None),
175        **kwargs,
176    ) -> Message:
177        """Sends a message to a given Thread.
178
179        Parameters
180        ----------
181        threadId : str
182            the thread to send the message to
183        content : str
184            the content of the message to send, must be within 2000 characters
185        ndcId : Optional[str], optional
186            The community the Thread is in, if not given (or 0) it will look for a global chat, by default ""
187        embed : Optional[Embed], optional
188            Embed to send alongside the message, by default empty embed
189
190        Returns
191        -------
192        Message
193            Returns the `Message` object of the sent message
194        """
195        response = await (
196            await self._http.request(
197                "POST",
198                f"/x{ndcId}/s/chat/thread/{threadId}/message"
199                if ndcId
200                else f"/g/s/chat/thread/{threadId}",
201                json={
202                    "type": 0,
203                    "content": content,
204                    "clientRefId": int(time() % 86400),
205                    "timestamp": int(time() * 1000),
206                    "attachedObject": embed.__dict__(),
207                    **kwargs,
208                },
209            )
210        ).json()
211
212        if response.get("api:statuscode") != 0:
213            return exceptions.handle_exception(response.get("api:statuscode"), response)
214        return Message(**(response["message"]), client=self)
215
216    async def start_dm(self, userId: str, *, ndcId: Optional[str] = "") -> Thread:
217        """Start direct messaging a user or return the Thread if a DM already exists
218
219        Parameters
220        ----------
221        userId : str
222            the user you want to start a DM with
223        ndcId : Optional[str], optional
224            the community the member is in, if sending to a user, this  will be empty
225
226        Returns
227        -------
228        Thread
229            Thread of the DMs
230        """
231        response = await (
232            await self._http.request(
233                "POST",
234                f"/x{ndcId}/s/chat/thread/" if ndcId else f"/g/s/chat/thread/",
235                json={
236                    "title": None,
237                    "content": None,
238                    "initialMessageContent": None,
239                    "timestamp": int(time() * 1000),
240                    "inviteeUids": [userId],
241                    "type": 0,
242                },
243            )
244        ).json()
245
246        if response.get("api:statuscode") != 0:
247            return exceptions.handle_exception(response.get("api:statuscode"), response)
248
249        return Thread(**(response["thread"]), client=self)
250
251    async def message_user(self, userId: str, **kwargs) -> Message:
252        """Send a message to a user's DMs, this starts the DMs (via `start_dm()`) if they don't exist already
253
254        Parameters
255        ----------
256        userId : str
257            userId to send a message to
258
259        Returns
260        -------
261        Message
262            Return an object representing the sent message
263        """
264        return await (
265            await self.start_dm(userId=userId, ndcId=kwargs.pop("ndcId", ""))
266        ).send(**kwargs)
267
268    async def upload_image(self, image: Union[bytes, BinaryIO, PathLike]) -> str:
269        """Upload an image to the amino servers
270
271        Parameters
272        ----------
273        image : Union[bytes, BinaryIO, PathLike]
274            Either the read out image, an IO object representing the Image, or the image path
275
276        Returns
277        -------
278        str
279            The direct link of the image
280        """
281        if isinstance(image, (BinaryIO, PathLike)):
282            kwargs = {"file": image}
283        else:
284            kwargs = {"data": image}
285        response = await (
286            await self._http.request("POST", "/g/s/media/upload", **kwargs)
287        ).json()
288
289        if response.get("api:statuscode") != 0:
290            return exceptions.handle_exception(response.get("api:statuscode"), response)
291
292        return response["mediaValue"]
293
294    async def set_device(self, ndcId: str = "") -> Optional[dict]:
295        """Set the device pushToken to receive notifications on the websocket
296
297        Parameters
298        ----------
299        ndcId : str, optional
300            Optionally a community to set the token in, by default ""
301
302        Returns
303        -------
304        Optional[dict]
305            A dictionary containing devOptions, returns None if it doesn't exist
306        """
307        response = await (
308            await self._http.request(
309                "POST",
310                f"/x{ndcId}/s/device" if ndcId else "/g/s/device",
311                json={
312                    "deviceID": self._http.device,
313                    "bundleID": "com.narvii.amino.master",
314                    "clientType": 100,
315                    "timezone": 60,
316                    "systemPushEnabled": True,
317                    "locale": "en_DE",
318                    "deviceToken": self._http.token,
319                    "deviceTokenType": 1,
320                    "timestamp": int(time() * 1000),
321                },
322            )
323        ).json()
324        if response.get("api:statuscode") != 0:
325            return exceptions.handle_exception(response.get("api:statuscode"), response)
326
327        return response["devOptions"]
328
329    async def fetch_communities(
330        self, start: int = 0, size: int = 25
331    ) -> List[Community]:
332        """Fetch a list of communities that the bot is in, this can be used to set tokens for each community
333
334        Parameters
335        ----------
336        start : int, optional
337            start index, by default 0
338        size : int, optional
339            amount of communities to fetch, by default 25
340
341        Returns
342        -------
343        List[Community]
344            List of `Community` objects describing the communities
345        """
346        response = await (
347            await self._http.request(
348                "GET", "/g/s/community/joined", params={"start": start, "size": size}
349            )
350        ).json()
351
352        if response.get("api:statuscode") != 0:
353            return exceptions.handle_exception(response.get("api:statuscode"), response)
354        return [
355            Community(**community, client=self)
356            for community in response["communityList"]
357        ]
358
359    async def fetch_blogs(
360        self, ndcId: str, start: int = 0, size: int = 25, *, userId: Optional[str] = ""
361    ) -> List[Blog]:
362        """Fetches blogs by user or community if no user is supplied
363
364        Parameters
365        ----------
366        ndcId : Optional[str], optional
367            the community id, by default ""
368        start : int, optional
369            offset to start at, by default 0
370        size : int, optional
371            amount of blogs to get, by default 25
372        userId : Optional[str], optional
373            the id of the user, by default ""
374        """
375        params = {
376            "start": start,
377            "size": size,
378            "url": f"/x{ndcId}/s/blog" if userId else f"/x{ndcId}/s/feed/blog-all",
379            **({"type": "user", "q": userId} if userId else {}),
380        }
381        response = await (
382            await self._http.request("GET", params.pop("url"), params=params)
383        ).json()
384
385        if response.get("api:statuscode") != 0:
386            return exceptions.handle_exception(response.get("api:statuscode"), response)
387        return [Blog(**blog, client=self) for blog in response["blogList"]]
388
389    async def tip_blog(self, ndcId: str, blogId: str, amount: int):
390        """Sends coins to a blog
391
392        Parameters
393        ----------
394        ndcId : str
395            community id that the blog is in
396        blogId : str
397            id of the blog
398        amount : int
399            the amount of coins to send
400        """
401        response = await (
402            await self._http.request(
403                "POST",
404                f"/x{ndcId}/s/blog/{blogId}/tipping",
405                json={
406                    "coins": amount,
407                    "tippingContext": {"transactionId": str(uuid4())},
408                    "timestamp": int(time() * 1000),
409                },
410            )
411        ).json()
412        if response.get("api:statuscode") != 0:
413            return exceptions.handle_exception(response.get("api:statuscode"), response)

ApiClient skeleton to reduce repeating API calls in code and to clean up code

async def login(self, email: str, password: str) -> aminoacid.abc.User: 
24    async def login(self, email: str, password: str) -> User:
25        """Authenticates with the given email and password
26
27        Parameters
28        ----------
29        email : str
30            Email to login with
31        password : str
32            Password to login with
33
34        Returns
35        -------
36        User
37            The user that was authenticated
38        """
39        response = await (
40            await self._http.request(
41                "POST",
42                "/g/s/auth/login",
43                json={
44                    "email": email,
45                    "v": 2,
46                    "secret": f"0 {password}",
47                    "deviceID": self._http.device,
48                    "clientType": 100,
49                    "action": "normal",
50                    "timestamp": int(time() * 1000),
51                },
52            )
53        ).json()
54
55        if response.get("api:statuscode") != 0:
56            return exceptions.handle_exception(response.get("api:statuscode"), response)
57
58        self.profile = User(**(response["userProfile"]), client=self)
59        self._http.session = Session(response.get("sid"))
60
61        return self.profile

Authenticates with the given email and password

Parameters
  • email (str): Email to login with
  • password (str): Password to login with
Returns
  • User: The user that was authenticated
async def fetch_thread(self, threadId: str, ndcId: Optional[str] = '') -> aminoacid.abc.Thread: 
63    async def fetch_thread(self, threadId: str, ndcId: Optional[str] = "") -> Thread:
64        """Fetches a given `Thread`
65
66        Parameters
67        ----------
68        threadId : str
69            the ID of the thread
70        ndcId : Optional[str], optional
71            the community the thread is in, if not given (or 0) it will look for the thread in global, by default ""
72
73        Returns
74        -------
75        Thread
76            The `Thread` object that was requested
77        """
78        response = await (
79            await self._http.request(
80                "GET",
81                f"/x{ndcId}/s/chat/thread/{threadId}"
82                if ndcId
83                else f"/g/s/chat/thread/{threadId}",
84            )
85        ).json()
86
87        if response.get("api:statuscode") != 0:
88            return exceptions.handle_exception(response.get("api:statuscode"), response)
89        return Thread(**(response["thread"]), client=self)

Fetches a given Thread

Parameters
  • threadId (str): the ID of the thread
  • ndcId (Optional[str], optional): the community the thread is in, if not given (or 0) it will look for the thread in global, by default ""
Returns
  • Thread: The Thread object that was requested
async def fetch_message( self, messageId: str, threadId: str, ndcId: Optional[str] = '') -> aminoacid.abc.Message: 
 91    async def fetch_message(
 92        self, messageId: str, threadId: str, ndcId: Optional[str] = ""
 93    ) -> Message:
 94        """Fetches a given `Message` from a `Thread`
 95
 96        Parameters
 97        ----------
 98        messageId : str
 99            the ID of the `Message` to fetch
100        threadId : str
101            the ID of the `Thread` to fetch from
102        ndcId : Optional[str], optional
103            the community the thread is in, if not given (or 0) it will look for the thread in global, by default, by default ""
104
105        Returns
106        -------
107        Message
108            The `Message` object that was requested
109        """
110        response = await (
111            await self._http.request(
112                "GET",
113                f"/x{ndcId}/s/chat/thread/{threadId}/message/{messageId}"
114                if ndcId
115                else f"/g/s/chat/thread/{threadId}/message/{messageId}",
116            )
117        ).json()
118
119        if response.get("api:statuscode") != 0:
120            return exceptions.handle_exception(response.get("api:statuscode"), response)
121        return Thread(**(response["message"]), client=self)

Fetches a given Message from a Thread

Parameters
  • messageId (str): the ID of the Message to fetch
  • threadId (str): the ID of the Thread to fetch from
  • ndcId (Optional[str], optional): the community the thread is in, if not given (or 0) it will look for the thread in global, by default, by default ""
Returns
  • Message: The Message object that was requested
async def fetch_user(self, userId: str) -> aminoacid.abc.User: 
123    async def fetch_user(self, userId: str) -> User:
124        """Fetches a user from the API
125
126        Parameters
127        ----------
128        userId : str
129            the userId to search for
130
131        Returns
132        -------
133        User
134            The `User` object that was requested
135        """
136        response = await (
137            await self._http.request("GET", f"/g/s/user-profile/{userId}")
138        ).json()
139
140        if response.get("api:statuscode") != 0:
141            return exceptions.handle_exception(response.get("api:statuscode"), response)
142        return User(**(response["userProfile"]), client=self)

Fetches a user from the API

Parameters
  • userId (str): the userId to search for
Returns
  • User: The User object that was requested
async def fetch_member(self, userId: str, ndcId: str) -> aminoacid.abc.Member: 
144    async def fetch_member(self, userId: str, ndcId: str) -> Member:
145        """Fetches a member for a given community
146
147        Parameters
148        ----------
149        userId : str
150            The userId to search for
151        ndcId : str
152            The community the user is a member in
153
154        Returns
155        -------
156        Member
157            The `Member` object requested for the given community
158        """
159        response = await (
160            await self._http.request("GET", f"/x{ndcId}/s/user-profile/{userId}")
161        ).json()
162
163        if response.get("api:statuscode") != 0:
164            return exceptions.handle_exception(response.get("api:statuscode"), response)
165
166        return Member(**(response["userProfile"]), client=self)

Fetches a member for a given community

Parameters
  • userId (str): The userId to search for
  • ndcId (str): The community the user is a member in
Returns
  • Member: The Member object requested for the given community
async def send_message( self, threadId: str, content: str, *, ndcId: Optional[str] = '', embed: Optional[aminoacid.abc.Embed] = Embed(), **kwargs) -> aminoacid.abc.Message: 
168    async def send_message(
169        self,
170        threadId: str,
171        content: str,
172        *,
173        ndcId: Optional[str] = "",
174        embed: Optional[Embed] = Embed(None, None, None, None, None, None),
175        **kwargs,
176    ) -> Message:
177        """Sends a message to a given Thread.
178
179        Parameters
180        ----------
181        threadId : str
182            the thread to send the message to
183        content : str
184            the content of the message to send, must be within 2000 characters
185        ndcId : Optional[str], optional
186            The community the Thread is in, if not given (or 0) it will look for a global chat, by default ""
187        embed : Optional[Embed], optional
188            Embed to send alongside the message, by default empty embed
189
190        Returns
191        -------
192        Message
193            Returns the `Message` object of the sent message
194        """
195        response = await (
196            await self._http.request(
197                "POST",
198                f"/x{ndcId}/s/chat/thread/{threadId}/message"
199                if ndcId
200                else f"/g/s/chat/thread/{threadId}",
201                json={
202                    "type": 0,
203                    "content": content,
204                    "clientRefId": int(time() % 86400),
205                    "timestamp": int(time() * 1000),
206                    "attachedObject": embed.__dict__(),
207                    **kwargs,
208                },
209            )
210        ).json()
211
212        if response.get("api:statuscode") != 0:
213            return exceptions.handle_exception(response.get("api:statuscode"), response)
214        return Message(**(response["message"]), client=self)

Sends a message to a given Thread.

Parameters
  • threadId (str): the thread to send the message to
  • content (str): the content of the message to send, must be within 2000 characters
  • ndcId (Optional[str], optional): The community the Thread is in, if not given (or 0) it will look for a global chat, by default ""
  • embed (Optional[Embed], optional): Embed to send alongside the message, by default empty embed
Returns
  • Message: Returns the Message object of the sent message
async def start_dm(self, userId: str, *, ndcId: Optional[str] = '') -> aminoacid.abc.Thread: 
216    async def start_dm(self, userId: str, *, ndcId: Optional[str] = "") -> Thread:
217        """Start direct messaging a user or return the Thread if a DM already exists
218
219        Parameters
220        ----------
221        userId : str
222            the user you want to start a DM with
223        ndcId : Optional[str], optional
224            the community the member is in, if sending to a user, this  will be empty
225
226        Returns
227        -------
228        Thread
229            Thread of the DMs
230        """
231        response = await (
232            await self._http.request(
233                "POST",
234                f"/x{ndcId}/s/chat/thread/" if ndcId else f"/g/s/chat/thread/",
235                json={
236                    "title": None,
237                    "content": None,
238                    "initialMessageContent": None,
239                    "timestamp": int(time() * 1000),
240                    "inviteeUids": [userId],
241                    "type": 0,
242                },
243            )
244        ).json()
245
246        if response.get("api:statuscode") != 0:
247            return exceptions.handle_exception(response.get("api:statuscode"), response)
248
249        return Thread(**(response["thread"]), client=self)

Start direct messaging a user or return the Thread if a DM already exists

Parameters
  • userId (str): the user you want to start a DM with
  • ndcId (Optional[str], optional): the community the member is in, if sending to a user, this will be empty
Returns
  • Thread: Thread of the DMs
async def message_user(self, userId: str, **kwargs) -> aminoacid.abc.Message: 
251    async def message_user(self, userId: str, **kwargs) -> Message:
252        """Send a message to a user's DMs, this starts the DMs (via `start_dm()`) if they don't exist already
253
254        Parameters
255        ----------
256        userId : str
257            userId to send a message to
258
259        Returns
260        -------
261        Message
262            Return an object representing the sent message
263        """
264        return await (
265            await self.start_dm(userId=userId, ndcId=kwargs.pop("ndcId", ""))
266        ).send(**kwargs)

Send a message to a user's DMs, this starts the DMs (via start_dm()) if they don't exist already

Parameters
  • userId (str): userId to send a message to
Returns
  • Message: Return an object representing the sent message
async def upload_image(self, image: Union[bytes, BinaryIO, os.PathLike]) -> str: 
268    async def upload_image(self, image: Union[bytes, BinaryIO, PathLike]) -> str:
269        """Upload an image to the amino servers
270
271        Parameters
272        ----------
273        image : Union[bytes, BinaryIO, PathLike]
274            Either the read out image, an IO object representing the Image, or the image path
275
276        Returns
277        -------
278        str
279            The direct link of the image
280        """
281        if isinstance(image, (BinaryIO, PathLike)):
282            kwargs = {"file": image}
283        else:
284            kwargs = {"data": image}
285        response = await (
286            await self._http.request("POST", "/g/s/media/upload", **kwargs)
287        ).json()
288
289        if response.get("api:statuscode") != 0:
290            return exceptions.handle_exception(response.get("api:statuscode"), response)
291
292        return response["mediaValue"]

Upload an image to the amino servers

Parameters
  • image (Union[bytes, BinaryIO, PathLike]): Either the read out image, an IO object representing the Image, or the image path
Returns
  • str: The direct link of the image
async def set_device(self, ndcId: str = '') -> Optional[dict]: 
294    async def set_device(self, ndcId: str = "") -> Optional[dict]:
295        """Set the device pushToken to receive notifications on the websocket
296
297        Parameters
298        ----------
299        ndcId : str, optional
300            Optionally a community to set the token in, by default ""
301
302        Returns
303        -------
304        Optional[dict]
305            A dictionary containing devOptions, returns None if it doesn't exist
306        """
307        response = await (
308            await self._http.request(
309                "POST",
310                f"/x{ndcId}/s/device" if ndcId else "/g/s/device",
311                json={
312                    "deviceID": self._http.device,
313                    "bundleID": "com.narvii.amino.master",
314                    "clientType": 100,
315                    "timezone": 60,
316                    "systemPushEnabled": True,
317                    "locale": "en_DE",
318                    "deviceToken": self._http.token,
319                    "deviceTokenType": 1,
320                    "timestamp": int(time() * 1000),
321                },
322            )
323        ).json()
324        if response.get("api:statuscode") != 0:
325            return exceptions.handle_exception(response.get("api:statuscode"), response)
326
327        return response["devOptions"]

Set the device pushToken to receive notifications on the websocket

Parameters
  • ndcId (str, optional): Optionally a community to set the token in, by default ""
Returns
  • Optional[dict]: A dictionary containing devOptions, returns None if it doesn't exist
async def fetch_communities(self, start: int = 0, size: int = 25) -> List[aminoacid.abc.Community]: 
329    async def fetch_communities(
330        self, start: int = 0, size: int = 25
331    ) -> List[Community]:
332        """Fetch a list of communities that the bot is in, this can be used to set tokens for each community
333
334        Parameters
335        ----------
336        start : int, optional
337            start index, by default 0
338        size : int, optional
339            amount of communities to fetch, by default 25
340
341        Returns
342        -------
343        List[Community]
344            List of `Community` objects describing the communities
345        """
346        response = await (
347            await self._http.request(
348                "GET", "/g/s/community/joined", params={"start": start, "size": size}
349            )
350        ).json()
351
352        if response.get("api:statuscode") != 0:
353            return exceptions.handle_exception(response.get("api:statuscode"), response)
354        return [
355            Community(**community, client=self)
356            for community in response["communityList"]
357        ]

Fetch a list of communities that the bot is in, this can be used to set tokens for each community

Parameters
  • start (int, optional): start index, by default 0
  • size (int, optional): amount of communities to fetch, by default 25
Returns
  • List[Community]: List of Community objects describing the communities
async def fetch_blogs( self, ndcId: str, start: int = 0, size: int = 25, *, userId: Optional[str] = '') -> List[aminoacid.abc.Blog]: 
359    async def fetch_blogs(
360        self, ndcId: str, start: int = 0, size: int = 25, *, userId: Optional[str] = ""
361    ) -> List[Blog]:
362        """Fetches blogs by user or community if no user is supplied
363
364        Parameters
365        ----------
366        ndcId : Optional[str], optional
367            the community id, by default ""
368        start : int, optional
369            offset to start at, by default 0
370        size : int, optional
371            amount of blogs to get, by default 25
372        userId : Optional[str], optional
373            the id of the user, by default ""
374        """
375        params = {
376            "start": start,
377            "size": size,
378            "url": f"/x{ndcId}/s/blog" if userId else f"/x{ndcId}/s/feed/blog-all",
379            **({"type": "user", "q": userId} if userId else {}),
380        }
381        response = await (
382            await self._http.request("GET", params.pop("url"), params=params)
383        ).json()
384
385        if response.get("api:statuscode") != 0:
386            return exceptions.handle_exception(response.get("api:statuscode"), response)
387        return [Blog(**blog, client=self) for blog in response["blogList"]]

Fetches blogs by user or community if no user is supplied

Parameters
  • ndcId (Optional[str], optional): the community id, by default ""
  • start (int, optional): offset to start at, by default 0
  • size (int, optional): amount of blogs to get, by default 25
  • userId (Optional[str], optional): the id of the user, by default ""
async def tip_blog(self, ndcId: str, blogId: str, amount: int): 
389    async def tip_blog(self, ndcId: str, blogId: str, amount: int):
390        """Sends coins to a blog
391
392        Parameters
393        ----------
394        ndcId : str
395            community id that the blog is in
396        blogId : str
397            id of the blog
398        amount : int
399            the amount of coins to send
400        """
401        response = await (
402            await self._http.request(
403                "POST",
404                f"/x{ndcId}/s/blog/{blogId}/tipping",
405                json={
406                    "coins": amount,
407                    "tippingContext": {"transactionId": str(uuid4())},
408                    "timestamp": int(time() * 1000),
409                },
410            )
411        ).json()
412        if response.get("api:statuscode") != 0:
413            return exceptions.handle_exception(response.get("api:statuscode"), response)

Sends coins to a blog

Parameters
  • ndcId (str): community id that the blog is in
  • blogId (str): id of the blog
  • amount (int): the amount of coins to send
Inherited Members
aminoacid.abc.AminoBaseClass
AminoBaseClass