3 Commits
1.2.3 ... 1.1.3

Author SHA1 Message Date
FoxMaSk
e892d3ceeb upd few things 2018-04-18 11:07:46 +02:00
FoxMaSk
e52b8befe4 upd few things 2018-04-18 11:07:23 +02:00
FoxMaSk
259f62efd9 last wallagab api - with requests 2017-09-24 22:35:20 +02:00
6 changed files with 162 additions and 289 deletions

7
.gitignore vendored
View File

@@ -1,8 +1 @@
.idea .idea
*.log
*.pot
*.pyc
.coverage
*.json
.tox
build

View File

@@ -1,24 +1,13 @@
.. image:: http://img.shields.io/badge/python-3.6-orange.svg
:target: https://pypi.python.org/pypi/django-th/
:alt: Python version supported
.. image:: http://img.shields.io/badge/license-BSD-blue.svg
:target: https://pypi.python.org/pypi/django-th/
:alt: License
============ ============
Wallabag API Wallabag API
============ ============
Python API for Wallabag v2.2.3 Python API for Wallabag v2
Requirements : Requirements :
============== ==============
* aiohttp * requests
Installation: Installation:
============= =============
@@ -27,7 +16,7 @@ to get the project, from your virtualenv, do :
.. code:: python .. code:: python
git clone https://github.com/push-things/wallabag_api/ git clone https://github.com/foxmask/wallabag-api/
or or
@@ -46,71 +35,30 @@ Creating a post :
.. code:: python .. code:: python
#!/usr/bin/env python
import aiohttp
import asyncio
from wallabag_api.wallabag import Wallabag from wallabag_api.wallabag import Wallabag
# settings # settings
params = {'username': 'foxmask',
'password': 'mypass',
'client_id': 'myid',
'client_secret': 'mysecret'}
my_host = 'http://localhost:8080' my_host = 'http://localhost:8080'
# get token
token = Wallabag.get_token(host=my_host, **params)
# create a post
wall = Wallabag(host=my_host, client_secret='mysecret', client_id='myid', token=token)
async def main(loop): my_url = 'https://blog.trigger-happy.eu'
my_title = 'Trigger Happy blog'
my_tags = ['python', 'wallabag']
params = {'username': 'foxmask', wall.post_entries(url=my_url, title=my_title, tags=my_tags)
'password': 'mypass',
'client_id': 'myid',
'client_secret': 'mysecret',
'extension': 'pdf'}
# get a new token
token = await Wallabag.get_token(host=my_host, **params)
# initializing
async with aiohttp.ClientSession(loop=loop) as session:
wall = Wallabag(host=my_host,
client_secret=params.get('client_secret'),
client_id=params.get('client_id'),
token=token,
extension=params['extension'],
aio_sess=session)
url = 'https://foxmask.trigger-happy.eu'
title = 'foxmask\'s blog'
await wall.post_entries(url, title, '', 0, 0)
url = 'https://trigger-happy.eu'
title = 'Project TrigerHappy'
await wall.post_entries(url, title, '', 0, 0)
# get all the articles
my_wallabag = await wall.get_entries()
all_article = my_wallabag['_embedded']['items']
for article in all_article:
print(article['id'], article['title'])
# get the version of wallabag
version = await wall.version
print(f"version {version}")
# export one article into PDF
my_wallabag = await wall.get_entry_export(entry=1)
with open("foobar.pdf", "wb") as f:
f.write(my_wallabag)
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
this will give you something like this : this will give you something like this :
.. image:: https://github.com/push-things/wallabag_api/blob/master/wallabag.png .. image:: https://github.com/foxmask/wallabag_api/blob/master/wallabag.png
Testing : Testing :
@@ -118,13 +66,11 @@ Testing :
Install Wallabag V2 on your own host like explain here http://doc.wallabag.org/en/v2/user/installation.html Install Wallabag V2 on your own host like explain here http://doc.wallabag.org/en/v2/user/installation.html
Then run the development version (with make run)
Then create a client API like explain here http://doc.wallabag.org/en/v2/developer/api.html Then create a client API like explain here http://doc.wallabag.org/en/v2/developer/api.html
this will give you something like this this will give you somthing like this
.. image:: https://github.com/push-things/wallabag_api/blob/master/wallabag_api_key.png .. image:: https://github.com/foxmask/wallabag_api/blob/master/wallabag_api_key.png
Then replace the client_id / client_secret / login / pass to wallabag_test.py and run Then replace the client_id / client_secret / login / pass to wallabag_test.py and run
@@ -132,3 +78,4 @@ Then replace the client_id / client_secret / login / pass to wallabag_test.py an
python wallabag_test.py python wallabag_test.py

View File

@@ -1,29 +1,29 @@
from setuptools import setup, find_packages from setuptools import setup, find_packages
from wallabag_api import __version__ as version from wallabag_api import __version__ as version
desc = 'Wallabag API to add every pages you want to your Wallabag account'
long_desc = 'Wallabag is a "read it later" service, and that Wallabag API allow you to save web pages ' \
'to your own account'
install_requires = [ install_requires = [
'aiohttp', 'requests',
] ]
setup( setup(
name='wallabag_api', name='wallabag_api',
version=version, version=version,
description=desc, description='Wallabag API to add every pages you want to your Wallabag account',
long_description=long_desc, long_desc='Wallabag is a "read it later" service, and that Wallabag API allow you to save web pages '
'to your own account',
author='FoxMaSk', author='FoxMaSk',
author_email='foxmaskhome@gmail.com', author_email='foxmask@trigger-happy.eu',
url='https://github.com/push-things/wallabag_api', url='https://github.com/push-things/wallabag_api',
download_url="https://github.com/push-things/wallabag_api/archive/wallabag_api-" + version + ".zip", download_url="https://github.com/push-things/wallabag_api/archive/wallabag_api-" + version + ".zip",
packages=find_packages(), packages=find_packages(),
classifiers=[ classifiers=[
'Development Status :: 5 - Production/Stable', 'Development Status :: 4 - Beta',
'Intended Audience :: Developers', 'Intended Audience :: Developers',
'License :: OSI Approved :: BSD License', 'License :: OSI Approved :: BSD License',
'Operating System :: OS Independent', 'Operating System :: OS Independent',
'Programming Language :: Python', 'Programming Language :: Python',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6', 'Programming Language :: Python :: 3.6',
'Topic :: Internet', 'Topic :: Internet',
'Topic :: Communications', 'Topic :: Communications',

View File

@@ -1,2 +1,2 @@
VERSION = (1, 2, 3) # PEP 386 VERSION = (1, 1, 3) # PEP 386
__version__ = ".".join([str(x) for x in VERSION]) __version__ = ".".join([str(x) for x in VERSION])

View File

@@ -1,11 +1,11 @@
# coding: utf-8 # coding: utf-8
import requests
from requests import HTTPError
import logging import logging
import aiohttp
from aiohttp.http_exceptions import HttpProcessingError
from aiohttp.client_exceptions import ClientResponseError
__author__ = 'foxmask' __author__ = 'foxmask'
logging.basicConfig(format='%(message)s', level=logging.INFO) logging.basicConfig(format='%(message)s', level=logging.INFO)
__all__ = ['Wallabag'] __all__ = ['Wallabag']
@@ -25,7 +25,6 @@ class Wallabag(object):
format = '' format = ''
username = '' username = ''
password = '' password = ''
aio_sess = None
def __init__(self, def __init__(self,
host='', host='',
@@ -33,9 +32,8 @@ class Wallabag(object):
client_id='', client_id='',
client_secret='', client_secret='',
extension='json', extension='json',
user_agent="WallabagPython/1.2.2 " user_agent="WallabagPython/1.0 "
" +https://github.com/push-things/wallabag-api", "+https://github.com/foxmask/wallabag-api"):
aio_sess=None):
""" """
init variable init variable
:param host: string url to the official API Wallabag :param host: string url to the official API Wallabag
@@ -44,7 +42,6 @@ class Wallabag(object):
:param client_secret client secret :param client_secret client secret
:param extension: xml|json|txt|csv|pdf|epub|mobi|html :param extension: xml|json|txt|csv|pdf|epub|mobi|html
:param user_agent :param user_agent
:param aio_sess aiohttp session
""" """
self.host = host self.host = host
self.client_id = client_id self.client_id = client_id
@@ -52,111 +49,94 @@ class Wallabag(object):
self.token = token self.token = token
self.format = extension self.format = extension
self.user_agent = user_agent self.user_agent = user_agent
self.aio_sess = aio_sess
if self.format not in self.EXTENTIONS: if self.format not in self.EXTENTIONS:
raise ValueError("format invalid {0} should be one of {1}".format( raise ValueError("format invalid {0} should be one of {1}".format(
self.format, self.EXTENTIONS)) self.format, self.EXTENTIONS))
async def query(self, path, method='get', **params): def get_host(self):
""" """
Do a query to the System API get the host from which to get API
:return host
"""
return self.host
:param path: url to the API def query(self, path, method='get', **params):
:param method: the kind of query to do
:param params: a dict with all the
necessary things to query the API
:return json data
""" """
if method in ('get', 'post', 'patch', 'delete', 'put'): Do a query to the System API
full_path = self.host + path :param path: url to the API
:param method: the kind of query to do
:param params: a dict with all the
necessary things to query the API
:return json data
"""
if method in ('get', 'post', 'patch', 'delete', 'put', 'get_token'):
if method == 'get': if method == 'get':
resp = await self.aio_sess.get(full_path, params=params) r = requests.get(self.get_host() + path, params=params)
elif method == 'post': elif method == 'post':
resp = await self.aio_sess.post(full_path, data=params) r = requests.post(self.get_host() + path, data=params)
elif method == 'patch': elif method == 'patch':
resp = await self.aio_sess.patch(full_path, data=params) r = requests.patch(self.get_host() + path, data=params)
elif method == 'delete': elif method == 'delete':
resp = await self.aio_sess.delete(full_path, headers=params) r = requests.delete(self.get_host() + path, headers=params)
elif method == 'put': elif method == 'put':
resp = await self.aio_sess.put(full_path, data=params) r = requests.put(self.get_host() + path, params=params)
return self.handle_json_response(r)
async with resp:
# return the content if its a binary one
if resp.content_type.startswith('application/pdf') or \
resp.content_type.startswith('application/epub'):
return await resp.read()
return await self.handle_json_response(resp)
else: else:
raise ValueError('method expected: get, post, patch, delete, put') raise ValueError('method expected : get, post, patch, delete or put')
@staticmethod @staticmethod
async def handle_json_response(responses): def handle_json_response(responses):
""" """
get the json data response get the json data response
:param responses: the json response :param responses: the json response
:return the json data without 'root' node :return the json data without 'root' node
""" """
json_data = {} json_data = {}
if responses.status != 200: if responses.status_code != 200:
err_msg = HttpProcessingError(code=responses.status, err_msg = HTTPError(responses.status_code, responses.reason)
message=await responses.json()) msg = "Wallabag: error {err_msg}".format(err_msg=err_msg)
logging.error("Wallabag: aiohttp error {err_msg}".format( logging.error(msg)
err_msg=err_msg)) raise HTTPError(msg)
else: else:
try: try:
json_data = responses.json() json_data = responses.json()
except ClientResponseError as e: except:
# sometimes json_data does not return any json() without for error in json_data['errors']:
# any error. This is due to the grabbing URL which "rejects" error_json = json_data['errors'][error]['content']
# the URL logging.error("Wallabag: {error}".format(error=error_json))
logging.error("Wallabag: aiohttp error {code} {message}" return json_data
.format(code=e.code, message=e.message))
return await json_data
@staticmethod
def __get_attr(what, type_attr, value_attr, **kwargs):
"""
get the value of a parm
:param what: string parm
:param type_attr: type of parm
:param value_attr:
:param kwargs:
:return: value of the parm
"""
value = int(kwargs[what]) if type_attr == 'int' else kwargs[what]
if what in kwargs and value in value_attr:
return value
# ENTRIES # ENTRIES
async def get_entries(self, **kwargs): def get_entries(self, **kwargs):
""" """
GET /api/entries.{_format}
Retrieve all entries. It could be filtered by many options. GET /api/entries.{_format}
:param kwargs: can contain one of the following filters Retrieve all entries. It could be filtered by many options.
archive: '0' or '1', default '0' filter by archived status.
star: '0' or '1', default '0' filter by starred status. :param kwargs: can contain one of the following filters
delete: '0' or '1', default '0' filter by deleted status. archive: '0' or '1', default '0' filter by archived status.
sort: 'created' or 'updated', default 'created' star: '0' or '1', default '0' filter by starred status.
order: 'asc' or 'desc', default 'desc' delete: '0' or '1', default '0' filter by deleted status.
page: int default 1 what page you want sort: 'created' or 'updated', default 'created'
perPage: int default 30 result per page order: 'asc' or 'desc', default 'desc'
tags: list of tags url encoded. page: int default 1 what page you want
since: int default 0 from what timestamp you want perPage: int default 30 result per page
Will returns entries that matches ALL tags tags: list of tags url encoded.
:return data related to the ext Will returns entries that matches ALL tags
:return data related to the ext
""" """
# default values # default values
params = dict({'access_token': self.token, params = {'access_token': self.token,
'delete': 0, 'archive': 0,
'sort': 'created', 'star': 0,
'order': 'desc', 'delete': 0,
'page': 1, 'sort': 'created',
'perPage': 30, 'order': 'desc',
'tags': '', 'page': 1,
'since': 0}) 'perPage': 30,
'tags': []}
if 'archive' in kwargs and int(kwargs['archive']) in (0, 1): if 'archive' in kwargs and int(kwargs['archive']) in (0, 1):
params['archive'] = int(kwargs['archive']) params['archive'] = int(kwargs['archive'])
@@ -171,16 +151,13 @@ class Wallabag(object):
if 'perPage' in kwargs and isinstance(kwargs['perPage'], int): if 'perPage' in kwargs and isinstance(kwargs['perPage'], int):
params['perPage'] = kwargs['perPage'] params['perPage'] = kwargs['perPage']
if 'tags' in kwargs and isinstance(kwargs['tags'], list): if 'tags' in kwargs and isinstance(kwargs['tags'], list):
params['tags'] = ', '.join(kwargs['tags']) params['tags'] = kwargs['tags']
if 'since' in kwargs and isinstance(kwargs['since'], int):
params['since'] = kwargs['since']
path = '/api/entries.{ext}'.format(ext=self.format) path = '/api/entries.{ext}'.format(ext=self.format)
return await self.query(path, "get", **params) return self.query(path, "get", **params)
async def post_entries(self, url, title='', tags='', starred=0, archive=0, content='', language='', published_at='', def post_entries(self, url, title='', tags='', starred=0, archive=0):
authors='', public=1, original_url=''):
""" """
POST /api/entries.{_format} POST /api/entries.{_format}
@@ -191,24 +168,16 @@ class Wallabag(object):
:param tags: tag1,tag2,tag3 a comma-separated list of tags. :param tags: tag1,tag2,tag3 a comma-separated list of tags.
:param starred entry already starred :param starred entry already starred
:param archive entry already archived :param archive entry already archived
:param content additionnal html content
:param language
:param published_at
:param authors
:param public
:param original_url
:return result :return result
""" """
params = {'access_token': self.token, 'url': url, 'title': title, params = {'access_token': self.token, 'url': url, 'title': title,
'tags': tags, 'starred': starred, 'archive': archive, 'tags': tags, 'starred': starred, 'archive': archive}
'content': content, 'language': language, 'published_at': published_at,
'authors': authors, 'public': public, 'original_url': original_url}
if len(tags) > 0 and isinstance(tags, list): if len(tags) > 0 and isinstance(tags, list):
params['tags'] = ', '.join(tags) params['tags'] = ', '.join(tags)
path = '/api/entries.{ext}'.format(ext=self.format) path = '/api/entries.{ext}'.format(ext=self.format)
return await self.query(path, "post", **params) return self.query(path, "post", **params)
async def get_entry(self, entry): def get_entry(self, entry):
""" """
GET /api/entries/{entry}.{_format} GET /api/entries/{entry}.{_format}
@@ -218,11 +187,10 @@ class Wallabag(object):
:return data related to the ext :return data related to the ext
""" """
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/entries/{entry}.{ext}'.format(entry=entry, url = '/api/entries/{entry}.{ext}'.format(entry=entry, ext=self.format)
ext=self.format) return self.query(url, "get", **params)
return await self.query(url, "get", **params)
async def patch_entries(self, entry, **kwargs): def patch_entries(self, entry, **kwargs):
""" """
PATCH /api/entries/{entry}.{_format} PATCH /api/entries/{entry}.{_format}
@@ -250,29 +218,17 @@ class Wallabag(object):
params['title'] = kwargs['title'] params['title'] = kwargs['title']
if 'tags' in kwargs and isinstance(kwargs['tags'], list): if 'tags' in kwargs and isinstance(kwargs['tags'], list):
params['tags'] = ', '.join(kwargs['tags']) params['tags'] = ', '.join(kwargs['tags'])
if 'archive' in kwargs and int(kwargs['archive']) in (0, 1):
params['archive'] = self.__get_attr(what='archive', params['archive'] = int(kwargs['archive'])
type_attr=int, if 'star' in kwargs and int(kwargs['star']) in (0, 1):
value_attr=(0, 1), params['star'] = int(kwargs['star'])
**kwargs) if 'delete' in kwargs and int(kwargs['delete']) in (0, 1):
params['star'] = self.__get_attr(what='star', params['delete'] = int(kwargs['delete'])
type_attr=int,
value_attr=(0, 1),
**kwargs)
params['delete'] = self.__get_attr(what='delete',
type_attr=int,
value_attr=(0, 1),
**kwargs)
params['order'] = self.__get_attr(what='order',
type_attr=str,
value_attr=('asc', 'desc'),
**kwargs)
path = '/api/entries/{entry}.{ext}'.format( path = '/api/entries/{entry}.{ext}'.format(
entry=entry, ext=self.format) entry=entry, ext=self.format)
return await self.query(path, "patch", **params) return self.query(path, "patch", **params)
async def get_entry_export(self, entry): def get_entry_export(self, entry):
""" """
GET /api/entries/{entry}/export.{_format} GET /api/entries/{entry}/export.{_format}
@@ -284,9 +240,9 @@ class Wallabag(object):
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/entries/{entry}/export.{ext}'.format(entry=entry, url = '/api/entries/{entry}/export.{ext}'.format(entry=entry,
ext=self.format) ext=self.format)
return await self.query(url, "get", **params) return self.query(url, "get", **params)
async def patch_entry_reload(self, entry): def patch_entry_reload(self, entry):
""" """
PATCH /api/entries/{entry}/reload.{_format} PATCH /api/entries/{entry}/reload.{_format}
@@ -300,9 +256,9 @@ class Wallabag(object):
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/entries/{entry}/reload.{ext}'.format(entry=entry, url = '/api/entries/{entry}/reload.{ext}'.format(entry=entry,
ext=self.format) ext=self.format)
return await self.query(url, "patch", **params) return self.query(url, "patch", **params)
async def delete_entries(self, entry): def delete_entries(self, entry):
""" """
DELETE /api/entries/{entry}.{_format} DELETE /api/entries/{entry}.{_format}
@@ -315,9 +271,9 @@ class Wallabag(object):
params = {'Authorization': 'Bearer {}'.format(self.token)} params = {'Authorization': 'Bearer {}'.format(self.token)}
path = '/api/entries/{entry}.{ext}'.format( path = '/api/entries/{entry}.{ext}'.format(
entry=entry, ext=self.format) entry=entry, ext=self.format)
return await self.query(path, "delete", **params) return self.query(path, "delete", **params)
async def entries_exists(self, url, urls=''): def entries_exists(self, url, urls=''):
""" """
GET /api/entries/exists.{_format} GET /api/entries/exists.{_format}
@@ -330,16 +286,14 @@ class Wallabag(object):
:return result :return result
""" """
params = {'access_token': self.token, params = {'url': url,
'url': url,
'urls': urls} 'urls': urls}
path = '/api/entries/exists.{ext}'.format(ext=self.format) path = '/api/entries/exists.{ext}'.format(ext=self.format)
return await self.query(path, "get", **params) return self.query(path, "get", **params)
# TAGS # TAGS
def get_entry_tags(self, entry):
async def get_entry_tags(self, entry):
""" """
GET /api/entries/{entry}/tags.{_format} GET /api/entries/{entry}/tags.{_format}
@@ -351,9 +305,9 @@ class Wallabag(object):
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/entries/{entry}/tags.{ext}'.format( url = '/api/entries/{entry}/tags.{ext}'.format(
entry=entry, ext=self.format) entry=entry, ext=self.format)
return await self.query(url, "get", **params) return self.query(url, "get", **params)
async def post_entry_tags(self, entry, tags): def post_entry_tags(self, entry, tags):
""" """
POST /api/entries/{entry}/tags.{_format} POST /api/entries/{entry}/tags.{_format}
@@ -364,13 +318,13 @@ class Wallabag(object):
:return result :return result
""" """
params = {'access_token': self.token, 'tags': []} params = {'access_token': self.token, 'tags': []}
if len(tags) > 0 and ',' in tags: if isinstance(tags, list):
params['tags'] = tags.split(',') params['tags'] = tags
path = '/api/entries/{entry}/tags.{ext}'.format( path = '/api/entries/{entry}/tags.{ext}'.format(
entry=entry, ext=self.format) entry=entry, ext=self.format)
return await self.query(path, "post", **params) return self.query(path, "post", **params)
async def delete_entry_tag(self, entry, tag): def delete_entry_tag(self, entry, tag):
""" """
DELETE /api/entries/{entry}/tags/{tag}.{_format} DELETE /api/entries/{entry}/tags/{tag}.{_format}
@@ -383,9 +337,9 @@ class Wallabag(object):
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/entries/{entry}/tags/{tag}.{ext}'.format( url = '/api/entries/{entry}/tags/{tag}.{ext}'.format(
entry=entry, tag=tag, ext=self.format) entry=entry, tag=tag, ext=self.format)
return await self.query(url, "delete", **params) return self.query(url, "delete", **params)
async def get_tags(self): def get_tags(self):
""" """
GET /api/tags.{_format} GET /api/tags.{_format}
@@ -395,9 +349,9 @@ class Wallabag(object):
""" """
params = {'access_token': self.token} params = {'access_token': self.token}
path = '/api/tags.{ext}'.format(ext=self.format) path = '/api/tags.{ext}'.format(ext=self.format)
return await self.query(path, "get", **params) return self.query(path, "get", **params)
async def delete_tag(self, tag): def delete_tag(self, tag):
""" """
DELETE /api/tags/{tag}.{_format} DELETE /api/tags/{tag}.{_format}
@@ -408,9 +362,9 @@ class Wallabag(object):
""" """
path = '/api/tags/{tag}.{ext}'.format(tag=tag, ext=self.format) path = '/api/tags/{tag}.{ext}'.format(tag=tag, ext=self.format)
params = {'access_token': self.token} params = {'access_token': self.token}
return await self.query(path, "delete", **params) return self.query(path, "delete", **params)
async def delete_tag_label(self, tag): def delete_tag_label(self, tag):
""" """
DELETE /api/tag/label.{_format} DELETE /api/tag/label.{_format}
@@ -422,9 +376,9 @@ class Wallabag(object):
path = '/api/tag/label.{ext}'.format(ext=self.format) path = '/api/tag/label.{ext}'.format(ext=self.format)
params = {'access_token': self.token, params = {'access_token': self.token,
'tag': tag} 'tag': tag}
return await self.query(path, "delete", **params) return self.query(path, "delete", **params)
async def delete_tags_label(self, tags): def delete_tags_label(self, tags):
""" """
DELETE /api/tags/label.{_format} DELETE /api/tags/label.{_format}
@@ -436,10 +390,10 @@ class Wallabag(object):
path = '/api/tag/label.{ext}'.format(ext=self.format) path = '/api/tag/label.{ext}'.format(ext=self.format)
params = {'access_token': self.token, params = {'access_token': self.token,
'tags': tags} 'tags': tags}
return await self.query(path, "delete", **params) return self.query(path, "delete", **params)
# ANNOTATIONS # ANNOTATIONS
async def delete_annotations(self, annotation): def delete_annotations(self, annotation):
""" """
DELETE /api/annotations/{annotation}.{_format} DELETE /api/annotations/{annotation}.{_format}
@@ -453,9 +407,9 @@ class Wallabag(object):
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/annotations/{annotation}.{ext}'.format( url = '/api/annotations/{annotation}.{ext}'.format(
annotation=annotation, ext=self.format) annotation=annotation, ext=self.format)
return await self.query(url, "delete", **params) return self.query(url, "delete", **params)
async def put_annotations(self, annotation): def put_annotations(self, annotation):
""" """
PUT /api/annotations/{annotation}.{_format} PUT /api/annotations/{annotation}.{_format}
@@ -469,9 +423,9 @@ class Wallabag(object):
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/annotations/{annotation}.{ext}'.format( url = '/api/annotations/{annotation}.{ext}'.format(
annotation=annotation, ext=self.format) annotation=annotation, ext=self.format)
return await self.query(url, "put", **params) return self.query(url, "put", **params)
async def get_annotations(self, entry): def get_annotations(self, entry):
""" """
GET /api/annotations/{entry}.{_format} GET /api/annotations/{entry}.{_format}
@@ -485,9 +439,9 @@ class Wallabag(object):
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/annotations/{entry}.{ext}'.format(entry=entry, url = '/api/annotations/{entry}.{ext}'.format(entry=entry,
ext=self.format) ext=self.format)
return await self.query(url, "get", **params) return self.query(url, "get", **params)
async def post_annotations(self, entry, **kwargs): def post_annotations(self, entry, **kwargs):
""" """
POST /api/annotations/{entry}.{_format} POST /api/annotations/{entry}.{_format}
@@ -510,11 +464,11 @@ class Wallabag(object):
url = '/api/annotations/{entry}.{ext}'.format(entry=entry, url = '/api/annotations/{entry}.{ext}'.format(entry=entry,
ext=self.format) ext=self.format)
return await self.query(url, "post", **params) return self.query(url, "post", **params)
# VERSION # VERSION
@property @property
async def version(self): def version(self):
""" """
GET /api/version.{_format} GET /api/version.{_format}
@@ -524,10 +478,10 @@ class Wallabag(object):
""" """
params = {'access_token': self.token} params = {'access_token': self.token}
url = '/api/version.{ext}'.format(ext=self.format) url = '/api/version.{ext}'.format(ext=self.format)
return await self.query(url, "get", **params) return self.query(url, "get", **params)
@classmethod @classmethod
async def get_token(cls, host, **params): def get_token(cls, host, **params):
""" """
POST /oauth/v2/token POST /oauth/v2/token
@@ -546,7 +500,6 @@ class Wallabag(object):
""" """
params['grant_type'] = "password" params['grant_type'] = "password"
path = "/oauth/v2/token" path = "/oauth/v2/token"
async with aiohttp.ClientSession() as sess: resp = requests.post(host + path, data=params)
async with sess.post(host + path, data=params) as resp: data = cls.handle_json_response(resp)
data = await cls.handle_json_response(resp) return data.get('access_token')
return data.get("access_token")

View File

@@ -1,12 +1,11 @@
# coding: utf-8 # coding: utf-8
import datetime
import unittest import unittest
from wallabag import Wallabag from wallabag import Wallabag
class TestWallabag(unittest.TestCase): class TestWallabag(unittest.TestCase):
host = 'http://localhost:8000' host = 'http://bag.foxmask'
client_id = '' client_id = ''
client_secret = '' client_secret = ''
token = '' token = ''
@@ -21,36 +20,28 @@ class TestWallabag(unittest.TestCase):
def test_get_token(self): def test_get_token(self):
params = {"grant_type": "password", params = {"grant_type": "password",
"client_id": "client_id": '1_3to3042y05gk8g4wcsk40w40k8kk00s04gwcoo4sows8wskcg0',
'1_4wqe1riwt0qoks844kwc4go08koogkgk88go4cckkwg0408kcg', "client_secret": 'ift9k07vd20ccg4ocosg0cw4kkgk4o8080848scwg0oosowow',
"client_secret": '4mzw3qwi1xyc0cks4k80s4c8kco40wwkkkw0g40kwk4o4c44co', "username": 'foxmask',
"username": 'wallabag', "password": 'ratatab00m'}
"password": 'wallabag'}
print(self.host)
data = Wallabag.get_token(host=self.host, **params) data = Wallabag.get_token(host=self.host, **params)
print(data)
self.assertTrue(isinstance(data, str), True) self.assertTrue(isinstance(data, str), True)
return data return data
def create_entry(self): def create_entry(self):
title = 'foobar title' title = 'foobar title'
url = 'https://somwhere.over.the.raibow.com/' url = 'https://trigger-happy.eu/'
tags = ['foo', 'bar'] tags = ['foo', 'bar']
starred = 0 starred = 0
archive = 0 archive = 0
content = '<p>Additional content</p>' data = self.w.post_entries(url, title, tags, starred, archive)
language = 'FR'
published_at = datetime.datetime.now()
authors = 'John Doe'
public = 0
original_url = 'http://localhost'
data = self.w.post_entries(url, title, tags, starred, archive, content, language, published_at, authors,
public, original_url)
return data return data
def test_get_entries(self): def test_get_entries(self):
params = {'delete': 0, params = {'archive': 0,
'star': 0,
'delete': 0,
'sort': 'created', 'sort': 'created',
'order': 'desc', 'order': 'desc',
'page': 1, 'page': 1,
@@ -84,7 +75,6 @@ class TestWallabag(unittest.TestCase):
params = {'title': 'I change the title', params = {'title': 'I change the title',
'archive': 0, 'archive': 0,
'tags': ["bimbo", "pipo"], 'tags': ["bimbo", "pipo"],
'order': 'asc',
'star': 0, 'star': 0,
'delete': 0} 'delete': 0}
self.assertTrue(isinstance(entry, int), True) self.assertTrue(isinstance(entry, int), True)
@@ -106,16 +96,6 @@ class TestWallabag(unittest.TestCase):
data = self.w.post_entry_tags(entry, tags) data = self.w.post_entry_tags(entry, tags)
self.assertTrue(data, True) self.assertTrue(data, True)
"""
def test_delete_entry_tag(self):
entry = self.create_entry()
tag = 'bar'
self.assertTrue(isinstance(entry['id'], int), True)
self.assertTrue(isinstance(tag, str), True)
resp = self.w.delete_entry_tag(entry['id'], tag)
self.assertTrue(resp, True)
"""
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()