（數據科學學習手札120）Python+Dash快速web應用開發——整合數據庫

本文示例代碼已上傳至我的Github倉庫https://github.com/CNFeffery/DataScienceStudyNotes

• 😋由我開源的先進Dash組件庫feffery-antd-components正處於早期測試版本階段，歡迎前往官網http://fac.feffery.tech/了解更多

1 簡介

　　 這是我的系列教程Python+Dash快速web應用開發的第十七期，在之前的各期教程中，我們針對Dash中各種基礎且常用的概念展開了學習，但一直沒有針對與數據庫之間交互進行專門的介紹，只是在某些示例中利用pandas、SQLAlchemy等工具簡陋地操作數據庫。

　　而在今天的教程中，我就將帶大家學習在Dash中利用簡單好用的ORM庫peewee，快速高效地將數據庫整合進Dash應用中。

圖1

2 利用peewee在Dash中整合數據庫

　　說起peewee，很多使用過ORM（Object Relational Mapping，對象關系映射）工具的朋友都聽說過，它跟SQLAlchemy等框架從功能上看都大同小異，目的都是為了不寫SQL，而是利用面向對象編程的方式，在Python中實現常用的SQL功能。

圖2

　　peewee雖然相比SQLAlchemy等重型的ORM框架已經輕量很多了，但內容還是非常豐富，我們今天就針對一些典型場景，展示一下其與Dash應用如何相互結合。

2.1 創建數據表

　　利用peewee構建數據表，需要定義相應的Model類，在類中構建的屬性即對應表中的字段，並且在Meta類中定義其他的一些屬性，譬如下面的例子我們就以最簡單的SQLite數據庫為例：

model1.py

from peewee import SqliteDatabase, Model
from peewee import CharField, IntegerField, DateTimeField

from datetime import datetime

# 關聯數據庫，對於sqlite數據庫若不存在則會直接創建
db = SqliteDatabase('17 整合數據庫/model1.db')

class Model1(Model):

    # 用戶名為字符型，並設置唯一性約束
    username = CharField(unique=True)

    # 用戶等級設定為整數型
    level = IntegerField()

    # 用戶加入時間為時間日期類型
    join_datetime = DateTimeField()

    class Meta:
        database = db # 指定數據庫
        table_name = 'user_info' # 自定義數據表名，不設置則自動根據類名推導

# 創建數據表，若對應數據庫中已存在此表，則會跳過
db.create_tables([Model1])

　　上述的代碼在執行之后，便會在關聯到的SQLite數據庫中創建對應的表：

圖3

　　而除了最簡單的SQLite之外，peewee還支持MySQL、PostgreSQL，你可以在http://docs.peewee-orm.com/en/latest/peewee/database.html查看更多使用示例，關於更多有關Model創建的知識可以參考http://docs.peewee-orm.com/en/latest/peewee/models.html。

2.2 向表中新增記錄

　　在數據表創建完成之后，我們第一件事當然是要向表中插入數據，這在peewee中操作非常簡單：

• 插入單條數據

　　在peewee中向表中插入單條記錄可以使用create()方法：

# 創建單條記錄
Model1.create(username='張三', level=6, join_datetime=datetime(2020, 1, 1, 10, 28, 45))

Model1.create(username='李四', level=1, join_datetime=datetime(2020, 5, 1, 10, 28, 45))

　　執行完上述命令后旋即會更新到數據庫表中：

圖4

• 插入多條數據

　　在peewee中批量插入數據可以使用insert_many()方法傳入對應每行內容的字典列表，記得最后要跟着執行execute()方法才會真正向數據庫執行：

# 批量插入數據
(
    Model1
    .insert_many([
    {'username': '王五', 'level': 3, 'join_datetime': datetime(2020, 3, 1, 10, 28, 45)},
    {'username': '趙六', 'level': 2, 'join_datetime': datetime(2020, 4, 1, 10, 28, 45)}])
    .execute()
)

圖5

2.3 從表中刪除數據

　　對於已存在數據的表，進行數據刪除可以使用到delete()方法其后再鏈式上where()來聲明判斷條件，最后同樣跟上execute()方法執行即可，如果要清空整張表則不用加where()，譬如我們要刪除level小於3的記錄：

# 刪除level小於3的記錄
Model1.delete().where(Model1.level < 3).execute()

圖6

　　更多關於peewee數據刪除的知識可以參考官方文檔http://docs.peewee-orm.com/en/latest/peewee/querying.html#deleting-records部分內容。

2.4 對表中數據進行更新

　　作為增刪改查中非常重要的改，在peewee中實現也是非常的方便，基礎的用法是配合update()與where()如下面的例子那樣：

# 修改username為張三的記錄值level字段為8
Model1.update(level=8).where(Model1.username == '張三').execute()

圖7

　　更多內容可參考官方文檔http://docs.peewee-orm.com/en/latest/peewee/querying.html#updating-existing-records。

2.5 對表中數據進行查詢

　　作為增刪改查中使用頻次最高的查，在peewee中涉及到的知識內容非常之龐大，但基礎的格式都是利用select()方法，常用的有以下方式：

# 獲取查詢結果方式1：
query_results = Model1.select().where(Model1.level > 2).execute()

for query_result in query_results:
    print(query_result.username)

圖8

# 獲取查詢結果方式2：
query_results = Model1.select().where(Model1.level > 2).dicts()
list(query_results)

圖9

　　而有關跨表連接等進階的查詢操作，請參考官方文檔http://docs.peewee-orm.com/en/latest/peewee/query_examples.html#query-examples。

2.6 基於已存在的表逆向生成Model

　　如果你的數據庫表已然存在，又希望生成相應的Model類，peewee提供了命令行工具幫我們做這件事，以SQLite為例：

python -m pwiz -e sqlite model1.db >model2.py

　　自動生成的model2.py代碼如下，在這個基礎上我們可以進一步的優化修改：

from peewee import *

database = SqliteDatabase('model1.db')


class UnknownField(object):
    def __init__(self, *_, **__): pass


class BaseModel(Model):
    class Meta:
        database = database


class UserInfo(BaseModel):
    join_datetime = DateTimeField()
    level = IntegerField()
    username = CharField(unique=True)

    class Meta:
        table_name = 'user_info'

　　而更多關於peewee利用pwiz生成Model類的參數和用法可參考官方文檔http://docs.peewee-orm.com/en/latest/peewee/playhouse.html#pwiz-a-model-generator。

3 peewee配合Dash實現在線留言板功能

　　get到peewee的常用基礎用法之后，我們回到本文的重點——結合Dash整合數據庫，要實現的功能很簡單，就是實現一個在線留言板，每個訪問應用的用戶都可以在填寫若干信息后，發表自己的留言，其他用戶后續訪問可以看到前面用戶發表過的留言信息。

　　為了方便演示，我選擇SQLite作為示例數據庫，首先我們需要構建一個model.py來設計表模型，來存放每條留言信息，並自定義一些功能函數：

model.py

from peewee import SqliteDatabase, Model
from peewee import CharField, DateTimeField, TextField
from datetime import datetime

db = SqliteDatabase('message_board.db')


class MessageBoard(Model):
    nickname = CharField()

    pub_datetime = DateTimeField()

    message_content = TextField()

    class Meta:
        database = db  # 指定數據庫
        table_name = 'message_board'  # 自定義數據表名，不設置則自動根據類名推導


db.create_tables([MessageBoard])


# 新增留言記錄
def submit_new_message(nickname, message_content):
    MessageBoard.create(
        nickname=nickname,
        pub_datetime=datetime.now(),
        message_content=message_content
    )


# 獲取全部留言記錄
def fetch_all_message():
    return list(MessageBoard.select().dicts())

　　接着我們只需要在對應Dash應用的app.py中調用model.py中的相關功能即可，效果如下（動圖錄制有些花屏，大家可以自己運行嘗試，效果更佳）：

圖10

app.py

import dash
import dash_html_components as html
import dash_bootstrap_components as dbc
from dash.dependencies import Input, Output, State

from model import MessageBoard, submit_new_message, fetch_all_message

app = dash.Dash(__name__)

app.layout = html.Div(
    dbc.Container(
        [
            html.Div(style={'height': '20px'}),
            html.H2('Dash示例留言板'),
            dbc.Container(
                id='history-message',
                style={
                    'paddingTop': '50px',
                    'width': '70%',
                    'height': '70%',
                    'overflowY': 'auto',
                    'backgroundColor': '#fafafa'
                }
            ),
            dbc.Container(
                dbc.Row(
                    [
                        dbc.Col(
                            dbc.Input(placeholder='輸入昵稱：', id='nickname', style={'width': '100%'}),
                            width=3,
                            style={
                                'padding': 0
                            }
                        ),
                        dbc.Col(
                            dbc.Input(placeholder='輸入留言內容：', id='message', style={'width': '100%'}),
                            width=7,
                            style={
                                'padding': 0
                            }
                        ),
                        dbc.Col(
                            dbc.Button('提交', id='submit', color='primary', block=True),
                            width=2,
                            style={
                                'padding': 0
                            }
                        )
                    ]
                ),
                style={
                    'paddingTop': '10px',
                    'width': '70%',
                }
            )
        ],
        style={
            'height': '800px',
            'boxShadow': 'rgb(0 0 0 / 20%) 0px 13px 30px, rgb(255 255 255 / 80%) 0px -13px 30px',
            'borderRadius': '10px'
        }
    ),
    style={
        'paddingTop': '50px'
    }
)


@app.callback(
    Output('history-message', 'children'),
    Input('submit', 'n_clicks'),
    [State('nickname', 'value'),
     State('message', 'value')]
)
def refresh_message_board(n_clicks, nickname, message):
    if nickname and message:
        submit_new_message(nickname, message)

    return [
        html.Div(
            [
                html.Strong(record['nickname']),
                html.Span('　'),
                html.Em(record['pub_datetime'].strftime(format='%Y-%m-%d %H:%M:%S')),
                html.Br(),
                html.P(record['message_content'])
            ]
        )
        for record in fetch_all_message()
    ]


if __name__ == '__main__':
    app.run_server(debug=True)

---

　　有關peewee的內容非常豐富，想要完全記住不太現實，大家可以養成多查官網http://docs.peewee-orm.com/en/latest/的習慣，內容非常詳細生動，給官方點個贊！

　　以上就是本文的全部內容，歡迎在評論區發表你的意見和想法。