<jupyter_start><jupyter_text>0-1　套件分類 在 Python 中，資料夾被視為 Package我們可以創造 models、resources 資料夾，並在裡面放入 `__init__.py` 檔案標示其為 Packagemodels 中的類別用來儲存及操作資料，例如：- item.find_by_name()- item.insert()通常 models 中的函式會是 classmethod (關於 classmethod 與 staticmethod 的 [補充說明](https://stackoverflow.com/questions/12179271/meaning-of-classmethod-and-staticmethod-for-beginner) )`item.find_by_name()` 在執行完後會回傳一個類別，故為 classmethod而 `item.insert()`、`item.update()`是插入或更新自己，故使用 self 而不會是 classmethodresources 裡的類別用來與外部用戶端溝通，例如 HTTP 方法：- item.get()- item.post()  0-2　分類實作 以下節錄說明各目錄下程式重點內容： - 主目錄下的 app.py<jupyter_code>from resources.user import UserRegister
from resources.item import Item, ItemList
from resources.store import Store, StoreList<jupyter_output><empty_output><jupyter_text>可以匯入在 resources 資料夾中的 user.py，並使用其中的 `UserRegister` 類別，以此類推 - resources 目錄下的 item.py<jupyter_code>from flask_restful import Resource, reqparse
from models.item import ItemModel

class Item(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('price',
                        type=float,
                        required=True,
                        help="This field cannot be left blank!"
                        )
    parser.add_argument('store_id',
                        type=int,
                        required=True,
                        help="Every item needs a store_id."
                        )

    def post(self, name):
        if ItemModel.find_by_name(name):
            return {'message': "An item with name '{}' already exists.".format(name)}, 400

        data = Item.parser.parse_args()
        item = ItemModel(name, **data)

        try:
            item.save_to_db()
        except:
            return {"message": "An error occurred inserting the item."}, 500

        return item.json(), 201<jupyter_output><empty_output><jupyter_text>resources 下的 item.py 裡面的函式會使用 models 下的 item.py 裡定義的函式被移到 models.item 的函式裡原本的 `self` 都改為使用 `ItemModel` 來呼叫 註一： `**data` 可還原成：`data['price'], data['store_id']`   　　　類似的作法是 `row[0], row[1]` 簡化成 `*row`，差別在有 key 時是兩個星號 註二：存入資料庫之動作適合使用 `try` 條件式，當錯誤發生時會執行 `except` 中之內容 - try：試著執行某段程式碼- except 某錯誤：當發生某錯誤時要進行的動作- else：無錯誤發生的時候進行動作- finally：無論有無錯誤最後都會進行的動作 ***  1　基本構成 在主目錄新增 db.py<jupyter_code>from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()<jupyter_output><empty_output><jupyter_text>如此創造一個 `SQLAlchemy` 物件來連結資料庫中的各個物件 - 主目錄下的 app.py<jupyter_code>@app.before_first_request
def create_tables():
    db.create_all()<jupyter_output><empty_output><jupyter_text>`before_first_request` 是來自 `flask` 的 decorator，`db.create_all()` 會讓 `SQLAlchemy` 自動產生所需要的表單 (table) 如果它們不存在<jupyter_code>if __name__ == '__main__':    
    from db import db
    db.init_app(app)
    app.run(port=5000, debug=True)<jupyter_output><empty_output><jupyter_text>由於 models 當中的 item 及 user 也會 import db為了避免 __circular import__，在 app.py 當中會選擇在此處 import 另外，可以在 app.py 中做以下設定：<jupyter_code>app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False<jupyter_output><empty_output><jupyter_text>將 Flask 版本的 SQLAlchemy 資料異動追蹤功能關閉，因為原始 SQLAlchemy 已有追蹤功能<jupyter_code>app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.db'<jupyter_output><empty_output><jupyter_text>指定目標資料庫的路徑，也可以是 PostgreSQL、SQLite、MySQL ***  2-1　實作語法 - 設定 - models 目錄下的 items.py<jupyter_code>from db import db

class ItemModel(db.Model):
    __tablename__ = 'items'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80))
    price = db.Column(db.Float(precision=2))
    store_id = db.Column(db.Integer, db.ForeignKey('stores.id'))
    store = db.relationship('StoreModel')<jupyter_output><empty_output><jupyter_text>`class ItemModel(db.Model)` 類別繼承 `db.Model` 以建立物件之間的連結關係`__tablename__ = 'items'` 在資料庫中創造名為 `items` 的表單`db.Column(db.資料型態, 主外鍵)` 建立欄位`db.relationship(model)` 指定外鍵所要連結的表單所在的 model<jupyter_code>def __init__(self, name, price, store_id):
        self.name = name
        self.price = price
        self.store_id = store_id<jupyter_output><empty_output><jupyter_text>由於 SQLAlchemy 會自動指定 `id`，故不需要輸入 `_id` 參數及加入 `self.id = _id`  2-2　實作語法 - 操作 - models 目錄下的 items.py<jupyter_code>@classmethod
    def find_by_name(cls, name):
        return cls.query.filter_by(name=name).first()<jupyter_output><empty_output><jupyter_text>SQL 語法對照：- `cls.query`　＝　`SELECT * FROM cls_table` (此處 cls_table 為 items)- `.filter_by(name=name)`　＝　`WHERE name=name`- `.first()`　＝　`LIMIT 1`SQLAlchemy 會將結果回傳並自動轉換為 `ItemModel` 物件 在 sqlite3 同樣功能的函式寫法會是：<jupyter_code>import sqlite3

    @classmethod
    def find_by_name(cls, name):
        connection = sqlite3.connect('data.db')
        cursor = connection.cursor()

        query = "SELECT * FROM items WHERE name=?"
        result = cursor.execute(query, (name,))
        row = result.fetchone()
        connection.close()

        if row:
            return cls(*row)<jupyter_output><empty_output><jupyter_text>註：在 sqlite3 的 execute query 需要傳入 tuple，單一元素的情況下寫成 `(element,)` 回到 SQLAlchemy，再加入其他函式：<jupyter_code>def save_to_db(self):
        db.session.add(self)
        db.session.commit()

    def delete_from_db(self):
        db.session.delete(self)
        db.session.commit()<jupyter_output><empty_output><jupyter_text>`db.session.add()` 為 upsert 作用，兼具 update 及 insert 功能 - resources 目錄下的 items.py<jupyter_code>class ItemList(Resource):

    def get(self):
        return {'items': [item.json() for item in ItemModel.query.all()]}<jupyter_output><empty_output><jupyter_text>`cls.query.all()` 選取所有元素並利用 list comprehension 轉換資料格式 list comprehension 具有 Python 風格但也可以使用其他程式語言會出現的 `map` 函式來達成：<jupyter_code>return {'items': list(map(lambda x: x.json(), ItemModel.query.all()))}<jupyter_output><empty_output><jupyter_text>註：這邊引用到的 `json()` 函式如下：<jupyter_code>def json(self):
        return {'name': self.name, 'price': self.price}<jupyter_output><empty_output><jupyter_text>轉換流程會是：1. `ItemModel.query.all()` 回傳許多 `ItemModel` 物件2. `item.json()` 將物件轉換為 dict3. `ItemList.get()` 回傳一個包含所有商品的 dict4. 由於 `ItemList` 類別繼承 `flask_restful` 的 `Resource`，此 dict 會自動轉換成 JSON ***  3-1　骨架整理 - APP<jupyter_code>from flask import Flask
from flask_restful import Api
from flask_jwt import JWT

from security import authenticate, identity
from resources.user import UserRegister
from resources.item import Item, ItemList
from resources.store import Store, StoreList

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['PROPAGATE_EXCEPTIONS'] = True
app.secret_key = 'jose'
api = Api(app)

@app.before_first_request
def create_tables():
    db.create_all()

jwt = JWT(app, authenticate, identity)  # /auth
api.add_resource(Store, '/store/<string:name>')
api.add_resource(StoreList, '/stores')
api.add_resource(Item, '/item/<string:name>')
api.add_resource(ItemList, '/items')
api.add_resource(UserRegister, '/register')

if __name__ == '__main__':
    from db import db
    db.init_app(app)
    app.run(port=5000, debug=True)<jupyter_output><empty_output><jupyter_text>3-2　骨架整理 - Model<jupyter_code>from db import db


class StoreModel(db.Model):
    __tablename__ = 'stores'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80))

    items = db.relationship('ItemModel', lazy='dynamic')

    def __init__(self, name):
        self.name = name

    def json(self):
        return {'name': self.name, 'items': [item.json() for item in self.items.all()]}

    @classmethod
    def find_by_name(cls, name):
        return cls.query.filter_by(name=name).first()

    def save_to_db(self):
        db.session.add(self)
        db.session.commit()

    def delete_from_db(self):
        db.session.delete(self)
        db.session.commit()<jupyter_output><empty_output><jupyter_text>db.py<jupyter_code>from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()<jupyter_output><empty_output><jupyter_text>3-3　骨架整理 - Resource<jupyter_code>from flask_restful import Resource
from models.store import StoreModel


class Store(Resource):
    def get(self, name):
        store = StoreModel.find_by_name(name)
        if store:
            return store.json()
        return {'message': 'Store not found'}, 404

    def post(self, name):
        if StoreModel.find_by_name(name):
            return {'message': "A store with name '{}' already exists.".format(name)}, 400

        store = StoreModel(name)
        try:
            store.save_to_db()
        except:
            return {"message": "An error occurred creating the store."}, 500

        return store.json(), 201

    def delete(self, name):
        store = StoreModel.find_by_name(name)
        if store:
            store.delete_from_db()

        return {'message': 'Store deleted'}


class StoreList(Resource):
    def get(self):
        return {'stores': list(map(lambda x: x.json(), StoreModel.query.all()))}<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1256787330316742, 'fraction_numerical': 0.007692307692307693, 'mean_word_length': 4.119281991893457, 'pattern_counts': {'":': 2, '<': 79, '<?xml version=': 0, '>': 79, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '19', 'dir': 'jupyter-structured-clean-dedup', 'id': '537312', 'license': 'Apache-2.0', 'n_tokens_mistral': 3688, 'n_tokens_neox': 3799, 'n_words': 613, 'path': 'Flask - SQLAlchemy.ipynb', 'repo_name': 'cokewolf/Python_web_notebooks', 'size': '8840'}