我有一個 SQLite 資料庫，我有多個函式用于多個查詢，如下所示。我的問題是關于連接到每個函式的資料庫和游標。我的代碼是否真實有效？還有另一種方法嗎？
def get_all_data(database_addr: str):
conn = sqlite3.connect(database_addr)
c = conn.cursor()
c.execute(
"""
SELECT
a.data
FROM data_table a
"""
)
df = pd.DataFrame(c.fetchall(), columns=["data"])
return df
def get_last_record(database_addr: str):
conn = sqlite3.connect(database_addr)
c = conn.cursor()
c.execute(
"""
SELECT *
FROM data_table
ORDER BY data_table.id DESC
LIMIT 1
"""
)
df = pd.DataFrame(c.fetchall(), columns=["id"])
return df
def clear_local_data(last_record: int, database_addr: str):
conn = sqlite3.connect(database_addr)
c = conn.cursor()
c.execute(
"""
DELETE
FROM data_table a
WHERE a.id<=?
""",
(last_record,),
)
conn.commit()
uj5u.com熱心網友回復：
您不應該在每次查詢時都連接到資料庫。如果您的程式是一次性的短期運行程式，您可以使用長期連接，但如果連接斷開，那么您的程式將停止作業。如果您的程式是一項長期運行的服務，請使用池化，它將為您處理斷開連接和其他問題。
conn = sqlite3.connect(database_addr)
def get_all_data(database_addr: str):
c = conn.cursor()
c.execute(
"""
SELECT
a.data
FROM data_table a
"""
)
df = pd.DataFrame(c.fetchall(), columns=["data"])
return df
def get_last_record(database_addr: str):
c = conn.cursor()
c.execute(
"""
SELECT *
FROM data_table
ORDER BY data_table.id DESC
LIMIT 1
"""
)
df = pd.DataFrame(c.fetchall(), columns=["id"])
return df
或者
import sqlalchemy.pool as pool
import sqlite3
mypool = pool.QueuePool(
lambda: sqlite3.connect(database_addr),
max_overflow=10,
pool_size=5,
)
def get_all_data(database_addr: str):
c = mypool.connect().cursor()
c.execute(
"""
SELECT
a.data
FROM data_table a
"""
)
df = pd.DataFrame(c.fetchall(), columns=["data"])
return df
def get_last_record(database_addr: str):
c = mypool.connect().cursor()
c.execute(
"""
SELECT *
FROM data_table
ORDER BY data_table.id DESC
LIMIT 1
"""
)
df = pd.DataFrame(c.fetchall(), columns=["id"])
return df
轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/490801.html
上一篇：如何洗掉包含相同欄位但具有不同時間戳的舊資料的Sqlite行？
下一篇：SQLmax是不確定的？