PK! ȭannict/__init__.py# -*- coding: utf-8 -*- """ python-annict ~~~~~~~~~~~~~~~~~~~~~ Annict API for Python. """ __title__ = 'python-annict' __version__ = '0.7.0' __author__ = 'Hiro Ashiya' __license__ = 'MIT' from .api import API # noqa PK!OUSS annict/api.py# -*- coding: utf-8 -*- from .services import APIMethod from .parsers import ModelParser from .cursors import cursor_support class API(object): """API wrapper for Annict. Basic Usage:: >>> from annict.api import API >>> api = API('your-access-token') >>> api.me() """ def __init__(self, token, base_url='https://api.annict.com', api_version='v1', parser=ModelParser): self.token = token self.base_url = base_url self.api_version = api_version self.parser = parser(self) @cursor_support def works(self, fields=None, filter_ids=None, filter_season=None, filter_title=None, page=None, per_page=None, sort_id=None, sort_season=None, sort_watchers_count=None): """Get works information :reference: https://docs.annict.com/ja/api/v1/works.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param filter_ids: (optional) Filter results by IDs. :type filter_ids: list of int :param str filter_season: (optional) Filter results by release time of season. :param str filter_title: (optional) Filter results by title. :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :param str sort_season: (optional) Sort the results by their release time of season. You can specify `asc` or `desc`. :param str sort_watchers_count: (optional) Sort the results by their watchers count. You can specify `asc` or `desc`. :return: list of :class:`Work ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='works', method='GET', allowed_params=('fields', 'filter_ids', 'filter_season', 'filter_title', 'page', 'per_page', 'sort_id', 'sort_season', 'sort_watchers_count'), payload_type='work', payload_is_list=True, ) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def episodes(self, fields=None, filter_ids=None, filter_work_id=None, page=None, per_page=None, sort_id=None, sort_sort_number=None): """Get episodes information :reference: https://docs.annict.com/ja/api/v1/episodes.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param filter_ids: (optional) Filter results by IDs. :type filter_ids: list of int :param int filter_work_id: (optional) Filter results by Work's ID. :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :param str sort_sort_number: (optional) Sort by number for sorting. You can specify `asc` or `desc`. :return: list of :class:`Episode ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='episodes', method='GET', allowed_params=('fields', 'filter_ids', 'filter_work_id', 'page', 'per_page', 'sort_id', 'sort_sort_number'), payload_type='episode', payload_is_list=True, ) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def records(self, fields=None, filter_ids=None, filter_episode_id=None, filter_has_record_comment=None, page=None, per_page=None, sort_id=None, sort_likes_count=None): """Get records to episodes :reference: https://docs.annict.com/ja/api/v1/records.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param filter_ids: (optional) Filter results by IDs. :type filter_ids: list of int :param int filter_episode_id: (optional) Filter results by Episode's ID. :param bool filter_has_record_comment: (optional) Filter the results by the presence or absence of comments. If you specify `True`, only records with comments will be filtered. Specifying `False` Filter records without comments. :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :param str sort_likes_count: (optional) Sort the results by their number of likes. You can specify `asc` or `desc`. :return: list of :class:`Record ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='records', method='GET', allowed_params=('fields', 'filter_ids', 'filter_episode_id', 'filter_has_record_comment', 'page', 'per_page', 'sort_id', 'sort_likes_count'), payload_type='record', payload_is_list=True, ) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def search_users(self, fields=None, filter_ids=None, filter_usernames=None, page=None, per_page=None, sort_id=None): """Get users information :reference: https://docs.annict.com/ja/api/v1/users.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param filter_ids: (optional) Filter results by IDs. :type filter_ids: list of int :param filter_usernames: (optional) Filter results by usernames. :type filter_usernames: list of str :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :return: list of :class:`User ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='users', method='GET', allowed_params=('fields', 'filter_ids', 'filter_usernames', 'page', 'per_page', 'sort_id'), payload_type='user', payload_is_list=True, ) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def following(self, fields=None, filter_user_id=None, filter_username=None, page=None, per_page=None, sort_id=None): """Get following information :reference: https://docs.annict.com/ja/api/v1/following.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param int filter_user_id: (optional) Filter results by User's ID. :param str filter_username: (optional) Filter results by username. :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :return: list of :class:`User ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='following', method='GET', allowed_params=('fields', 'filter_user_id', 'filter_username', 'page', 'per_page', 'sort_id'), payload_type='user', payload_is_list=True, ) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def followers(self, fields=None, filter_user_id=None, filter_username=None, page=None, per_page=None, sort_id=None): """Get followers information :reference: https://docs.annict.com/ja/api/v1/followers.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param int filter_user_id: (optional) Filter results by User's ID. :param str filter_username: (optional) Filter results by username. :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :return: list of :class:`User ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='followers', method='GET', allowed_params=('fields', 'filter_user_id', 'filter_username', 'page', 'per_page', 'sort_id'), payload_type='user', payload_is_list=True, ) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def activities(self, fields=None, filter_user_id=None, filter_username=None, page=None, per_page=None, sort_id=None): """Get activities :reference: https://docs.annict.com/ja/api/v1/activities.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param int filter_user_id: (optional) Filter results by User's ID. :param str filter_username: (optional) Filter results by username. :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :return: list of :class:`Activity ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='activities', method='GET', allowed_params=('fields', 'filter_user_id', 'filter_username', 'page', 'per_page', 'sort_id'), payload_type='activity', payload_is_list=True, ) params = api_method.build_parameters(locals()) return api_method(params) def me(self, fields=None): """Get your profile information :reference: https://docs.annict.com/ja/api/v1/me.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :return: :class:`User ` object of your user information. """ api_method = APIMethod( api=self, path='me', method='GET', allowed_params=('fields',), payload_type='user', ) params = api_method.build_parameters(locals()) return api_method(params) def set_status(self, work_id, kind): """Set the status of the work. :reference: https://docs.annict.com/ja/api/v1/me-statuses.html :param int work_id: Work's ID :param str kind: Types of status. You can specify `wanna_watch`, `watching`, `watched`, `on_hold`, `stop_watching`, or `no_select`. :return: Returns `True` if deletion succeeded. """ api_method = APIMethod( api=self, path='me/statuses', method='POST', allowed_params=('work_id', 'kind'), ) params = api_method.build_parameters(locals()) return api_method(params) def create_record(self, episode_id, comment=None, rating=None, share_twitter=False, share_facebook=False): """Create a record to the episode. :reference: https://docs.annict.com/ja/api/v1/me-records.html :param int episode_id: Episode's ID :param str comment: (optional) Comment. :param float rating: (optional) Rating. :param bool share_twitter: (optional) Whether to share the record on Twitter. You can enter `True` or `False`. :param bool share_facebook: (optional) Whether to share the record on Facebook. You can enter `True` or `False`. :return: :class:`Record ` object. """ api_method = APIMethod( api=self, path='me/records', method='POST', allowed_params=('episode_id', 'comment', 'rating', 'share_twitter', 'share_facebook'), payload_type='record', ) params = api_method.build_parameters(locals()) return api_method(params) def edit_record(self, id_, comment=None, rating=None, share_twitter=False, share_facebook=False): """Edit the created record. :reference: https://docs.annict.com/ja/api/v1/me-records.html :param int id_: Record's ID. :param str comment: (optional) Comment. :param float rating: (optional) Rating. :param bool share_twitter: (optional) Whether to share the record on Twitter. You can enter `True` or `False`. :param bool share_facebook: (optional) Whether to share the record on Facebook. You can enter `True` or `False`. :return: :class:`Record ` object after update. """ api_method = APIMethod( api=self, path='me/records', method='PATCH', allowed_params=('comment', 'rating', 'share_twitter', 'share_facebook'), payload_type='record', ) api_method.build_path(id_) params = api_method.build_parameters(locals()) return api_method(params) def delete_record(self, id_): """Delete the created record. :reference: https://docs.annict.com/ja/api/v1/me-records.html :param int id_: Recode's ID :return: Returns `True` if deletion succeeded. """ api_method = APIMethod( api=self, path='me/records', method='DELETE', allowed_params=(), ) api_method.build_path(id_) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def my_works(self, fields=None, filter_ids=None, filter_season=None, filter_title=None, filter_status=None, page=None, per_page=None, sort_id=None, sort_season=None, sort_watchers_count=None): """Get the information of the work you are setting status. :reference: https://docs.annict.com/ja/api/v1/me-works.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param filter_ids: (optional) Filter results by IDs. :type filter_ids: list of int :param str filter_season: (optional) Filter results by release time of season. :param str filter_title: (optional) Filter results by title. :param str filter_status: (optional) Filter results by status. You can specify `wanna_watch`, `watching`, `watched`, `on_hold`, `stop_watching`. :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :param str sort_season: (optional) Sort the results by their release time of season. You can specify `asc` or `desc`. :param str sort_watchers_count: (optional) Sort the results by their watchers count. You can specify `asc` or `desc`. :return: list of :class:`Work ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='me/works', method='GET', allowed_params=('fields', 'filter_ids', 'filter_season', 'filter_title', 'filter_status', 'page', 'per_page', 'sort_id', 'sort_season', 'sort_watchers_count'), payload_type='work', payload_is_list=True, ) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def my_programs(self, fields=None, filter_ids=None, filter_channel_ids=None, filter_work_ids=None, filter_started_at_gt=None, filter_started_at_lt=None, filter_unwatched=None, filter_rebroadcast=None, page=None, per_page=None, sort_id=None, sort_started_at=None): """Get the broadcast schedule. :reference: https://docs.annict.com/ja/api/v1/me-programs.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param filter_ids: (optional) Filter results by IDs. :type filter_ids: list of int :param filter_channel_ids: (optional) Filter results by Channel IDs. :type filter_channel_ids: list of int :param filter_work_ids: (optional) Filter results by Work IDs. :type filter_work_ids: list of int :param datetime filter_started_at_gt: (optional) Filter results results to those with the broadcast start date and time after the specified date and time. :param datetime filter_started_at_lt: (optional) Filter results results to those with the broadcast start date and time before the specified date and time. :param bool filter_unwatched: (optional) Only get unwatched broadcast schedules. :param bool filter_rebroadcast: (optional) Filter the broadcast schedule based on the rebroadcast flag. If you pass `True`, only rebroadcasting, passing `False` will get broadcast schedules other than rebroadcast. :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :param str sort_started_at: (optional) Sort the results by started_at. :return: list of :class:`Program ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='me/programs', method='GET', allowed_params=('fields', 'filter_ids', 'filter_channel_ids', 'filter_work_ids', 'filter_started_at_gt', 'filter_started_at_lt', 'filter_unwatched', 'filter_rebroadcast', 'page', 'per_page', 'sort_id', 'sort_started_at'), payload_type='program', payload_is_list=True ) params = api_method.build_parameters(locals()) return api_method(params) @cursor_support def following_activities(self, fields=None, filter_actions=None, filter_muted=None, page=None, per_page=None, sort_id=None): """Get the activity of the user you are following. :reference: https://docs.annict.com/ja/api/v1/me-following-activities.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param str filter_actions: (optional) Filter results by action (create_record|create_multiple_records|create_status). :param bool filter_muted: (optional) Specify whether to exclude muted users with the mute function. You can exclude with `True` and not exclude with `False`. The default is `True` (exclude). :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :return: list of :class:`Activity ` objects. :rtype: annict.models.ResultSet """ api_method = APIMethod( api=self, path='me/following_activities', method='GET', allowed_params=('fields', 'filter_actions', 'filter_muted', 'page', 'per_page', 'sort_id'), payload_type='activity', payload_is_list=True ) params = api_method.build_parameters(locals()) return api_method(params) PK!I211annict/auth.py# -*- coding: utf-8 -*- import json from urllib.parse import urljoin from rauth import OAuth2Service class OAuthHandler(object): """OAuth authentication handler""" def __init__(self, client_id, client_secret, name='annict', base_url='https://api.annict.com', redirect_uri='urn:ietf:wg:oauth:2.0:oob'): self.client_id = client_id self.client_secret = client_secret self.name = name self.base_url = base_url self.redirect_uri = redirect_uri self.auth_session = None self.oauth = OAuth2Service(client_id=client_id, client_secret=client_secret, name=name, base_url=base_url, authorize_url=urljoin(base_url, '/oauth/authorize'), access_token_url=urljoin(base_url, '/oauth/token')) def get_authorization_url(self, scope='read'): """Returns an authorization url :param scope: (optional) Specify authority to access resources. Readonly defaults. :return: URL of page requesting permission :rtype: str """ params = {'scope': scope, 'response_type': 'code', 'redirect_uri': self.redirect_uri} return self.oauth.get_authorize_url(**params) def authenticate(self, code, decoder=lambda s: json.loads(s.decode('utf8'))): """Acquire the access token using the authorization code acquired after approval. :param code: Authorization code obtained after approval. :param decoder: (optional) A function used to parse the Response content. Should return a dictionary. """ data = {'code': code, 'grant_type': 'authorization_code', 'redirect_uri': self.redirect_uri} session = self.oauth.get_auth_session(data=data, decoder=decoder) self.auth_session = session def get_access_token(self): """Returns the access token when authenticated.""" if self.auth_session: return self.auth_session.access_token PK!annict/cursors.py# -*- coding: utf-8 -*- from functools import wraps class SimpleCursor(object): """Simple cursor class""" def __init__(self, method, **kwargs): if not hasattr(method, 'cursor_support'): raise TypeError(f"Cursor does not support this method: {method.__func__.__qualname__}") self.method = method self.kwargs = kwargs if 'page' not in self.kwargs: self.kwargs['page'] = 1 def cursor(self): while 1: results = self.method(**self.kwargs) for result in results: yield result self.kwargs['page'] += 1 if not results.next_page or not results: return def cursor_support(api_method): """ Cursor support decorator :param api_method: API method that wan to correspond to the cursor. :return: wrapped method """ api_method.cursor_support = True @wraps(api_method) def wrapper(*args, **kwargs): return api_method(*args, **kwargs) return wrapper PK!ӯ822annict/models.py# -*- coding: utf-8 -*- import abc import arrow class ResultSet(list): """A list like object that holds results from an Annict API query.""" def __init__(self, total_count, prev_page=None, next_page=None): super().__init__() self.total_count = total_count self.prev_page = prev_page self.next_page = next_page class Model(metaclass=abc.ABCMeta): """Abstract class of each models.""" def __init__(self, api=None): self._api = api self._children = [] @classmethod @abc.abstractmethod def parse(cls, api, json): """Parse a JSON object into a model instance.""" raise NotImplementedError @classmethod def parse_list(cls, api, json, payload_type): """Parse JSON objects into list of model instances. :param api: instance of :class:`API ` . :type api: annict.api.API :param dict json: JSON from Annict API. :param str payload_type: Type of payload. :return: list of Model objects. :rtype: ResultSet """ results = ResultSet( total_count=json['total_count'], prev_page=json['prev_page'], next_page=json['next_page'], ) results._json = json if payload_type == 'activity': pluralized_payload_name = 'activities' else: pluralized_payload_name = '{}s'.format(payload_type) for obj in json[pluralized_payload_name]: if obj: results.append(cls.parse(api, obj)) return results class User(Model): """User information model""" def __repr__(self): return f'' @classmethod def parse(cls, api, json): """Parse a JSON object into a model instance. :param api: instance of :class:`API ` . :type api: annict.api.API :param dict json: JSON from Annict API. :return: :class:`User ` object :rtype: User """ user = cls(api) user._json = json for k, v in json.items(): if k == 'created_at': setattr(user, k, arrow.get(v).datetime) else: setattr(user, k, v) return user def following(self, fields=None, page=None, per_page=None, sort_id=None): """Get following information of this user. :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :return: list of :class:`User ` objects. :rtype: annict.models.ResultSet """ return self._api.following(fields=fields, filter_user_id=self.id, page=page, per_page=per_page, sort_id=sort_id) def followers(self, fields=None, page=None, per_page=None, sort_id=None): """Get following information of this user. :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :return: list of :class:`User ` objects. :rtype: annict.models.ResultSet """ return self._api.followers(fields=fields, filter_user_id=self.id, page=page, per_page=per_page, sort_id=sort_id) class Work(Model): """Work information model""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._episodes = None def __repr__(self): return f'' @classmethod def parse(cls, api, json): """Parse a JSON object into a model instance. :param api: instance of :class:`API ` . :type api: annict.api.API :param dict json: JSON from Annict API. :return: :class:`Work ` object :rtype: Work """ work = cls(api) work._json = json for k, v in json.items(): if k == 'released_on': if v: date = arrow.get(v).date() else: date = None setattr(work, k, date) else: setattr(work, k, v) return work def set_status(self, kind): """Set the status of the work. :param str kind: Types of status. You can specify `wanna_watch`, `watching`, `watched`, `on_hold`, `stop_watching`, or `no_select`. :return: Returns `True` if deletion succeeded. """ return self._api.set_status(self.id, kind) def episodes(self, fields=None, filter_ids=None, page=None, per_page=None, sort_id=None, sort_sort_number=None): """Get episodes information :reference: https://docs.annict.com/ja/api/v1/episodes.html :param fields: (optional) Narrow down the fields of data contained in the response body. :type fields: list of str :param filter_ids: (optional) Filter results by IDs. :type filter_ids: list of int :param int page: (optional) Specify the number of pages. :param int per_page: (optional) Specify how many items to acquire per page. :param str sort_id: (optional) Sort the results by their ID. You can specify `asc` or `desc`. :param str sort_sort_number: (optional) Sort by number for sorting. You can specify `asc` or `desc`. :return: list of :class:`Episode ` objects. :rtype: annict.models.ResultSet """ return self._api.episodes(fields=fields, filter_ids=filter_ids, filter_work_id=self.id, page=page, per_page=per_page, sort_id=sort_id, sort_sort_number=sort_sort_number) def select_episodes(self, *numbers): """Select multiple episodes :param numbers: Episode number. :return: list of :class:`Episode ` :rtype: list of :class:`Episode ` """ if not self._episodes: self._episodes = self.episodes(sort_sort_number='asc') if not numbers: return self._episodes return [self._episodes[n - 1] for n in numbers] def get_episode(self, number): """Get Episode object :param number: Episode number :return: :class:`Episode ` object :rtype: Episode """ return self.select_episodes(number)[0] class Episode(Model): """Episode information model""" def __repr__(self): return f'' @classmethod def parse(cls, api, json): """Parse a JSON object into a model instance. :param api: instance of :class:`API ` . :type api: annict.api.API :param dict json: JSON from Annict API. :return: :class:`Episode ` object :rtype: Episode """ episode = cls(api) episode._json = json for k, v in json.items(): if k == 'episode': setattr(episode, 'episode_id', v) elif k == 'work': work = Work.parse(api, v) setattr(episode, k, work) elif k == 'prev_episode' and v: prev_episode = cls.parse(api, v) setattr(episode, k, prev_episode) elif k == 'next_episode' and v: next_episode = cls.parse(api, v) setattr(episode, k, next_episode) else: setattr(episode, k, v) return episode def create_record(self, comment=None, rating=None, share_twitter=False, share_facebook=False): """Create a record for this episode. :param str comment: (optional) Comment. :param float rating: (optional) Rating. :param bool share_twitter: (optional) Whether to share the record on Twitter. You can enter `True` or `False`. :param bool share_facebook: (optional) Whether to share the record on Facebook. You can enter `True` or `False`. :return: :class:`Record ` object. """ return self._api.create_record(self.id, comment, rating, share_twitter, share_facebook) class Record(Model): """Record information model""" def __repr__(self): return f'' @classmethod def parse(cls, api, json): """Parse a JSON object into a model instance. :param api: instance of :class:`API ` . :type api: annict.api.API :param dict json: JSON from Annict API. :return: :class:`Record ` object :rtype: Record """ record = cls(api) record._json = json for k, v in json.items(): if k == 'created_at': setattr(record, k, arrow.get(v).datetime) elif k == 'user': user = User.parse(api, v) setattr(record, k, user) elif k == 'work': work = Work.parse(api, v) setattr(record, k, work) elif k == 'episode': episode = Episode.parse(api, v) setattr(record, k, episode) else: setattr(record, k, v) return record def edit(self, comment=None, rating=None, share_twitter=False, share_facebook=False): """Edit the created record. :param str comment: (optional) Comment. :param float rating: (optional) Rating. :param bool share_twitter: (optional) Whether to share the record on Twitter. You can enter `True` or `False`. :param bool share_facebook: (optional) Whether to share the record on Facebook. You can enter `True` or `False`. :return: :class:`Record ` object after edit. """ return self._api.edit_record(self.id, comment, rating, share_twitter, share_facebook) def delete(self): """Delete the created record. :return: Returns `True` if deletion succeeded. """ return self._api.delete_record(self.id) class Program(Model): """Program information model""" def __repr__(self): return f'' @classmethod def parse(cls, api, json): """Parse a JSON object into a model instance. :param api: instance of :class:`API ` . :type api: annict.api.API :param dict json: JSON from Annict API. :return: :class:`Program ` object :rtype: Program """ program = cls(api) program._json = json for k, v in json.items(): if k == 'started_at': setattr(program, k, arrow.get(v).datetime) elif k == 'work': work = Work.parse(api, v) setattr(program, k, work) elif k == 'episode': episode = Episode.parse(api, v) setattr(program, k, episode) else: setattr(program, k, v) return program class Activity(Model): """Activity information model""" def __repr__(self): return f'' @classmethod def parse(cls, api, json): """Parse a JSON object into a model instance. :param api: instance of :class:`API ` . :type api: annict.api.API :param dict json: JSON from Annict API. :return: :class:`Activity ` object :rtype: Activity """ activity = cls(api) activity._json = json for k, v in json.items(): if k == 'created_at': setattr(activity, k, arrow.get(v).datetime) elif k == 'user': user = User.parse(api, v) setattr(activity, k, user) elif k == 'work': work = Work.parse(api, v) setattr(activity, k, work) elif k == 'episode': episode = Episode.parse(api, v) setattr(activity, k, episode) else: setattr(activity, k, v) return activity MODEL_MAPPING = { 'user': User, 'work': Work, 'episode': Episode, 'record': Record, 'program': Program, 'activity': Activity, } PK!)Nannict/parsers.py# -*- coding: utf-8 -*- from .models import MODEL_MAPPING class ModelParser(object): def __init__(self, api, model_mapping=None): self.model_mapping = model_mapping if model_mapping else MODEL_MAPPING self._api = api def parse(self, json, payload_type, payload_is_list=False): model = self.model_mapping[payload_type] if payload_is_list: return model.parse_list(self._api, json, payload_type) else: return model.parse(self._api, json) PK!X؅<  annict/services.py# -*- coding: utf-8 -*- from furl import furl import requests from .utils import stringify class APIMethod(object): """A class abstracting each method of AnnictAPI :param api: instance of :class:`API ` . :type api: annict.api.API :param str path: Endpoint path :param str method: HTTP Method :param tuple allowed_params: (optional) List of request parameter names that can be sent. :param str payload_type: Type of payload :param bool payload_list: Specifies whether the payload is a list or not. """ def __init__(self, api, path, method, allowed_params=None, payload_type=None, payload_is_list=False): self.api = api self.path = path self.method = method self.allowed_params = allowed_params self.payload_type = payload_type self.payload_is_list = payload_is_list def build_path(self, id_=None): """Build an suitable path If `id_` is given, it is embedded into path. :param int id_: Target resource ID """ if id_ is not None: self.path = '/'.join([self.path, str(id_)]) def build_url(self): """Build request url :return: request url :rtype: str """ url = furl(self.api.base_url) url.path.add(self.api.api_version).add(self.path) return url.url def build_parameters(self, dic): """Build a suitable parameters for request. It filters the given dictionary based on `self.allowed_params` and returns a dictionary with an additional access token. :param dict dic: dict of arguments given to annict.API's method. :return: dict for request parameter :rtype: dict """ params = {key: stringify(dic[key]) for key in self.allowed_params if key in dic and dic[key]} params['access_token'] = self.api.token return params def __call__(self, params): url = self.build_url() resp = requests.request(self.method, url, params=params) resp.raise_for_status() if resp.status_code == 200: return self.api.parser.parse(resp.json(), self.payload_type, self.payload_is_list) elif resp.status_code == 204: return True else: return resp PK!?annict/utils.py# -*- coding: utf-8 -*- from functools import singledispatch @singledispatch def stringify(arg): return str(arg) @stringify.register(str) def do_not_stringify(arg): return arg @stringify.register(tuple) @stringify.register(list) @stringify.register(set) def stringify_list(arg): return ','.join([str(o) for o in arg]) @stringify.register(bool) def stringify_boolean(arg): return str(arg).lower() PK! $$annict-0.7.0.dist-info/LICENSEMIT License Copyright (c) 2016 kk6 Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HnHTUannict-0.7.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H-(annict-0.7.0.dist-info/METADATAW[oE~_1PA޵DРBCRPzٝef6Uc6E9;{;Z*ɩRlmp%jH4ƴ"jH#F1HC0,EE H&&v=#E,i4&hX+#2x Qkѭh Ş XG4PR@=S'zoˍrɍ؊jܸL@RĐ3;ˤ+(JfH18qn{ "I5N\BeoU]ȄbZf8f٢[B2㫹) ~u4æ_#1I=*zg]Vm+ˮ5Ҁf,"ᔼ@ά jkl`h$ p ag8׻F;Q}Zh  PsNJ2Yݞ6 D,OF %,t_IFYmz` Qdas,} x|& ./髣9#NuSc,an!_`ڙ\L6ŷ< ѡH^")G5b!.Jj4BesE4Vy˕"./1QΧDY&$CɊH`1H Qv  ^2|Ux-g!Hm^#f!YX-<% hE}d$ER-Nn7qrVp>~ryzқ@}'[K@A(AWRݚ6ܟ;cjG9{> q$Yi(u3R]C+ R[!00Nk39CMY QÞ^R)2a C SS_o[f8pnka/Z,)Nev#9e@ Or-Rib:Tjh} Z,ϫBPSqx)e\LfSb6R^In9Lsedraf?}'~WSTQ^zbv}}}[>)?.?(fGbE1ʵ)v鐋xq4Ans^]i_a7]MvݵM\<wfb ju)w#ס i$'wWYa-a'o|vӯ'4z˷VHVrl^DHŤA&x5v5=$M1L6}kW,u`c$.mOA2dNKah#yR_jpMQsl*]B.G(but^/3RҮmՋ+rp䎵jj,Z9{G~[㪿䠃: 7K $ !Z"8#41t }^ T)򦖣戵z6a#lis$k=;,sv] 0P&V9aG{V_7~G(ide&ٿchDnZDY7j/ӌK_LPK!H6 .iannict-0.7.0.dist-info/RECORDu˒kPy? i&$$.ar >@ s5Xgn<0' |3C9X R8S,;8\7|~DCm&U4N@za,T`4I/eDQ.^5%lپI6p53%+uG?iC~a:pp= %)zf 0wy^E8a6&F]*?O -饪9sӮ[,V#"KԆ:+r̵o﷿$S|\]`m)ԥiN|]ҿd"o/54H_]jPɗVHGffW`eM5%A3E PK! ȭannict/__init__.pyPK!OUSS annict/api.pyPK!I211Tannict/auth.pyPK!5]annict/cursors.pyPK!ӯ822uaannict/models.pyPK!)Neannict/parsers.pyPK!X؅<  annict/services.pyPK!?՟annict/utils.pyPK! $$annict-0.7.0.dist-info/LICENSEPK!HnHTUannict-0.7.0.dist-info/WHEELPK!H-(annict-0.7.0.dist-info/METADATAPK!H6 .iannict-0.7.0.dist-info/RECORDPK