Merge pull request #5 from a-luna/part-5

Part 5

Author: a-luna

migrations/versions/7c66df0e878f_add_widget_model.py

@@ -0,0 +1,39 @@
+"""add widget model
+
+Revision ID: 7c66df0e878f
+Revises: 079d26d45cc9
+Create Date: 2020-02-29 04:35:12.651711
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "7c66df0e878f"
+down_revision = "079d26d45cc9"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "widget",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("name", sa.String(length=100), nullable=False),
+        sa.Column("info_url", sa.String(length=255), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=True),
+        sa.Column("deadline", sa.DateTime(), nullable=True),
+        sa.Column("owner_id", sa.Integer(), nullable=False),
+        sa.ForeignKeyConstraint(["owner_id"], ["site_user.id"],),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("name"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("widget")
+    # ### end Alembic commands ###

run.py

@@ -1,13 +1,42 @@
 """Flask CLI/Application entry point."""
 import os
 
+import click
+
 from flask_api_tutorial import create_app, db
 from flask_api_tutorial.models.token_blacklist import BlacklistedToken
 from flask_api_tutorial.models.user import User
+from flask_api_tutorial.models.widget import Widget
 
 app = create_app(os.getenv("FLASK_ENV", "development"))
 
 
 @app.shell_context_processor
 def shell():
-    return {"db": db, "User": User, "BlacklistedToken": BlacklistedToken}
+    return {
+        "db": db,
+        "User": User,
+        "BlacklistedToken": BlacklistedToken,
+        "Widget": Widget,
+    }
+
+
+@app.cli.command("add-user", short_help="Add a new user")
+@click.argument("email")
+@click.option(
+    "--admin", is_flag=True, default=False, help="New user has administrator role"
+)
+@click.password_option(help="Do not set password on the command line!")
+def add_user(email, admin, password):
+    """Add a new user to the database with email address = EMAIL."""
+    if User.find_by_email(email):
+        error = f"Error: {email} is already registered"
+        click.secho(f"{error}\n", fg="red", bold=True)
+        return 1
+    new_user = User(email=email, password=password, admin=admin)
+    db.session.add(new_user)
+    db.session.commit()
+    user_type = "admin user" if admin else "user"
+    message = f"Successfully added new {user_type}:\n {new_user}"
+    click.secho(message, fg="blue", bold=True)
+    return 0

src/flask_api_tutorial/api/__init__.py

@@ -3,6 +3,7 @@
 from flask_restx import Api
 
 from flask_api_tutorial.api.auth.endpoints import auth_ns
+from flask_api_tutorial.api.widgets.endpoints import widget_ns
 
 api_bp = Blueprint("api", __name__, url_prefix="/api/v1")
 authorizations = {"Bearer": {"type": "apiKey", "in": "header", "name": "Authorization"}}
@@ -17,3 +18,4 @@
 )
 
 api.add_namespace(auth_ns, path="/auth")
+api.add_namespace(widget_ns, path="/widgets")

src/flask_api_tutorial/api/widgets/business.py

@@ -0,0 +1,27 @@
+"""Business logic for /widgets API endpoints."""
+from http import HTTPStatus
+
+from flask import jsonify, url_for
+from flask_restx import abort
+
+from flask_api_tutorial import db
+from flask_api_tutorial.api.auth.decorators import admin_token_required
+from flask_api_tutorial.models.user import User
+from flask_api_tutorial.models.widget import Widget
+
+
+@admin_token_required
+def create_widget(widget_dict):
+    name = widget_dict["name"]
+    if Widget.find_by_name(name):
+        error = f"Widget name: {name} already exists, must be unique."
+        abort(HTTPStatus.CONFLICT, error, status="fail")
+    widget = Widget(**widget_dict)
+    owner = User.find_by_public_id(create_widget.public_id)
+    widget.owner_id = owner.id
+    db.session.add(widget)
+    db.session.commit()
+    response = jsonify(status="success", message=f"New widget added: {name}.")
+    response.status_code = HTTPStatus.CREATED
+    response.headers["Location"] = url_for("api.widget", name=name)
+    return response

src/flask_api_tutorial/api/widgets/dto.py

@@ -0,0 +1,68 @@
+"""Parsers and serializers for /widgets API endpoints."""
+import re
+from datetime import date, datetime, time, timezone
+
+from dateutil import parser
+from flask_restx.inputs import URL
+from flask_restx.reqparse import RequestParser
+
+from flask_api_tutorial.util.datetime_util import make_tzaware, DATE_MONTH_NAME
+
+
+def widget_name(name):
+    """Validation method for a string containing only letters, numbers, '-' and '_'."""
+    if not re.compile(r"^[\w-]+$").match(name):
+        raise ValueError(
+            f"'{name}' contains one or more invalid characters. Widget name must "
+            "contain only letters, numbers, hyphen and underscore characters."
+        )
+    return name
+
+
+def future_date_from_string(date_str):
+    """Validation method for a date in the future, formatted as a string."""
+    try:
+        parsed_date = parser.parse(date_str)
+    except ValueError:
+        raise ValueError(
+            f"Failed to parse '{date_str}' as a valid date. You can use any format "
+            "recognized by dateutil.parser. For example, all of the strings below "
+            "are valid ways to represent the same date: '2018-5-13' -or- '05/13/2018' "
+            "-or- 'May 13 2018'."
+        )
+
+    if parsed_date.date() < date.today():
+        raise ValueError(
+            f"Successfully parsed {date_str} as "
+            f"{parsed_date.strftime(DATE_MONTH_NAME)}. However, this value must be a "
+            f"date in the future and {parsed_date.strftime(DATE_MONTH_NAME)} is BEFORE "
+            f"{datetime.now().strftime(DATE_MONTH_NAME)}"
+        )
+    deadline = datetime.combine(parsed_date.date(), time.max)
+    deadline_utc = make_tzaware(deadline, use_tz=timezone.utc)
+    return deadline_utc
+
+
+create_widget_reqparser = RequestParser(bundle_errors=True)
+create_widget_reqparser.add_argument(
+    "name",
+    type=widget_name,
+    location="form",
+    required=True,
+    nullable=False,
+    case_sensitive=True,
+)
+create_widget_reqparser.add_argument(
+    "info_url",
+    type=URL(schemes=["http", "https"]),
+    location="form",
+    required=True,
+    nullable=False,
+)
+create_widget_reqparser.add_argument(
+    "deadline",
+    type=future_date_from_string,
+    location="form",
+    required=True,
+    nullable=False,
+)

src/flask_api_tutorial/api/widgets/endpoints.py

@@ -0,0 +1,27 @@
+"""API endpoint definitions for /widgets namespace."""
+from http import HTTPStatus
+
+from flask_restx import Namespace, Resource
+
+from flask_api_tutorial.api.widgets.dto import create_widget_reqparser
+from flask_api_tutorial.api.widgets.business import create_widget
+
+widget_ns = Namespace(name="widgets", validate=True)
+
+
+@widget_ns.route("", endpoint="widget_list")
+@widget_ns.response(int(HTTPStatus.BAD_REQUEST), "Validation error.")
+@widget_ns.response(int(HTTPStatus.UNAUTHORIZED), "Unauthorized.")
+@widget_ns.response(int(HTTPStatus.INTERNAL_SERVER_ERROR), "Internal server error.")
+class WidgetList(Resource):
+    """Handles HTTP requests to URL: /widgets."""
+
+    @widget_ns.doc(security="Bearer")
+    @widget_ns.response(int(HTTPStatus.CREATED), "Added new widget.")
+    @widget_ns.response(int(HTTPStatus.FORBIDDEN), "Administrator token required.")
+    @widget_ns.response(int(HTTPStatus.CONFLICT), "Widget name already exists.")
+    @widget_ns.expect(create_widget_reqparser)
+    def post(self):
+        """Create a widget."""
+        widget_dict = create_widget_reqparser.parse_args()
+        return create_widget(widget_dict)

src/flask_api_tutorial/models/widget.py

@@ -0,0 +1,61 @@
+"""Class definition for Widget model."""
+from datetime import datetime, timezone, timedelta
+
+from sqlalchemy.ext.hybrid import hybrid_property
+
+from flask_api_tutorial import db
+from flask_api_tutorial.util.datetime_util import (
+    utc_now,
+    format_timedelta_str,
+    get_local_utcoffset,
+    localized_dt_string,
+    make_tzaware,
+)
+
+
+class Widget(db.Model):
+    """Widget model for a generic resource in a REST API."""
+
+    __tablename__ = "widget"
+
+    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
+    name = db.Column(db.String(100), unique=True, nullable=False)
+    info_url = db.Column(db.String(255))
+    created_at = db.Column(db.DateTime, default=utc_now)
+    deadline = db.Column(db.DateTime)
+
+    owner_id = db.Column(db.Integer, db.ForeignKey("site_user.id"), nullable=False)
+    owner = db.relationship("User", backref=db.backref("widgets"))
+
+    def __repr__(self):
+        return f"<Widget name={self.name}, info_url={self.info_url}>"
+
+    @hybrid_property
+    def created_at_str(self):
+        created_at_utc = make_tzaware(
+            self.created_at, use_tz=timezone.utc, localize=False
+        )
+        return localized_dt_string(created_at_utc, use_tz=get_local_utcoffset())
+
+    @hybrid_property
+    def deadline_str(self):
+        deadline_utc = make_tzaware(self.deadline, use_tz=timezone.utc, localize=False)
+        return localized_dt_string(deadline_utc, use_tz=get_local_utcoffset())
+
+    @hybrid_property
+    def deadline_passed(self):
+        return datetime.now(timezone.utc) > self.deadline.replace(tzinfo=timezone.utc)
+
+    @hybrid_property
+    def time_remaining(self):
+        time_remaining = self.deadline.replace(tzinfo=timezone.utc) - utc_now()
+        return time_remaining if not self.deadline_passed else timedelta(0)
+
+    @hybrid_property
+    def time_remaining_str(self):
+        timedelta_str = format_timedelta_str(self.time_remaining)
+        return timedelta_str if not self.deadline_passed else "No time remaining"
+
+    @classmethod
+    def find_by_name(cls, name):
+        return cls.query.filter_by(name=name).first()