219 lines
5.2 KiB
Python
219 lines
5.2 KiB
Python
from types import SimpleNamespace
|
|
import urllib
|
|
import requests
|
|
|
|
DEFAULT_URL = "http://127.0.0.1:8000"
|
|
|
|
API_TODO_ITEMS_LIST = "api/todo_items/"
|
|
API_TODO_ITEMS_CREATE = "api/todo_items/"
|
|
API_TODO_ITEMS_READ = "api/todo_items/{0}/"
|
|
API_TODO_ITEMS_UPDATE = "api/todo_items/{0}/"
|
|
API_TODO_ITEMS_PARTIAL_UPDATE = "api/todo_items/{0}/"
|
|
API_TODO_ITEMS_DELETE = "api/todo_items/{0}/"
|
|
|
|
API_LISTS_LIST = "api/lists/"
|
|
API_LISTS_CREATE = "api/lists/"
|
|
API_LISTS_READ = "lists/{0}/"
|
|
API_LISTS_UPDATE = "lists/{0}/"
|
|
API_LISTS_PARTIAL_UPDATE = "lists/{0}/"
|
|
API_LISTS_DELETE = "lists/{0}/"
|
|
|
|
API_TOKEN = "api/token/"
|
|
|
|
|
|
class UserApi(object):
|
|
def __init__(self, url=DEFAULT_URL, token=None):
|
|
"""
|
|
Constructor
|
|
|
|
Parameters
|
|
----------
|
|
url : str, optional
|
|
Server url. The default is DEFAULT_URL.
|
|
token : dict, optional
|
|
Existing user tokens to bypass authorization.
|
|
The default is None.
|
|
|
|
Returns
|
|
-------
|
|
None.
|
|
|
|
"""
|
|
self.token = token
|
|
self.get_api = lambda x: urllib.parse.urljoin(url, x)
|
|
|
|
# ToDo - store tokens in config
|
|
def auth(self, user, passwd):
|
|
"""
|
|
Authosization
|
|
|
|
Parameters
|
|
----------
|
|
user : str
|
|
Login.
|
|
passwd : str
|
|
Password.
|
|
|
|
Returns
|
|
-------
|
|
dict
|
|
Generated auth token.
|
|
|
|
"""
|
|
token = UserApi._raise_or_return_(
|
|
requests.post(url=self.get_api(API_TOKEN), json={"username": user, "password": passwd})
|
|
)
|
|
self.token = SimpleNamespace(**token)
|
|
return self.token
|
|
|
|
def lists_list(self, **argv):
|
|
"""
|
|
List all the exsiting to-do lists.
|
|
Auth required
|
|
|
|
Returns
|
|
-------
|
|
list
|
|
to-do lists.
|
|
|
|
"""
|
|
return UserApi._raise_or_return_(
|
|
requests.get(
|
|
url=self.get_api(API_LISTS_LIST), headers=self._access_token_(), params=argv
|
|
)
|
|
)
|
|
|
|
def lists_create(self, title="Untitled"):
|
|
"""
|
|
Create a new to-do list
|
|
Auth required
|
|
|
|
Parameters
|
|
----------
|
|
title : str, optional
|
|
New list name. The default is "Untitled".
|
|
|
|
"""
|
|
return UserApi._raise_or_return_(
|
|
requests.post(
|
|
url=self.get_api(API_LISTS_CREATE),
|
|
json={"title": title},
|
|
headers=self._access_token_(),
|
|
)
|
|
)
|
|
|
|
def todo_items_list(self, **argv):
|
|
"""
|
|
List all the exsiting to-do items.
|
|
Auth required
|
|
|
|
Returns
|
|
-------
|
|
list
|
|
to-do items.
|
|
|
|
"""
|
|
return UserApi._raise_or_return_(
|
|
requests.get(
|
|
url=self.get_api(API_TODO_ITEMS_LIST), headers=self._access_token_(), params=argv
|
|
)
|
|
)
|
|
|
|
# def create(self, title="Untitled"):
|
|
# """
|
|
# Create a new to-do list
|
|
# Auth required
|
|
|
|
# Parameters
|
|
# ----------
|
|
# title : str, optional
|
|
# New list name. The default is "Untitled".
|
|
|
|
# """
|
|
# response = requests.post(
|
|
# url=self.get_api(API_LISTS_CREATE), json={"title": title}, headers=self._access_token_()
|
|
# url=self.get_api(API_TODO_ITEMS_CREATE), json={"title": title}, headers=self._access_token_()
|
|
# )
|
|
# response.raise_for_status()
|
|
# return response.json()
|
|
|
|
# def read(self, id):
|
|
# """
|
|
# Read a to-do list contents
|
|
|
|
# Parameters
|
|
# ----------
|
|
# id : int
|
|
# List id.
|
|
|
|
# Returns
|
|
# -------
|
|
# list
|
|
# Requested contents
|
|
|
|
# """
|
|
# response = requests.post(
|
|
# url=self.get_api(API_LISTS_READ).format(id), headers=self._access_token_()
|
|
# )
|
|
# response.raise_for_status()
|
|
# return response.json()
|
|
|
|
# def update(self, id, title="Untitled"):
|
|
# """
|
|
# Add a to-do item to the list
|
|
|
|
# Parameters
|
|
# ----------
|
|
# id : int
|
|
# List id.
|
|
# title : str, optional
|
|
# To-do item title. The default is "Untitled".
|
|
|
|
# """
|
|
# response = requests.put(
|
|
# json={"title": title},
|
|
# url=self.get_api(API_LISTS_UPDATE).format(id),
|
|
# headers=self._access_token_(),
|
|
# )
|
|
# response.raise_for_status()
|
|
# return response.json()
|
|
|
|
# def partial_update(self, id, title="Untitled"):
|
|
# """
|
|
# Update list item - untrusted
|
|
|
|
# """
|
|
# response = requests.patch(
|
|
# json={"title": title},
|
|
# url=self.get_api(API_LISTS_PARTIAL_UPDATE).format(id),
|
|
# headers=self._access_token_(),
|
|
# )
|
|
# response.raise_for_status()
|
|
# return response.json()
|
|
|
|
# def delete(self, id):
|
|
# """
|
|
# Delete list
|
|
|
|
# Parameters
|
|
# ----------
|
|
# id : int
|
|
# List id to delete.
|
|
|
|
# """
|
|
# response = requests.delete(
|
|
# url=self.get_api(API_LISTS_DELETE).format(id), headers=self._access_token_()
|
|
# )
|
|
# response.raise_for_status()
|
|
# return response.json()
|
|
|
|
def _access_token_(self):
|
|
if self.token is None:
|
|
raise RuntimeError("Authosization required for requested operation!")
|
|
return {"Authorization": f"Bearer {self.token.access}"}
|
|
|
|
@staticmethod
|
|
def _raise_or_return_(response):
|
|
response.raise_for_status()
|
|
return response.json()
|