Python3 Tkinter データベースにデータを書いてからCSV出力したい。

Question

まずTkinterで、ボタンを押したらデータベースにデータを保存するプログラムを作りました。

def ボタンが押されたら():
    MySQLへデータを保存

のような要領です。

しかし、Raspberry Piではデータベースの書き込みに時間がかかるようで、ボタンを押したら数秒間ボタンが押し込みっぱなしの画面になります。

それが嫌だったので、別スレッドでデータを保存することにしました。

以下、ソースコードです。

from datetime import datetime
import csv
import pymysql
import queue
import threading
import tkinter as tk

tbl = "testtb"
tmp = []

def cnct():
    global con, cur
    # データベース接続
    con = pymysql.connect(
        host = "localhost",
        user = "root",
        password = "root",
        db = "testdb",
        charset = "utf8"
    )
    cur = con.cursor()


def discnct():
    global con, cur
    # データベース切断
    cur.close()
    con.close()

"""
def read_and_csvout():
    global cur, tbl
    cur.execute("SELECT * FROM " + tbl + ";")
    results = cur.fetchall()
    lst = []
    for i in range(len(results)):
        tmp = results[i]
        lst.append([ tmp[0], tmp[1], tmp[2] ])
    with open("/home/pi/デスクトップ/data.csv", "a", newline="") as f:
        writer = csv.writer(f, lineterminator="\r\n")
        for i in range(len(lst)):
            writer.writerow(lst[i])
"""
    

def pushed1(e):
    global data
    # データベース保存用データをキューに格納
    for i in range(50):
        data.put([str(i+1), "ButtonPushed1", ""])


def pushed2(e):
    global data
    # データベース保存用データをキューに格納
    for i in range(50):
        data.put([str(i+1), "ButtonPushed2", ""])



def loop():
    global loopflag, data, cur, con, tmp
    # キュー監視ループ
    while loopflag:
        # キューにデータが存在したら、順番通りにデータベースへ保存する
        while not data.empty():
            # データベースとの接続がタイムアウトしていたときのための再接続。
            con.ping(reconnect=True)
            # キューにある先頭データを取り出し、tmpに格納
            tmp = data.get()
            # tmpにあるデータをデータベースに保存
            cur.execute("INSERT INTO " + tbl + " VALUES(%s,%s,%s);", (tmp[0], tmp[1], tmp[2]))
        # コミット。データベースに変更を加えたらそれを反映させるために必要。
        con.commit()


def close():
    global loopflag, win, thread
    # スレッド内のループフラグをFalseにする
    loopflag = False
    # スレッドが終わるまで待機
    thread.join()
    # データベースから切断
    discnct()
    # Tkinterアプリの終了
    win.destroy()


if __name__ == "__main__":

    # データベースへ接続
    cnct()

    # スレッド内のループ用フラグ
    loopflag = True

    # データベースに保存するデータ格納用
    data = queue.Queue()

    # Tkinter UI
    win = tk.Tk()

    # ボタン
    button1 = tk.Button(win, text="Button1")
    button1.grid(row=0, column=0, sticky="nsew")
    button2 = tk.Button(win, text="Button2")
    button2.grid(row=0, column=1, sticky="nsew")

    # データベース操作スレッド
    # ※データベース操作はスレッドセーフではないため、1つのスレッドでのみ操作が可能
    thread = threading.Thread(target=loop)
    thread.start()

    # ボタンバインド
    button1.bind("<ButtonRelease>", pushed1)
    button2.bind("<ButtonRelease>", pushed2)

    # UIクローズ検知
    win.protocol("WM_DELETE_WINDOW", close)

    # イベントループ
    win.mainloop()


"""
MariaDB [testdb]> DESC testtb;
+-------+------+------+-----+---------+-------+
| Field | Type | Null | Key | Default | Extra |
+-------+------+------+-----+---------+-------+
| data1 | text | YES  |     | NULL    |       |
| data2 | text | YES  |     | NULL    |       |
| data3 | text | YES  |     | NULL    |       |
+-------+------+------+-----+---------+-------+
3 rows in set (0.00 sec)
"""

ボタンを押したら保存するところまではできたのですが、
ここからさらに機能の実装をしたいと考えています。

内容として、Button2を押したらデータをデータベースに保存してからCSV出力するということをしたいです。

例えば、Button1押下→Button2押下とした場合、上のソースコードだと

1,ButtonPushed1
2,ButtonPushed1
3,ButtonPushed1
(省略)
49,ButtonPushed1
50,ButtonPushed1
1,ButtonPushed2
2,ButtonPushed2
(省略)
49,ButtonPushed2
50,ButtonPushed2

というCSVを出力したいです。

このときにButton2を押したときの処理が、メインスレッドだとバッファにキューに保存しているだけなのでMySQL操作しているスレッドの動作まで検知できません。

（グローバル変数で適当なフラグを立てることを考えたのですが、Button1のデータ書き込みが終わったのか、Button2のデータ書き込みが終わったのか、そもそも書き込んでいないのかまではloop()内で区別できないような気がします。変数だけじゃなくコールバック関数などを駆使すればできる？）

どのようにすれば 50,ButtonPushed2 をデータベースに保存した後でCSV出力をすることが可能でしょうか。

（ただし、出力データ 50,ButtonPushed2 はあくまでサンプルデータにすぎないので、実際に保存するデータはどうなるか分かりません。そのためスレッドのループ内に下記のような指定をするのはなしでお願いします。）

def loop():
    global loopflag, data, cur, con, tmp
    # キュー監視ループ
    while loopflag:
        # キューにデータが存在したら、順番通りにデータベースへ保存する
        while not data.empty():
            # データベースとの接続がタイムアウトしていたときのための再接続。
            con.ping(reconnect=True)
            # キューにある先頭データを取り出し、tmpに格納
            tmp = data.get()
            # tmpにあるデータをデータベースに保存
            cur.execute("INSERT INTO " + tbl + " VALUES(%s,%s,%s);", (tmp[0], tmp[1], tmp[2]))
            # データの内容を指定して、CSV出力（これは解決策としては×）
            if tmp == ["50", "ButtonPushed2", ""]:
                sqldata = データベース読み込み()
                CSV出力(sqldata)
                データベースクリア()
        # コミット。データベースに変更を加えたらそれを反映させるために必要。
        con.commit()

質問後追記 変更（2020/10/08 14:39）

from datetime import datetime
import csv
import pymysql
import queue
import threading
import time
import tkinter as tk

tbl = "testtb"
tmp = []

def cnct():
    # データベース接続
    con = pymysql.connect(
        host = "localhost",
        user = "root",
        password = "root",
        db = "testdb",
        charset = "utf8"
    )
    cur = con.cursor()

    return con, cur


def discnct(con, cur):
    # データベース切断
    cur.close()
    con.close()


def read_and_csvout(cur):
    global tbl
    cur.execute("SELECT * FROM " + tbl + ";")
    results = cur.fetchall()
    lst = []
    for i in range(len(results)):
        tmp = results[i]
        lst.append([ tmp[0], tmp[1], tmp[2] ])
    with open("/home/pi/デスクトップ/data.csv", "a", newline="") as f:
        writer = csv.writer(f, lineterminator="\r\n")
        for i in range(len(lst)):
            writer.writerow(lst[i])


def clear(cur):
    global tbl
    cur.execute("DELETE FROM " + tbl + ";")
    

def pushed1(e):
    global data
    # データベース保存用データをキューに格納
    for i in range(50):
        data.put(("DB-INSERT", [str(i+1), "ButtonPushed1", ""]))
    data.put(("DB-COMMIT", None))


def pushed2(e):
    global data
    # データベース保存用データをキューに格納
    for i in range(50):
        data.put(("DB-INSERT", [str(i+1), "ButtonPushed2", ""]))
    data.put(("DB-COMMIT", None))
    data.put(("CSV-OUTPUT", None))



def loop(data):
    con, cur = cnct()
    # キュー監視ループ
    while True:
        # キューにある先頭データを取り出す
        msg, args = data.get()
        print(msg, args)
        if msg == "DB-INSERT":
            # データベースとの接続がタイムアウトしていたときのための再接続。
            con.ping(reconnect=True)
            # tmpにあるデータをデータベースに保存
            cur.execute("INSERT INTO " + tbl + " VALUES(%s,%s,%s);", (args[0], args[1], args[2]))
        elif msg == "CSV-OUTPUT":
            # データベースとの接続がタイムアウトしていたときのための再接続。
            con.ping(reconnect=True)
            read_and_csvout(cur)
            clear(cur)
            con.commit()
        elif msg == "DB-COMMIT":
            con.commit()
        elif msg == "DB-DISCNCT":
            break
        else:
            break
    discnct(con, cur)


def close():
    global data, win, thread
    data.put(("DB-DISCNCT", None))
    thread.join()
    # Tkinterアプリの終了
    win.destroy()


if __name__ == "__main__":

    # データベースに保存するデータ格納用
    data = queue.Queue()

    # Tkinter UI
    win = tk.Tk()

    # ボタン
    button1 = tk.Button(win, text="Button1")
    button1.grid(row=0, column=0, sticky="nsew")
    button2 = tk.Button(win, text="Button2")
    button2.grid(row=0, column=1, sticky="nsew")

    # データベース操作スレッド
    # ※データベース操作はスレッドセーフではないため、1つのスレッドでのみ操作が可能
    thread = threading.Thread(target=loop, args=(data, ))
    thread.start()

    # ボタンバインド
    button1.bind("<ButtonRelease>", pushed1)
    button2.bind("<ButtonRelease>", pushed2)

    # UIクローズ検知
    win.protocol("WM_DELETE_WINDOW", close)

    # イベントループ
    win.mainloop()


"""
MariaDB [testdb]> DESC testtb;
+-------+------+------+-----+---------+-------+
| Field | Type | Null | Key | Default | Extra |
+-------+------+------+-----+---------+-------+
| data1 | text | YES  |     | NULL    |       |
| data2 | text | YES  |     | NULL    |       |
| data3 | text | YES  |     | NULL    |       |
+-------+------+------+-----+---------+-------+
3 rows in set (0.00 sec)
"""

Answer 1

Button2を押したらデータをデータベースに保存してからCSV出力するということをしたいです。

キューに入れるデータに、処理の振り分けフラグを追加します。

# キューの読み出し

msgtype, args = data.get()

if msgtype == "DB-INSERT":
   pass  # DBへ書き込み
elif msgtype == "CSV-OUTPUT":
   pass  # CSV 出力
else:
   break

# キュー書き込み側（細部は省略）

for _ in range(10):
    data.put(("DB-INSERT", [...]))
else:
    data.put(("CSV-OUTPUT", "test.csv"))

この例ではタプルですが、他の処理をスレッドで行いたい場合
引数の数が変わってくると思うので、メッセージを処理別に
namedtuple や dataclasses を使いクラスを定義すると良いです。

---

別アプローチでの解決策

concurrent.futures モジュールを使うと、
スレッドで処理を行った後、任意の関数を実行できます。
但し、スレッド部分のコードの大幅な変更が必要になるので、参考適度に。

---

追記: スレッド内ではグローバル変数は極力使わない方が良いです。
スレッドセーフな操作を保証できません。

現状で問題が無ければ無理に修正しなくても良いかもしれませんが、
もし終了時にエラーが出るようなことがあれば、以下を検討して見て下さい。

• データベースの初期化や後始末はスレッド内(関数loopの頭と末尾）で行う
• 双方からアクセスするキューはThread のargs引数としてスレッドで実行する関数に渡す。

　キューはスレッドセーフなので、双方のスレッドで扱っても問題ない。

• キューの読み出しループは、特定のメッセージを受けるとループを中断出来るようにする。
  (loopflag ではなく内側のループ。読み出すメッセージが無いと get で待機したまま止まってます)

Comments

[person]

> キューに入れるデータに、処理の振り分けフラグを追加します。
この方法でできました。ありがとうございます。
ちなみに、キュー書き込み側に書かれている
for _ in range(10):
data.put(("DB-INSERT", [...]))
else:
data.put(("CSV-OUTPUT", "test.csv"))

のelse:に特別な意味があるのでしょうか？
forとelseを一緒に書いているのを今まで見たことがないので･･･。

> キューの読み出しループは、特定のメッセージを受けるとループを中断出来るようにする。
else:
break
と記述されている部分でしょうか？
また、「読み出すメッセージが無い」というのは
data.put(("message", ["a", "b"]))
のようにメッセージを含むデータのみputしていても、そういった状態に陥る可能性があるのでしょうか？
（記述するうえで誤ってメッセージを書き損じたりしなければ問題ない？）

[teamikl]

> のelse:に特別な意味があるのでしょうか？

for-else は、途中で break した場合、else は実行されません。
中断処理を想定しましたが、ここではキューを使ってるので
else は不要でしたね。現状では特に意味はありません。

＞メッセージを含むデータのみputしていても、そういった状態に陥る可能性があるのでしょうか？

データがあれば大丈夫ですが、
何もない状態でアプリケーションを終了した場合
スレッド側のコードは loopflag 判別にいかず、data.get() で待機されたままになってます。

終了時の処理を以下の様に工夫して見て下さい。
（loop関数の末尾でログ出力して、ループを抜けたのを確認）

- loopflag 変更
- キューに何かデータを送る
　受け取り側は、そのデータを受け取るとループを抜け loop関数の実行を終了するようにする
- thread.join()

[teamikl]

キューでループを抜けられれば loopflag は不要になります。

後、while not data.empty() ～ con.commit() では、
何も無い時にずっと commit してることになりませんか？

con.commit()
print("COMMIT")

として、コミットが呼ばれている回数を確認して見て下さい。

対策: キューで "DB-COMMIT" メッセージを追加する。

[teamikl]

キューの読み出しループについて

# ファイル先頭グローバルに配置。
# この値は定数なのでスレッドを跨いだ参照でも問題ありません
QUEUE_QUIT = ('QUEUE-QUIT', None)

# スレッド内のキュー読み出しループ
# iter() の第二引数の値が終了条件になります。
for msgtype, args in iter(data.get, QUEUE_QUIT):
　...

# 終了時の処理
data.put(QUEUE_QUIT)
thread.join()

[person]

> データベースの初期化や後始末はスレッド内(関数loopの頭と末尾）で行う
> 双方からアクセスするキューはThread のargs引数としてスレッドで実行する関数に渡す。
> キューでループを抜けられれば loopflag は不要になります。
> キューで "DB-COMMIT" メッセージを追加する。

質問文に上記を考慮して変えたコードを載せました。ただ、エラーが出ていて、内容としては実行時に

Warning (from warnings module):
File "/home/pi/デスクトップ/test.py", line 84
global cur, con
SyntaxWarning: name 'cur' is used prior to global declaration
>>>
Warning (from warnings module):
File "/home/pi/デスクトップ/test.py", line 84
global cur, con
SyntaxWarning: name 'con' is used prior to global declaration
>>>
Warning (from warnings module):
File "/home/pi/デスクトップ/test.py", line 91
global con
SyntaxWarning: name 'con' is used prior to global declaration
>>>

と出ます。globalをdefの1行下に書いてないからだと思うのですが、1行下に書いてしまうと、データベースを作成していない（con, curを定義していない）状態でglobal con, curとしてしまうため、これはこれでエラーが出てしまいます。

cur, conはグローバル変数にしない方がいいのでしょうか？

[person]

先ほどの私のコメントですが、
data.put(("DB=CNCT", None))
をthread定義の直前に書いたら解消したようです。すみません。

[teamikl]

＞cur, conはグローバル変数にしない方がいいのでしょうか？

可能な限り、不要なglobal宣言は避けるべきです。

今回はスレッドも絡んでいて、スレッドを跨いだ変数のアクセスがあります
mainスレッドでDBの接続切断、サブスレッドでDBへクエリ発行
では、タイミング次第では切断済みのコネクションに対してのアクセスが
発生する可能性があるので、スレッド側でDBの接続切断をした方が安全になります。

- cnct, discnct は 関数loopの頭と末尾で呼び出すようにします。
- con, cur は関数の戻り値や引数にして受け渡します。

[teamikl]

loopflag が消えましたが、
while が２重になってるので、内側のbreak ではループを抜けられないです。

return でも良いのですが、
ループの後に後始末のコードを書きたい場合があるので

while not data.empty(): を省きましょう。ループを１重にできます。
キューにデータがない場合、data.get() はputがあるまで待機します。

[person]

質問文の追記のコードを更新しました。
このような感じでしょうか？
気のせいか書き込む時間が少し遅くなったような気がします。ターミナルでSELECT文を投げたときに反映されるのが修正前と比べて遅いような気がしたので。while内の分岐に時間がかかっているのでしょうか？
データの欠落はなさそうなので問題というほどではないですが。

[teamikl]

時間が掛かっているのはtime.sleep(0.1) の影響かな？
ping については解りません。

無駄にループを消化して監視するわけではないので、
time.sleep は不要です。data.get が待機してます。

ループ内にprint 文を入れる等して、
ループがどのタイミングで実行されてるか確認して見て下さい。

＞質問文の追記のコードを更新しました。
＞このような感じでしょうか？

ループを抜けた時に後始末のコードが必ず実行されるようにしたいので、

DB接続
while True:
　キュー読み出し
DB切断

もしくは、コンテキストマネージャを使い、
後始末の処理が保証されるようにします。
具体的なコードは調べてないですが、sqlite3 ではサポートされてるので
mysqlのモジュールにも同様の仕組みがあるはず。

with DB接続 as cur:
　while True:
　　キュー読み出し

任意のタイミングでDB接続したい場合は、今のコードでも良いと思いますが、
その場合、DB切断の処理が確実に呼ばれるように気を付ける必要が出てくるので、
将来的に何らかのエラーの原因になる可能性が残ってしまいます。
（現状大丈夫なら問題ありません。）

[person]

前者でやってみます。遅かったのはsleep()のせいでした。get()で高負荷がかからいようであればsleep()は消すことにします。
一応、ご指摘頂いた内容で質問文の方に修正したプログラム載せました。

[teamikl]

＞get()で高負荷がかからいようであれば

get() 引数なしで呼び出した場合は、ブロッキングモードなので
put() があるまでコードの実行はそこで待機されます。ループの負荷はありません。

以前のコードでは while not data.empty() があったため、
get() で待たずに外側のwhile loopflag: のループが回されてました。これは負荷の掛かるループです。

修正した現在のコードでは解消されてます。

----
文章で説明が難しくなってきたので、比較用にコード書きました。
リンク先READMEに説明を書いてます。

https://repl.it/@MiKLTea/TestTkQueueThread

オフトピになりますが、他に気になった点
CSV出力と、テーブル名のSQLインジェクションについても追記。

[teamikl]

自分の書き込みが矛盾してるように感じたので、補足

>>メッセージを含むデータのみputしていても、そういった状態に陥る可能性があるのでしょうか？

> 何もない状態でアプリケーションを終了した場合
> スレッド側のコードは loopflag 判別にいかず、「data.get() で待機されたまま」になってます。

これは、事前の not data.empty() を失念してました。

「他スレッドからキューを読みださなければ」「事前に not data.empty()で確認している為」
data.get() が待機したままスレッドが終われないという事はありません。ですが、
いくつかの条件次第では崩れてしまう潜在的な問題ではあるので、
対策した方が良い事には変わりません。（追記されたコードでは解消済み）

マルチスレッドにおいては、not data.empty() で確認した後に、data.get() しても
キューの中身が既に空になっている可能性があります。(スレッドとキューの使い方次第)

>「 get() で待たずに」外側のwhile loopflag: のループが回されてました。これは負荷の掛かるループです

質問が投稿された時点のコードではこちらの状況です。