Files
django-todo/frontend/api.py

334 lines
8.1 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 = "api/lists/{0}/"
API_LISTS_UPDATE = "api/lists/{0}/"
API_LISTS_PARTIAL_UPDATE = "api/lists/{0}/"
API_LISTS_DELETE = "api/lists/{0}/"
API_TOKEN_CREATE = "api/token/"
API_TOKEN_REFRESH = "api/token/refresh/"
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_CREATE), json={"username": user, "password": passwd}
)
)
self.token = SimpleNamespace(**token)
return self.token
def refresh(self):
"""
Refresh existing token
"""
token = UserApi._raise_or_return_(
requests.post(url=self.get_api(API_TOKEN_REFRESH), json={"refresh": self.token.refresh})
)
self.token.access = token["access"]
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 lists_delete(self, id):
"""
Auth required
Deletes a to-do list by id
Parameters
----------
id: to-do list id to delete
"""
return UserApi._raise_or_return_(
requests.delete(
url=self.get_api(API_LISTS_DELETE.format(id)),
headers=self._access_token_(),
)
)
def lists_update(self, title, id):
"""
Rename a new to-do list
Auth required
Parameters
----------
title : str
New name for a list.
id : int
"""
return UserApi._raise_or_return_(
requests.put(
url=self.get_api(API_LISTS_UPDATE.format(id)),
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 todo_items_create(self, parent, text="Note"):
"""
Create a new to-do item
Auth required
Parameters
----------
parent : id of parent list
text : str, optional
New note. The default is "Note".
"""
return UserApi._raise_or_return_(
requests.post(
url=self.get_api(API_TODO_ITEMS_CREATE),
json={"text": text, "parent": parent, "finished": False},
headers=self._access_token_(),
)
)
def todo_items_delete(self, id):
"""
Auth required
Deletes a to-do item by id
Parameters
----------
id: to-do item id to delete
"""
return UserApi._raise_or_return_(
requests.delete(
url=self.get_api(API_TODO_ITEMS_DELETE.format(id)),
headers=self._access_token_(),
)
)
def todo_items_update(self, id, text, finished, parent):
"""
Rename a new to-do list
Auth required
Parameters
----------
id : int
Note id
text : str
New note for the item.
finished : bool
New state for the item
parent : int
Parent list id
"""
return UserApi._raise_or_return_(
requests.put(
url=self.get_api(API_TODO_ITEMS_UPDATE.format(id)),
json={"text": text, "finished": finished, "parent": parent},
headers=self._access_token_(),
)
)
# 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()
try:
return response.json()
except Exception as e:
print(e)
return response.content