Skip to content

Commit 8a9b358

Browse files
a-lunaa-luna
committed
Part 6
1 parent 8eac3b4 commit 8a9b358

10 files changed

+607
-6
lines changed

‎src/flask_api_tutorial/api/widgets/business.py

+76-2
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,11 @@
22
from http import HTTPStatus
33

44
from flask import jsonify, url_for
5-
from flask_restx import abort
5+
from flask_restx import abort, marshal
66

77
from flask_api_tutorial import db
8-
from flask_api_tutorial.api.auth.decorators import admin_token_required
8+
from flask_api_tutorial.api.auth.decorators import token_required, admin_token_required
9+
from flask_api_tutorial.api.widgets.dto import pagination_model, widget_name
910
from flask_api_tutorial.models.user import User
1011
from flask_api_tutorial.models.widget import Widget
1112

@@ -25,3 +26,76 @@ def create_widget(widget_dict):
2526
response.status_code = HTTPStatus.CREATED
2627
response.headers["Location"] = url_for("api.widget", name=name)
2728
return response
29+
30+
31+
@token_required
32+
def retrieve_widget_list(page, per_page):
33+
pagination = Widget.query.paginate(page, per_page, error_out=False)
34+
response_data = marshal(pagination, pagination_model)
35+
response_data["links"] = _pagination_nav_links(pagination)
36+
response = jsonify(response_data)
37+
response.headers["Link"] = _pagination_nav_header_links(pagination)
38+
response.headers["Total-Count"] = pagination.total
39+
return response
40+
41+
42+
@token_required
43+
def retrieve_widget(name):
44+
return Widget.query.filter_by(name=name.lower()).first_or_404(
45+
description=f"{name} not found in database."
46+
)
47+
48+
49+
@admin_token_required
50+
def update_widget(name, widget_dict):
51+
widget = Widget.find_by_name(name.lower())
52+
if widget:
53+
for k, v in widget_dict.items():
54+
setattr(widget, k, v)
55+
db.session.commit()
56+
message = f"'{name}' was successfully updated"
57+
response_dict = dict(status="success", message=message)
58+
return response_dict, HTTPStatus.OK
59+
try:
60+
valid_name = widget_name(name.lower())
61+
except ValueError as e:
62+
abort(HTTPStatus.BAD_REQUEST, str(e), status="fail")
63+
widget_dict["name"] = valid_name
64+
return create_widget(widget_dict)
65+
66+
67+
@admin_token_required
68+
def delete_widget(name):
69+
widget = Widget.query.filter_by(name=name.lower()).first_or_404(
70+
description=f"{name} not found in database."
71+
)
72+
db.session.delete(widget)
73+
db.session.commit()
74+
return "", HTTPStatus.NO_CONTENT
75+
76+
77+
def _pagination_nav_links(pagination):
78+
nav_links = {}
79+
per_page = pagination.per_page
80+
this_page = pagination.page
81+
last_page = pagination.pages
82+
nav_links["self"] = url_for("api.widget_list", page=this_page, per_page=per_page)
83+
nav_links["first"] = url_for("api.widget_list", page=1, per_page=per_page)
84+
if pagination.has_prev:
85+
nav_links["prev"] = url_for(
86+
"api.widget_list", page=this_page - 1, per_page=per_page
87+
)
88+
if pagination.has_next:
89+
nav_links["next"] = url_for(
90+
"api.widget_list", page=this_page + 1, per_page=per_page
91+
)
92+
nav_links["last"] = url_for("api.widget_list", page=last_page, per_page=per_page)
93+
return nav_links
94+
95+
96+
def _pagination_nav_header_links(pagination):
97+
url_dict = _pagination_nav_links(pagination)
98+
link_header = ""
99+
for rel, url in url_dict.items():
100+
link_header += f'<{url}>; rel="{rel}", '
101+
return link_header.strip().strip(",")

‎src/flask_api_tutorial/api/widgets/dto.py

+49-1
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,9 @@
33
from datetime import date, datetime, time, timezone
44

55
from dateutil import parser
6-
from flask_restx.inputs import URL
6+
from flask_restx import Model
7+
from flask_restx.fields import Boolean, DateTime, Integer, List, Nested, String, Url
8+
from flask_restx.inputs import positive, URL
79
from flask_restx.reqparse import RequestParser
810

911
from flask_api_tutorial.util.datetime_util import make_tzaware, DATE_MONTH_NAME
@@ -66,3 +68,49 @@ def future_date_from_string(date_str):
6668
required=True,
6769
nullable=False,
6870
)
71+
72+
update_widget_reqparser = create_widget_reqparser.copy()
73+
update_widget_reqparser.remove_argument("name")
74+
75+
pagination_reqparser = RequestParser(bundle_errors=True)
76+
pagination_reqparser.add_argument("page", type=positive, required=False, default=1)
77+
pagination_reqparser.add_argument(
78+
"per_page", type=positive, required=False, choices=[5, 10, 25, 50, 100], default=10
79+
)
80+
81+
widget_owner_model = Model("Widget Owner", {"email": String, "public_id": String})
82+
83+
widget_model = Model(
84+
"Widget",
85+
{
86+
"name": String,
87+
"info_url": String,
88+
"created_at": String(attribute="created_at_str"),
89+
"created_at_iso8601": DateTime(attribute="created_at"),
90+
"created_at_rfc822": DateTime(attribute="created_at", dt_format="rfc822"),
91+
"deadline": String(attribute="deadline_str"),
92+
"deadline_passed": Boolean,
93+
"time_remaining": String(attribute="time_remaining_str"),
94+
"owner": Nested(widget_owner_model),
95+
"link": Url("api.widget"),
96+
},
97+
)
98+
99+
pagination_links_model = Model(
100+
"Nav Links",
101+
{"self": String, "prev": String, "next": String, "first": String, "last": String},
102+
)
103+
104+
pagination_model = Model(
105+
"Pagination",
106+
{
107+
"links": Nested(pagination_links_model, skip_none=True),
108+
"has_prev": Boolean,
109+
"has_next": Boolean,
110+
"page": Integer,
111+
"total_pages": Integer(attribute="pages"),
112+
"items_per_page": Integer(attribute="per_page"),
113+
"total_items": Integer(attribute="total"),
114+
"items": List(Nested(widget_model)),
115+
},
116+
)

‎src/flask_api_tutorial/api/widgets/endpoints.py

+64-2
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,28 @@
33

44
from flask_restx import Namespace, Resource
55

6-
from flask_api_tutorial.api.widgets.dto import create_widget_reqparser
7-
from flask_api_tutorial.api.widgets.business import create_widget
6+
from flask_api_tutorial.api.widgets.dto import (
7+
create_widget_reqparser,
8+
update_widget_reqparser,
9+
pagination_reqparser,
10+
widget_owner_model,
11+
widget_model,
12+
pagination_links_model,
13+
pagination_model,
14+
)
15+
from flask_api_tutorial.api.widgets.business import (
16+
create_widget,
17+
retrieve_widget_list,
18+
retrieve_widget,
19+
update_widget,
20+
delete_widget,
21+
)
822

923
widget_ns = Namespace(name="widgets", validate=True)
24+
widget_ns.models[widget_owner_model.name] = widget_owner_model
25+
widget_ns.models[widget_model.name] = widget_model
26+
widget_ns.models[pagination_links_model.name] = pagination_links_model
27+
widget_ns.models[pagination_model.name] = pagination_model
1028

1129

1230
@widget_ns.route("", endpoint="widget_list")
@@ -16,6 +34,16 @@
1634
class WidgetList(Resource):
1735
"""Handles HTTP requests to URL: /widgets."""
1836

37+
@widget_ns.doc(security="Bearer")
38+
@widget_ns.response(HTTPStatus.OK, "Retrieved widget list.", pagination_model)
39+
@widget_ns.expect(pagination_reqparser)
40+
def get(self):
41+
"""Retrieve a list of widgets."""
42+
request_data = pagination_reqparser.parse_args()
43+
page = request_data.get("page")
44+
per_page = request_data.get("per_page")
45+
return retrieve_widget_list(page, per_page)
46+
1947
@widget_ns.doc(security="Bearer")
2048
@widget_ns.response(int(HTTPStatus.CREATED), "Added new widget.")
2149
@widget_ns.response(int(HTTPStatus.FORBIDDEN), "Administrator token required.")
@@ -25,3 +53,37 @@ def post(self):
2553
"""Create a widget."""
2654
widget_dict = create_widget_reqparser.parse_args()
2755
return create_widget(widget_dict)
56+
57+
58+
@widget_ns.route("/<name>", endpoint="widget")
59+
@widget_ns.param("name", "Widget name")
60+
@widget_ns.response(int(HTTPStatus.BAD_REQUEST), "Validation error.")
61+
@widget_ns.response(int(HTTPStatus.NOT_FOUND), "Widget not found.")
62+
@widget_ns.response(int(HTTPStatus.UNAUTHORIZED), "Unauthorized.")
63+
@widget_ns.response(int(HTTPStatus.INTERNAL_SERVER_ERROR), "Internal server error.")
64+
class Widget(Resource):
65+
"""Handles HTTP requests to URL: /widgets/{name}."""
66+
67+
@widget_ns.doc(security="Bearer")
68+
@widget_ns.response(int(HTTPStatus.OK), "Retrieved widget.", widget_model)
69+
@widget_ns.marshal_with(widget_model)
70+
def get(self, name):
71+
"""Retrieve a widget."""
72+
return retrieve_widget(name)
73+
74+
@widget_ns.doc(security="Bearer")
75+
@widget_ns.response(int(HTTPStatus.OK), "Widget was updated.", widget_model)
76+
@widget_ns.response(int(HTTPStatus.CREATED), "Added new widget.")
77+
@widget_ns.response(int(HTTPStatus.FORBIDDEN), "Administrator token required.")
78+
@widget_ns.expect(update_widget_reqparser)
79+
def put(self, name):
80+
"""Update a widget."""
81+
widget_dict = update_widget_reqparser.parse_args()
82+
return update_widget(name, widget_dict)
83+
84+
@widget_ns.doc(security="Bearer")
85+
@widget_ns.response(int(HTTPStatus.NO_CONTENT), "Widget was deleted.")
86+
@widget_ns.response(int(HTTPStatus.FORBIDDEN), "Administrator token required.")
87+
def delete(self, name):
88+
"""Delete a widget."""
89+
return delete_widget(name)

‎tests/conftest.py

+9-1
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
from flask_api_tutorial import create_app
55
from flask_api_tutorial import db as database
66
from flask_api_tutorial.models.user import User
7-
from tests.util import EMAIL, PASSWORD
7+
from tests.util import EMAIL, ADMIN_EMAIL, PASSWORD
88

99

1010
@pytest.fixture
@@ -32,3 +32,11 @@ def user(db):
3232
db.session.add(user)
3333
db.session.commit()
3434
return user
35+
36+
37+
@pytest.fixture
38+
def admin(db):
39+
admin = User(email=ADMIN_EMAIL, password=PASSWORD, admin=True)
40+
db.session.add(admin)
41+
db.session.commit()
42+
return admin

‎tests/test_create_widget.py

+89
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
"""Unit tests for POST requests sent to api.widget_list API endpoint."""
2+
from datetime import date, timedelta
3+
from http import HTTPStatus
4+
5+
import pytest
6+
7+
from tests.util import (
8+
EMAIL,
9+
ADMIN_EMAIL,
10+
BAD_REQUEST,
11+
FORBIDDEN,
12+
DEFAULT_NAME,
13+
login_user,
14+
create_widget,
15+
)
16+
17+
18+
@pytest.mark.parametrize("widget_name", ["abc123", "widget-name", "new_widget1"])
19+
def test_create_widget_valid_name(client, db, admin, widget_name):
20+
response = login_user(client, email=ADMIN_EMAIL)
21+
assert "access_token" in response.json
22+
access_token = response.json["access_token"]
23+
response = create_widget(client, access_token, widget_name=widget_name)
24+
assert response.status_code == HTTPStatus.CREATED
25+
assert "status" in response.json and response.json["status"] == "success"
26+
success = f"New widget added: {widget_name}."
27+
assert "message" in response.json and response.json["message"] == success
28+
location = f"http://localhost/api/v1/widgets/{widget_name}"
29+
assert "Location" in response.headers and response.headers["Location"] == location
30+
31+
32+
@pytest.mark.parametrize(
33+
"deadline_str",
34+
[
35+
date.today().strftime("%m/%d/%Y"),
36+
date.today().strftime("%Y-%m-%d"),
37+
(date.today() + timedelta(days=3)).strftime("%b %d %Y"),
38+
],
39+
)
40+
def test_create_widget_valid_deadline(client, db, admin, deadline_str):
41+
response = login_user(client, email=ADMIN_EMAIL)
42+
assert "access_token" in response.json
43+
access_token = response.json["access_token"]
44+
response = create_widget(client, access_token, deadline_str=deadline_str)
45+
assert response.status_code == HTTPStatus.CREATED
46+
assert "status" in response.json and response.json["status"] == "success"
47+
success = f"New widget added: {DEFAULT_NAME}."
48+
assert "message" in response.json and response.json["message"] == success
49+
location = f"http://localhost/api/v1/widgets/{DEFAULT_NAME}"
50+
assert "Location" in response.headers and response.headers["Location"] == location
51+
52+
53+
@pytest.mark.parametrize(
54+
"deadline_str",
55+
[
56+
"1/1/1970",
57+
(date.today() - timedelta(days=3)).strftime("%Y-%m-%d"),
58+
"a long time ago, in a galaxy far, far away",
59+
],
60+
)
61+
def test_create_widget_invalid_deadline(client, db, admin, deadline_str):
62+
response = login_user(client, email=ADMIN_EMAIL)
63+
assert "access_token" in response.json
64+
access_token = response.json["access_token"]
65+
response = create_widget(client, access_token, deadline_str=deadline_str)
66+
assert response.status_code == HTTPStatus.BAD_REQUEST
67+
assert "message" in response.json and response.json["message"] == BAD_REQUEST
68+
assert "errors" in response.json and "deadline" in response.json["errors"]
69+
70+
71+
def test_create_widget_already_exists(client, db, admin):
72+
response = login_user(client, email=ADMIN_EMAIL)
73+
assert "access_token" in response.json
74+
access_token = response.json["access_token"]
75+
response = create_widget(client, access_token)
76+
assert response.status_code == HTTPStatus.CREATED
77+
response = create_widget(client, access_token)
78+
assert response.status_code == HTTPStatus.CONFLICT
79+
name_conflict = f"Widget name: {DEFAULT_NAME} already exists, must be unique."
80+
assert "message" in response.json and response.json["message"] == name_conflict
81+
82+
83+
def test_create_widget_no_admin_token(client, db, user):
84+
response = login_user(client, email=EMAIL)
85+
assert "access_token" in response.json
86+
access_token = response.json["access_token"]
87+
response = create_widget(client, access_token)
88+
assert response.status_code == HTTPStatus.FORBIDDEN
89+
assert "message" in response.json and response.json["message"] == FORBIDDEN

‎tests/test_delete_widget.py

+40
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
"""Test cases for GET requests sent to the api.widget API endpoint."""
2+
from http import HTTPStatus
3+
4+
from tests.util import (
5+
ADMIN_EMAIL,
6+
EMAIL,
7+
DEFAULT_NAME,
8+
FORBIDDEN,
9+
login_user,
10+
create_widget,
11+
retrieve_widget,
12+
delete_widget,
13+
)
14+
15+
16+
def test_delete_widget(client, db, admin):
17+
response = login_user(client, email=ADMIN_EMAIL)
18+
assert "access_token" in response.json
19+
access_token = response.json["access_token"]
20+
response = create_widget(client, access_token)
21+
assert response.status_code == HTTPStatus.CREATED
22+
response = delete_widget(client, access_token, widget_name=DEFAULT_NAME)
23+
assert response.status_code == HTTPStatus.NO_CONTENT
24+
response = retrieve_widget(client, access_token, widget_name=DEFAULT_NAME)
25+
assert response.status_code == HTTPStatus.NOT_FOUND
26+
27+
28+
def test_delete_widget_no_admin_token(client, db, admin, user):
29+
response = login_user(client, email=ADMIN_EMAIL)
30+
assert "access_token" in response.json
31+
access_token = response.json["access_token"]
32+
response = create_widget(client, access_token)
33+
assert response.status_code == HTTPStatus.CREATED
34+
35+
response = login_user(client, email=EMAIL)
36+
assert "access_token" in response.json
37+
access_token = response.json["access_token"]
38+
response = delete_widget(client, access_token, widget_name=DEFAULT_NAME)
39+
assert response.status_code == HTTPStatus.FORBIDDEN
40+
assert "message" in response.json and response.json["message"] == FORBIDDEN

0 commit comments

Comments
 (0)