😊

【Webサービス開発】コーヒーショップアプリ

2023/04/05に公開

Flask

概要

コーヒーショップで、飲み物をオンライン上で注文できるアプリを開発する。画像は一番最後に記載する。

ユーザは、システムへログインして飲み物を注文（＊画像１）
飲み物は、いくつかの選択肢から注文。（＊画像２）
ユーザとは別に管理者アカウントがあり、飲み物の編集、追加、削除が可能（＊画像３）

事前準備

GitHub

今回は、GitHubの下記レポジトリを使用する。フォルダは、Project -> 03_coffee_shop_full_stackである。

リポジトリを落としたら、Readmeに記載の通り、まずはserverを立ち上げる。

pip install -r requirements.txt
export FLASK_APP=api.py;
flask run --reload

もし完成版のGitHubがみたい場合は、下記GitHubを確認すること

Auth0.com

Applications

Auth0で新しくweb applicationを作成する。自分の場合、CoffeeShopというアプリを立ち上げた。

Application URIsにて、下記の通り設定する。

Application Login URI: http://localhost:8080/login
Allowed Callback URLs: http://localhost:8100, http://localhost:8100/tabs/user-page
Allowed Logout URLs: http://localhost:8100, http://localhost:8080
Allowed Web Origins: http://localhost:8100, http://localhost:8080

そして最後に、CORSの許可をする。

APIs

次にAPIを新規作成する。今回は、CoffeeShopAPIとしてAPIを作成した。

APIを作成する上で、新しいAPIパーミッションを作成する必要がある。作成するパーミッションは以下の通り。

次に、BaristaとManagerという役割を作成して、以下の通り権限を割り当てる。ちなみに、ManagerはO、BaristaはYXというユーザに割り当てている。

Barista
- get:drinks-detail
- get:drinks
Manager
- 全てのアクション
  
  上記の通り、BaristaとManagerでアクションを分けると、下記の通り異なる画面が表示される。
  ManagerはCreate Drinkができる。
  
  一方で、BaristaはCreate Drinkができない。

Tokenを取得

最後にTokenを取得する。Tokenを取得するためのURLは下記の通り。

https://YOUR_DOMAIN/authorize?audience=API_IDENTIFIER&scope=SCOPE&response_type=code&client_id=YOUR_CLIENT_ID&redirect_uri=http://localhost:8080/login-results

YOUR_DOMAIN / YOUR_CLIENT_ID: 下記に記載
API_IDENTIFIER: 下記に記載

最後に実際にtokenを取得してみる。上記URLへアクセスすると、下記の通り画面表示される。こちらで、BaristaとManagerでアサインしたメールアドレス其々でアクセスする。そうすると、下記の通りtokenが取得できる。下記文章のaccess_token=という部分のみコピーして、BaristaとManager其々のアクセストークンを取得する。

http://localhost:8080/login-results#access_token=eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZ[…]bJO1OJytE_DTJAO3rb0w&expires_in=7200&token_type=Bearer

コード

auth.py

上記ファイルは、tokenを取得するために下記の通り書き換える。

import json
from flask import request, _request_ctx_stack, abort
from functools import wraps
from jose import jwt
from urllib.request import urlopen


AUTH0_DOMAIN = '@TODO: 自分のAUTH_DOMAIN'
ALGORITHMS = ['RS256']
API_AUDIENCE = '@TODO: 自分のAUTH_API'

## AuthError Exception
'''
AuthError Exception
A standardized way to communicate auth failure modes
'''
class AuthError(Exception):
    def __init__(self, error, status_code):
        self.error = error
        self.status_code = status_code


## Auth Header

'''
@TODO implement get_token_auth_header() method
    it should attempt to get the header from the request
        it should raise an AuthError if no header is present
    it should attempt to split bearer and the token
        it should raise an AuthError if the header is malformed
    return the token part of the header
'''
def get_token_auth_header():
    if 'Authorization' not in request.headers: 
        abort(401)
    auth_header = request.headers['Authorization']
    header_parts = auth_header.split('.')
    if len(header_parts) != 3:
        abort(401)
    return header_parts[1]

'''
@TODO implement check_permissions(permission, payload) method
    @INPUTS
        permission: string permission (i.e. 'post:drink')
        payload: decoded jwt payload

    it should raise an AuthError if permissions are not included in the payload
        !!NOTE check your RBAC settings in Auth0
    it should raise an AuthError if the requested permission string is not in the payload permissions array
    return true otherwise
'''
def check_permissions(permission, payload):
    if 'permission' not in payload:
        abort(403)
    if permission not in payload['permission']:
        abort(403)
    return True
'''
@TODO implement verify_decode_jwt(token) method
    @INPUTS
        token: a json web token (string)

    it should be an Auth0 token with key id (kid)
    it should verify the token using Auth0 /.well-known/jwks.json
    it should decode the payload from the token
    it should validate the claims
    return the decoded payload

    !!NOTE urlopen has a common certificate error described here: https://stackoverflow.com/questions/50236117/scraping-ssl-certificate-verify-failed-error-for-http-en-wikipedia-org
'''
def verify_decode_jwt(token):
    jsonurl = urlopen(f'https://{AUTH0_DOMAIN}/.well-known/jwks.json')
    jwks = json.loads(jsonurl.read())
    unverified_header = jwt.get_unverified_header(token)
    rsa_key = {}
    if 'kid' not in unverified_header:
        raise AuthError({
            'code': 'invalid_header',
            'description': 'Authorization malformed.'
        }, 401)

    for key in jwks['keys']:
        if key['kid'] == unverified_header['kid']:
            rsa_key = {
                'kty': key['kty'],
                'kid': key['kid'],
                'use': key['use'],
                'n': key['n'],
                'e': key['e']
            }
    if rsa_key:
        try:
            payload = jwt.decode(
                token,
                rsa_key,
                algorithms=ALGORITHMS,
                audience=API_AUDIENCE,
                issuer='https://' + AUTH0_DOMAIN + '/'
            )

            return payload

        except jwt.ExpiredSignatureError:
            raise AuthError({
                'code': 'token_expired',
                'description': 'Token expired.'
            }, 401)

        except jwt.JWTClaimsError:
            raise AuthError({
                'code': 'invalid_claims',
                'description': 'Incorrect claims. Please, check the audience and issuer.'
            }, 401)
        except Exception:
            raise AuthError({
                'code': 'invalid_header',
                'description': 'Unable to parse authentication token.'
            }, 400)
    raise AuthError({
                'code': 'invalid_header',
                'description': 'Unable to find the appropriate key.'
            }, 400)


'''
@TODO implement @requires_auth(permission) decorator method
    @INPUTS
        permission: string permission (i.e. 'post:drink')

    it should use the get_token_auth_header method to get the token
    it should use the verify_decode_jwt method to decode the jwt
    it should use the check_permissions method validate claims and check the requested permission
    return the decorator which passes the decoded payload to the decorated method
'''
def requires_auth(permission=''):
    def requires_auth_decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            token = get_token_auth_header()
            try:
                payload = verify_decode_jwt(token)
            except:
                abort(401)
            check_permissions(permission, payload)
            return f(payload, *args, **kwargs)
        return wrapper
    return requires_auth_decorator

api.py

上記ファイルは、下記の通り書き換える

import os
from flask import Flask, request, jsonify, abort
from sqlalchemy import exc
import json
from flask_cors import CORS

from .database.models import db_drop_and_create_all, setup_db, Drink
from .auth.auth import AuthError, requires_auth

app = Flask(__name__)
setup_db(app)
CORS(app)

'''
@TODO uncomment the following line to initialize the datbase
!! NOTE THIS WILL DROP ALL RECORDS AND START YOUR DB FROM SCRATCH
!! NOTE THIS MUST BE UNCOMMENTED ON FIRST RUN
!! Running this funciton will add one
'''
db_drop_and_create_all()

# ROUTES
'''
@TODO implement endpoint
    GET /drinks
        it should be a public endpoint
        it should contain only the drink.short() data representation
    returns status code 200 and json {"success": True, "drinks": drinks} where drinks is the list of drinks
        or appropriate status code indicating reason for failure
'''

@app.route('/drinks', methods=['GET'])
def get_drinks():
    print("test")
    try:
        drinksShortList = [drink.short() for drink in Drink.query.all()]
        print(drinksShortList)
        return json.dumps({
            'success': True,
            'drinks': drinksShortList
        }), 200
    except:
        return json.dumps({
            'success': False,
            'error': "Error with loading drinks occured"
        }), 500
'''
@TODO implement endpoint
    GET /drinks-detail
        it should require the 'get:drinks-detail' permission
        it should contain the drink.long() data representation
    returns status code 200 and json {"success": True, "drinks": drinks} where drinks is the list of drinks
        or appropriate status code indicating reason for failure
'''
@app.route('/drinks-detail')
@requires_auth('get:drinks-detail')
def get_drinks_detail(jwt):
    print(jwt)
    return 'Access Granted'
    

'''
@TODO implement endpoint
    POST /drinks
        it should create a new row in the drinks table
        it should require the 'post:drinks' permission
        it should contain the drink.long() data representation
    returns status code 200 and json {"success": True, "drinks": drink} where drink an array containing only the newly created drink
        or appropriate status code indicating reason for failure
'''
@app.route('/drinks', methods=['POST'], endpoint='post_drink')
@requires_auth('post:drinks')
def post_drinks_detail(jwt):
    print(jwt)
    return {"success": True, "drinks": drink}


'''
@TODO implement endpoint
    PATCH /drinks/<id>
        where <id> is the existing model id
        it should respond with a 404 error if <id> is not found
        it should update the corresponding row for <id>
        it should require the 'patch:drinks' permission
        it should contain the drink.long() data representation
    returns status code 200 and json {"success": True, "drinks": drink} where drink an array containing only the updated drink
        or appropriate status code indicating reason for failure
'''


'''
@TODO implement endpoint
    DELETE /drinks/<id>
        where <id> is the existing model id
        it should respond with a 404 error if <id> is not found
        it should delete the corresponding row for <id>
        it should require the 'delete:drinks' permission
    returns status code 200 and json {"success": True, "delete": id} where id is the id of the deleted record
        or appropriate status code indicating reason for failure
'''


# Error Handling
'''
Example error handling for unprocessable entity
'''


@app.errorhandler(422)
def unprocessable(error):
    return jsonify({
        "success": False,
        "error": 422,
        "message": "unprocessable"
    }), 422


'''
@TODO implement error handlers using the @app.errorhandler(error) decorator
    each error handler should return (with approprate messages):
             jsonify({
                    "success": False,
                    "error": 404,
                    "message": "resource not found"
                    }), 404

'''

'''
@TODO implement error handler for 404
    error handler should conform to general task above
'''


'''
@TODO implement error handler for AuthError
    error handler should conform to general task above
'''

接続テスト

上記作業が完了したら、最後に接続テストを行う。まずはlocalのbackendフォルダで、アプリケーションが立ち上がっていることを確認する。

次にDB画作成されていることを確認する。DBはdatabaseフォルダへ移動して、下記の通りコマンドを実行すると確認できる。

sqlite3 database.db
.tables
SELECT * FROM drink;

念のため、下記URLでDB情報が表示されることを確認する。
http://127.0.0.1:5000/drinks

最後に、Postmanでcollectionをimportする。importタブをクリックして、下記フォルダのcollection.jsonファイルをインポートする。
./starter_code/backend/udacity-fsnd-udaspicelatte.postman_collection.json

そうすると、添付のようにudacity-fsnd-udaspicelatteフォルダが表示されるので、drinksを右クリックする。ここで、Managerのtokenを入力して、URLを書き換えた上でSendをクリックする。そうすると、下記の通り緑色でPASS表示される。

認証アクションを整理

下記5つの認証アクションを整理する必要がある。

get:drinks
get:drinks-detail
post:drinks
patch:drinks
delete:drinks

get:drinks

上記の接続テストで確認済み。@requires_auth('*')による認証が不要で、情報をGETできればOK。

@app.route('/drinks', methods=['GET'])
def get_drinks():
    print("test")
    try:
        drinksShortList = [drink.short() for drink in Drink.query.all()]
        print(drinksShortList)
        return json.dumps({
            'success': True,
            'drinks': drinksShortList
        }), 200
    except:
        return json.dumps({
            'success': False,
            'error': "Error with loading drinks occured"
        }), 500

get:drinks-detail

下記の通り、@requires_authによるユーザ認証を行う。単純に、Drinkテーブルに含まれている情報を取得して、返却する形となっている。ユーザ認証の具体的な内容については、下記記事を参考にすること。

@app.route('/drinks-detail', methods=['GET'])
@requires_auth('get:drinks-detail')
def get_drinks_detail(payload):
    try:
        drinksLongList = [drink.long() for drink in Drink.query.all()]
        return jsonify({
            'success': True,
            'drinks': drinksLongList
        }), 200
    except:
        return json.dumps({
            'success': False,
            'error': "Error with loading drinks occured"
        }), 500

post:drinks

post:drinksは下記の通り。こちらは、managerアカウントは実行できて、baristaアカウントは実行できない形となっている。コードは、requestを受けたらそれをbodyへ格納する。格納したら、それぞれtitle / recipeといった形で保管して、new_drinkとしてDrinkテーブルに代入する。最終的に、.insert()でpostする形となっている。

@app.route('/drinks', methods=['POST'])
@requires_auth('post:drinks')
def post_drinks_detail(payload):
    body = request.get_json()
    try:
        title = body.get('title')
        recipe = json.dumps(body.get('recipe'))
        new_drink = Drink(title=title, recipe=recipe)
        print(new_drink)
        new_drink.insert()

        return jsonify({
            'success': True,
            'drinks': [new_drink.long()]
        }), 200
    except Exception as e:
        print(e)

patch:drinks

以下に説明を記載する

まず最初に、データベースから、drink_idに一致するドリンクオブジェクトを検索して、one_or_none()メソッドで、結果が一つだけの場合はオブジェクトを返して、それ以外の場合はNoneとしている。
次に、drink_id.titleとdrink_id.recipeの値を、クライアントの送信データであるbodyに基づいて更新している。この場合、データをJSON形式に変換してから、drink_id.update()で更新している。

def update_drinks_detail(payload, drink_id):
    body = request.get_json()
    try:
        drink_id = Drink.query.filter(Drink.id==drink_id).one_or_none()
        drink_id.title = body.get('title')
        drink_id.recipe = json.dumps(body.get('recipe'))
        drink_id.update()

        return jsonify({
            'success': True,
            'drinks': [drink_id.long()]
        }), 200
    except Exception as e:
        print(e)
        return json.dumps({
            'success': False,
            'error': f"Error with creating a new drink: {str(e)}"
        }), 500

delete:drinks

以下に説明を記載する

/drinks/int:drink_idというエンドポイントに対して、DELETEリクエストを送る。
requires_authによって、リクエストが適切な認証情報を持っていることを確認する。これは、delete:drinks権限を持つユーザのみが、アクセスできる。
delete_drinks関数は、payloadという認証デコレータの情報とdrink_idという削除するドリンクIDを引数として受け取る
関数内では、データベース上でdrink_idと一致するドリンクオブジェクトを検索している。one_or_none()メソッドに従って、結果が一つだけの場合はそのオブジェクトを返して、それ以外の場合はNoneとかえす。そして、そのドリンクをdrink.delete()で削除している。

@app.route('/drinks/<int:drink_id>', methods=['DELETE'])
@requires_auth('delete:drinks')
def delete_drinks(payload, drink_id):
    try:
        drink = Drink.query.filter(Drink.id==drink_id).one_or_none()
        print(drink_id)
        drink.delete()

        return jsonify({
            'success': True,
            'drinks': drink_id
        }), 200
    except Exception as e:
        print(e)
        return json.dumps({
            'success': False,
            'error': f"Error with creating a new drink: {str(e)}"
        }), 500

Postmanによる接続確認

ちなみに、上記の接続設定をしたあと、実際にPostmanで正しくアクションを実行できるか確認できる。その場合、まずは下記ファイルをPostmanでインポートする必要がある。

インポートしたら、下記画面が表示される。今回はいくつか例をあげて、Postmanによる接続確認を行う。

get:drinks

get:drinksはType=No Authとする。認証が不要なためである。GET http://127.0.0.1:5000/drinks として、Sendボタンを押下することで、接続確認ができる。そうすると、結果が返却される。

drinks-detail

GET http://127.0.0.1:5000/drinks-detail とする。あとは、Bear Tokenを設定する必要がある。そのため、TypeをBearer TokenとしてToken情報を書き込む。Token情報取得については、下記記事を参考とすること。

上記内容でレスポンスを送ると、200ステータスコードで返却される。

post:drinks　/ patch drinks

post:drinkとpatch drinkは上記のtokenに加えて、postする情報や変更する内容をbodyタブで設定することで、postmanによる処理ができる。

その他

上記内容は全て、managerタブで行なっている。manager権限ではなく、barista権限によるテストはbaristaタブで実行できる。baristaについては、delete:drinksやpost:drinksは許可されていないので、こちらもbarista用のtokenを発行して、postmanによるテストを行うと良い。

画像

画像１

画像２

画像３

概要

事前準備

GitHub

Auth0.com

Applications

APIs

Tokenを取得

コード

auth.py

api.py

接続テスト

認証アクションを整理

get:drinks

get:drinks-detail

post:drinks

patch:drinks

delete:drinks

Postmanによる接続確認

get:drinks

drinks-detail

post:drinks / patch drinks

その他

画像

Discussion

post:drinks　/ patch drinks