More moving functions out

pull/290/head
halcy 2022-12-01 00:11:17 +02:00
rodzic d9cd7547fd
commit a4b2b180d3
5 zmienionych plików z 256 dodań i 229 usunięć

Wyświetl plik

@ -1,5 +1,18 @@
version: 2.1
jobs:
run-tests-36:
docker:
- image: cimg/python:3.6
steps:
- checkout
- run:
name: "Install test deps"
command: "pip install .[test]"
- run:
name: "Run tests"
command: "python setup.py pytest --addopts '--junitxml=tests/result.xml'"
- store_test_results:
path: tests
run-tests-37:
docker:
- image: cimg/python:3.7
@ -8,9 +21,6 @@ jobs:
- run:
name: "Install test deps"
command: "pip install .[test]"
- run:
name: "Install codecov"
command: "pip install codecov"
- run:
name: "Run tests"
command: "python setup.py pytest --addopts '--junitxml=tests/result.xml'"
@ -42,10 +52,7 @@ jobs:
- checkout
- run:
name: "Install test deps"
command: "pip install .[test]"
- run:
name: "Install codecov"
command: "pip install codecov"
command: "pip install .[test]"
- run:
name: "Run tests"
command: "python setup.py pytest --addopts '--junitxml=tests/result.xml'"
@ -58,10 +65,7 @@ jobs:
- checkout
- run:
name: "Install test deps"
command: "pip install .[test]"
- run:
name: "Install codecov"
command: "pip install codecov"
command: "pip install .[test]"
- run:
name: "Run tests"
command: "python setup.py pytest --addopts '--junitxml=tests/result.xml'"
@ -70,6 +74,7 @@ jobs:
workflows:
run-tests-workflow:
jobs:
- run-tests-36
- run-tests-37
- run-tests-38-cov
- run-tests-39

Wyświetl plik

@ -43,11 +43,13 @@ from .internals import Mastodon as Internals
from .authentication import Mastodon as Authentication
from .accounts import Mastodon as Accounts
from .instance import Mastodon as Instance
from .timeline import Mastodon as Timeline
from .statuses import Mastodon as Statuses
##
# The actual Mastodon class
###
class Mastodon(Utility, Authentication, Accounts, Instance):
class Mastodon(Utility, Authentication, Accounts, Instance, Timeline, Statuses):
"""
Thorough and easy to use Mastodon
API wrapper in Python.
@ -64,221 +66,6 @@ class Mastodon(Utility, Authentication, Accounts, Instance):
"""
return Mastodon.__SUPPORTED_MASTODON_VERSION
###
# Reading data: Timelines
##
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline(self, timeline="home", max_id=None, min_id=None, since_id=None, limit=None, only_media=False, local=False, remote=False):
"""
Fetch statuses, most recent ones first. `timeline` can be 'home', 'local', 'public',
'tag/hashtag' or 'list/id'. See the following functions documentation for what those do.
The default timeline is the "home" timeline.
Specify `only_media` to only get posts with attached media. Specify `local` to only get local statuses,
and `remote` to only get remote statuses. Some options are mutually incompatible as dictated by logic.
May or may not require authentication depending on server settings and what is specifically requested.
Returns a list of :ref:`status dicts <status dicts>`.
"""
if max_id is not None:
max_id = self.__unpack_id(max_id, dateconv=True)
if min_id is not None:
min_id = self.__unpack_id(min_id, dateconv=True)
if since_id is not None:
since_id = self.__unpack_id(since_id, dateconv=True)
params_initial = locals()
if not local:
del params_initial['local']
if not remote:
del params_initial['remote']
if not only_media:
del params_initial['only_media']
if timeline == "local":
timeline = "public"
params_initial['local'] = True
params = self.__generate_params(params_initial, ['timeline'])
url = '/api/v1/timelines/{0}'.format(timeline)
return self.__api_request('GET', url, params)
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_home(self, max_id=None, min_id=None, since_id=None, limit=None, only_media=False, local=False, remote=False):
"""
Convenience method: Fetches the logged-in user's home timeline (i.e. followed users and self). Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
return self.timeline('home', max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote)
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_local(self, max_id=None, min_id=None, since_id=None, limit=None, only_media=False):
"""
Convenience method: Fetches the local / instance-wide timeline, not including replies. Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
return self.timeline('local', max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media)
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_public(self, max_id=None, min_id=None, since_id=None, limit=None, only_media=False, local=False, remote=False):
"""
Convenience method: Fetches the public / visible-network / federated timeline, not including replies. Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
return self.timeline('public', max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote)
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_hashtag(self, hashtag, local=False, max_id=None, min_id=None, since_id=None, limit=None, only_media=False, remote=False):
"""
Convenience method: Fetch a timeline of toots with a given hashtag. The hashtag parameter
should not contain the leading #. Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
if hashtag.startswith("#"):
raise MastodonIllegalArgumentError(
"Hashtag parameter should omit leading #")
return self.timeline('tag/{0}'.format(hashtag), max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote)
@api_version("2.1.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_list(self, id, max_id=None, min_id=None, since_id=None, limit=None, only_media=False, local=False, remote=False):
"""
Convenience method: Fetches a timeline containing all the toots by users in a given list. Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
id = self.__unpack_id(id)
return self.timeline('list/{0}'.format(id), max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote)
@api_version("2.6.0", "2.6.0", _DICT_VERSION_CONVERSATION)
def conversations(self, max_id=None, min_id=None, since_id=None, limit=None):
"""
Fetches a user's conversations.
Returns a list of :ref:`conversation dicts <conversation dicts>`.
"""
if max_id is not None:
max_id = self.__unpack_id(max_id, dateconv=True)
if min_id is not None:
min_id = self.__unpack_id(min_id, dateconv=True)
if since_id is not None:
since_id = self.__unpack_id(since_id, dateconv=True)
params = self.__generate_params(locals())
return self.__api_request('GET', "/api/v1/conversations/", params)
###
# Reading data: Statuses
###
@api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS)
def status(self, id):
"""
Fetch information about a single toot.
Does not require authentication for publicly visible statuses.
Returns a :ref:`status dict <status dict>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "3.0.0", _DICT_VERSION_CARD)
def status_card(self, id):
"""
Fetch a card associated with a status. A card describes an object (such as an
external video or link) embedded into a status.
Does not require authentication for publicly visible statuses.
This function is deprecated as of 3.0.0 and the endpoint does not
exist anymore - you should just use the "card" field of the status dicts
instead. Mastodon.py will try to mimic the old behaviour, but this
is somewhat inefficient and not guaranteed to be the case forever.
Returns a :ref:`card dict <card dict>`.
"""
if self.verify_minimum_version("3.0.0", cached=True):
return self.status(id).card
else:
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}/card'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "1.0.0", _DICT_VERSION_CONTEXT)
def status_context(self, id):
"""
Fetch information about ancestors and descendants of a toot.
Does not require authentication for publicly visible statuses.
Returns a :ref:`context dict <context dict>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}/context'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT)
def status_reblogged_by(self, id):
"""
Fetch a list of users that have reblogged a status.
Does not require authentication for publicly visible statuses.
Returns a list of :ref:`account dicts <account dicts>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}/reblogged_by'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT)
def status_favourited_by(self, id):
"""
Fetch a list of users that have favourited a status.
Does not require authentication for publicly visible statuses.
Returns a list of :ref:`account dicts <account dicts>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}/favourited_by'.format(str(id))
return self.__api_request('GET', url)
###
# Reading data: Scheduled statuses
###
@api_version("2.7.0", "2.7.0", _DICT_VERSION_SCHEDULED_STATUS)
def scheduled_statuses(self):
"""
Fetch a list of scheduled statuses
Returns a list of :ref:`scheduled status dicts <scheduled status dicts>`.
"""
return self.__api_request('GET', '/api/v1/scheduled_statuses')
@api_version("2.7.0", "2.7.0", _DICT_VERSION_SCHEDULED_STATUS)
def scheduled_status(self, id):
"""
Fetch information about the scheduled status with the given id.
Returns a :ref:`scheduled status dict <scheduled status dict>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/scheduled_statuses/{0}'.format(str(id))
return self.__api_request('GET', url)
###
# Reading data: Polls
###

Wyświetl plik

@ -0,0 +1,108 @@
from .versions import _DICT_VERSION_STATUS, _DICT_VERSION_CARD, _DICT_VERSION_CONTEXT, _DICT_VERSION_ACCOUNT, _DICT_VERSION_SCHEDULED_STATUS
from .utility import api_version
from .internals import Mastodon as Internals
class Mastodon(Internals):
###
# Reading data: Statuses
###
@api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS)
def status(self, id):
"""
Fetch information about a single toot.
Does not require authentication for publicly visible statuses.
Returns a :ref:`status dict <status dict>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "3.0.0", _DICT_VERSION_CARD)
def status_card(self, id):
"""
Fetch a card associated with a status. A card describes an object (such as an
external video or link) embedded into a status.
Does not require authentication for publicly visible statuses.
This function is deprecated as of 3.0.0 and the endpoint does not
exist anymore - you should just use the "card" field of the status dicts
instead. Mastodon.py will try to mimic the old behaviour, but this
is somewhat inefficient and not guaranteed to be the case forever.
Returns a :ref:`card dict <card dict>`.
"""
if self.verify_minimum_version("3.0.0", cached=True):
return self.status(id).card
else:
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}/card'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "1.0.0", _DICT_VERSION_CONTEXT)
def status_context(self, id):
"""
Fetch information about ancestors and descendants of a toot.
Does not require authentication for publicly visible statuses.
Returns a :ref:`context dict <context dict>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}/context'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT)
def status_reblogged_by(self, id):
"""
Fetch a list of users that have reblogged a status.
Does not require authentication for publicly visible statuses.
Returns a list of :ref:`account dicts <account dicts>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}/reblogged_by'.format(str(id))
return self.__api_request('GET', url)
@api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT)
def status_favourited_by(self, id):
"""
Fetch a list of users that have favourited a status.
Does not require authentication for publicly visible statuses.
Returns a list of :ref:`account dicts <account dicts>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/statuses/{0}/favourited_by'.format(str(id))
return self.__api_request('GET', url)
###
# Reading data: Scheduled statuses
###
@api_version("2.7.0", "2.7.0", _DICT_VERSION_SCHEDULED_STATUS)
def scheduled_statuses(self):
"""
Fetch a list of scheduled statuses
Returns a list of :ref:`scheduled status dicts <scheduled status dicts>`.
"""
return self.__api_request('GET', '/api/v1/scheduled_statuses')
@api_version("2.7.0", "2.7.0", _DICT_VERSION_SCHEDULED_STATUS)
def scheduled_status(self, id):
"""
Fetch information about the scheduled status with the given id.
Returns a :ref:`scheduled status dict <scheduled status dict>`.
"""
id = self.__unpack_id(id)
url = '/api/v1/scheduled_statuses/{0}'.format(str(id))
return self.__api_request('GET', url)

Wyświetl plik

@ -0,0 +1,121 @@
from .versions import _DICT_VERSION_STATUS, _DICT_VERSION_CONVERSATION
from .error import MastodonIllegalArgumentError, MastodonNotFoundError
from .utility import api_version
from .internals import Mastodon as Internals
class Mastodon(Internals):
###
# Reading data: Timelines
##
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline(self, timeline="home", max_id=None, min_id=None, since_id=None, limit=None, only_media=False, local=False, remote=False):
"""
Fetch statuses, most recent ones first. `timeline` can be 'home', 'local', 'public',
'tag/hashtag' or 'list/id'. See the following functions documentation for what those do.
The default timeline is the "home" timeline.
Specify `only_media` to only get posts with attached media. Specify `local` to only get local statuses,
and `remote` to only get remote statuses. Some options are mutually incompatible as dictated by logic.
May or may not require authentication depending on server settings and what is specifically requested.
Returns a list of :ref:`status dicts <status dicts>`.
"""
if max_id is not None:
max_id = self.__unpack_id(max_id, dateconv=True)
if min_id is not None:
min_id = self.__unpack_id(min_id, dateconv=True)
if since_id is not None:
since_id = self.__unpack_id(since_id, dateconv=True)
params_initial = locals()
if not local:
del params_initial['local']
if not remote:
del params_initial['remote']
if not only_media:
del params_initial['only_media']
if timeline == "local":
timeline = "public"
params_initial['local'] = True
params = self.__generate_params(params_initial, ['timeline'])
url = '/api/v1/timelines/{0}'.format(timeline)
return self.__api_request('GET', url, params)
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_home(self, max_id=None, min_id=None, since_id=None, limit=None, only_media=False, local=False, remote=False):
"""
Convenience method: Fetches the logged-in user's home timeline (i.e. followed users and self). Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
return self.timeline('home', max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote)
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_local(self, max_id=None, min_id=None, since_id=None, limit=None, only_media=False):
"""
Convenience method: Fetches the local / instance-wide timeline, not including replies. Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
return self.timeline('local', max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media)
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_public(self, max_id=None, min_id=None, since_id=None, limit=None, only_media=False, local=False, remote=False):
"""
Convenience method: Fetches the public / visible-network / federated timeline, not including replies. Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
return self.timeline('public', max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote)
@api_version("1.0.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_hashtag(self, hashtag, local=False, max_id=None, min_id=None, since_id=None, limit=None, only_media=False, remote=False):
"""
Convenience method: Fetch a timeline of toots with a given hashtag. The hashtag parameter
should not contain the leading #. Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
if hashtag.startswith("#"):
raise MastodonIllegalArgumentError(
"Hashtag parameter should omit leading #")
return self.timeline('tag/{0}'.format(hashtag), max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote)
@api_version("2.1.0", "3.1.4", _DICT_VERSION_STATUS)
def timeline_list(self, id, max_id=None, min_id=None, since_id=None, limit=None, only_media=False, local=False, remote=False):
"""
Convenience method: Fetches a timeline containing all the toots by users in a given list. Params as in `timeline()`.
Returns a list of :ref:`status dicts <status dicts>`.
"""
id = self.__unpack_id(id)
return self.timeline('list/{0}'.format(id), max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote)
@api_version("2.6.0", "2.6.0", _DICT_VERSION_CONVERSATION)
def conversations(self, max_id=None, min_id=None, since_id=None, limit=None):
"""
Fetches a user's conversations.
Returns a list of :ref:`conversation dicts <conversation dicts>`.
"""
if max_id is not None:
max_id = self.__unpack_id(max_id, dateconv=True)
if min_id is not None:
min_id = self.__unpack_id(min_id, dateconv=True)
if since_id is not None:
since_id = self.__unpack_id(since_id, dateconv=True)
params = self.__generate_params(locals())
return self.__api_request('GET', "/api/v1/conversations/", params)

Wyświetl plik

@ -1,7 +1,13 @@
import pytest
from mastodon.Mastodon import MastodonAPIError, MastodonNotFoundError
import datetime
import zoneinfo
try:
import zoneinfo
timezone = zoneinfo.ZoneInfo
except:
import pytz
timezone = pytz.timezone
import vcr
import time
import pickle
@ -154,7 +160,7 @@ def test_status_pin_unpin(status, api):
@pytest.mark.vcr(match_on=['path'])
def test_scheduled_status(api):
base_time = datetime.datetime(4000, 1, 1, 12, 13, 14, 0, zoneinfo.ZoneInfo("Etc/GMT+2"))
base_time = datetime.datetime(4000, 1, 1, 12, 13, 14, 0, timezone("Etc/GMT+2"))
the_future = base_time + datetime.timedelta(minutes=20)
scheduled_toot = api.status_post("please ensure adequate headroom", scheduled_at=the_future)
assert scheduled_toot