Added notificactions, me and account methods

This commit is contained in:
spla 2022-08-07 16:50:15 +02:00
pare 4d19b21940
commit 1354d25b4b

110
akkoma.py
Veure arxiu

@ -145,6 +145,7 @@ class Akkoma:
__DICT_VERSION_POLL = "2.8.0" __DICT_VERSION_POLL = "2.8.0"
__DICT_VERSION_STATUS = bigger_version(bigger_version(bigger_version(bigger_version(bigger_version("3.1.0", __DICT_VERSION_STATUS = bigger_version(bigger_version(bigger_version(bigger_version(bigger_version("3.1.0",
__DICT_VERSION_MEDIA), __DICT_VERSION_ACCOUNT), __DICT_VERSION_APPLICATION), __DICT_VERSION_MENTION), __DICT_VERSION_POLL) __DICT_VERSION_MEDIA), __DICT_VERSION_ACCOUNT), __DICT_VERSION_APPLICATION), __DICT_VERSION_MENTION), __DICT_VERSION_POLL)
__DICT_VERSION_NOTIFICATION = bigger_version(bigger_version("1.0.0", __DICT_VERSION_ACCOUNT), __DICT_VERSION_STATUS)
@staticmethod @staticmethod
def create_app(app_name, scopes=__DEFAULT_SCOPES, redirect_uris=None, website=None, to_file=None, api_base_url=__DEFAULT_BASE_URL, def create_app(app_name, scopes=__DEFAULT_SCOPES, redirect_uris=None, website=None, to_file=None, api_base_url=__DEFAULT_BASE_URL,
@ -335,6 +336,25 @@ class Akkoma:
self.akkoma_major, self.akkoma_minor, self.akkoma_patch = parse_version_string(version_str) self.akkoma_major, self.akkoma_minor, self.akkoma_patch = parse_version_string(version_str)
return version_str return version_str
def verify_minimum_version(self, version_str, cached=False):
"""
Update version info from server and verify that at least the specified version is present.
If you specify "cached", the version info update part is skipped.
Returns True if version requirement is satisfied, False if not.
"""
if not cached:
self.retrieve_akkoma_version()
major, minor, patch = parse_version_string(version_str)
if major > self.akkoma_major:
return False
elif major == self.akkoma_major and minor > self.akkoma_minor:
return False
elif major == self.akkoma_major and minor == self.akkoma_minor and patch > self.akkoma_patch:
return False
return True
def log_in(self, client_id=None, client_secret=None, grant_type=None, username=None, password=None, code=None, redirect_uri="urn:ietf:wg:oauth:2.0:oob", refresh_token=None, scopes=__DEFAULT_SCOPES, to_file=None): def log_in(self, client_id=None, client_secret=None, grant_type=None, username=None, password=None, code=None, redirect_uri="urn:ietf:wg:oauth:2.0:oob", refresh_token=None, scopes=__DEFAULT_SCOPES, to_file=None):
""" """
Get the access token for a user. Get the access token for a user.
@ -404,6 +424,72 @@ class Akkoma:
self.__logged_in_id = None self.__logged_in_id = None
return response['access_token'] return response['access_token']
###
# Reading data: Notifications
###
#@api_version("1.0.0", "2.9.0", __DICT_VERSION_NOTIFICATION)
def notifications(self, id=None, account_id=None, max_id=None, min_id=None, since_id=None, limit=None, mentions_only=None):
"""
Fetch notifications (mentions, favourites, reblogs, follows) for the logged-in
user. Pass `account_id` to get only notifications originating from the given account.
Can be passed an `id` to fetch a single notification.
Returns a list of `notification dicts`_.
"""
if max_id != None:
max_id = self.__unpack_id(max_id)
if min_id != None:
min_id = self.__unpack_id(min_id)
if since_id != None:
since_id = self.__unpack_id(since_id)
if account_id != None:
account_id = self.__unpack_id(account_id)
if id is None:
params = self.__generate_params(locals(), ['id'])
return self.__api_request('GET', '/api/v1/notifications', params)
else:
id = self.__unpack_id(id)
url = '/api/v1/notifications/{0}'.format(str(id))
return self.__api_request('GET', url)
###
# Reading data: Accounts
###
@api_version("1.0.0", "1.0.0", __DICT_VERSION_ACCOUNT)
def account(self, id):
"""
Fetch account information by user `id`.
Does not require authentication for publicly visible accounts.
Returns a `user dict`_.
"""
id = self.__unpack_id(id)
url = '/api/v1/accounts/{0}'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "2.1.0", __DICT_VERSION_ACCOUNT)
def account_verify_credentials(self):
"""
Fetch logged-in user's account information.
Returns a `user dict`_ (Starting from 2.1.0, with an additional "source" field).
"""
return self.__api_request('GET', '/api/v1/accounts/verify_credentials')
@api_version("1.0.0", "2.1.0", __DICT_VERSION_ACCOUNT)
def me(self):
"""
Get this users account. Symonym for `account_verify_credentials()`, does exactly
the same thing, just exists becase `account_verify_credentials()` has a confusing
name.
"""
return self.account_verify_credentials()
### ###
# Internal helpers, dragons probably # Internal helpers, dragons probably
### ###
@ -888,6 +974,26 @@ class Akkoma:
params = self.__generate_params(params_initial, ['idempotency_key']) params = self.__generate_params(params_initial, ['idempotency_key'])
return self.__api_request('POST', '/api/v1/statuses', params, headers = headers, use_json = use_json) return self.__api_request('POST', '/api/v1/statuses', params, headers = headers, use_json = use_json)
###
# Writing data: Notifications
###
#@api_version("1.0.0", "1.0.0", "1.0.0")
def notifications_clear(self):
"""
Clear out a users notifications
"""
self.__api_request('POST', '/api/v1/notifications/clear')
#@api_version("1.3.0", "2.9.2", "2.9.2")
def notifications_dismiss(self, id):
"""
Deletes a single notification
"""
id = self.__unpack_id(id)
url = '/api/v1/notifications/{0}/dismiss'.format(str(id))
self.__api_request('POST', url)
### ###
# Writing data: Media # Writing data: Media
### ###
@ -992,6 +1098,10 @@ class AkkomaAPIError(AkkomaError):
"""Raised when the akkoma API generates a response that cannot be handled""" """Raised when the akkoma API generates a response that cannot be handled"""
pass pass
class AkkomaNotFoundError(AkkomaAPIError):
"""Raised when the akkoma API returns a 404 Not Found error"""
pass
class AkkomaMalformedEventError(AkkomaError): class AkkomaMalformedEventError(AkkomaError):
"""Raised when the server-sent event stream is malformed""" """Raised when the server-sent event stream is malformed"""
pass pass