當前位置：首頁 > 编程语言 > python >内容正文

python

[python] 基于Dataset库操作数据库

發布時間：2023/12/31 python 31 coder

生活随笔收集整理的這篇文章主要介紹了 [python] 基于Dataset库操作数据库小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

dataset庫是Python中一個用于操作數據庫的簡單庫，它提供了一種簡潔的方式與各種關系型數據庫進行交互，例如SQLite、MySQL、PostgreSQL 等。你可以使用dataset庫來執行查詢、插入、更新和刪除操作，而無需編寫復雜的SQL語句。dataset庫適用于小規模的數據存儲和查詢場景，相比csv和json文件只能通過編程語言來處理數據，dataset支持使用SQL語言進行查詢，提供了豐富的查詢功能和靈活性。對于大規模數據和高并發場景，專業數據庫系統如MySQL、PostgreSQL等更具優勢。

dataset官方倉庫地址：dataset。dataset官方文檔地址：dataset-doc。安裝命令如下：

pip install dataset

import dataset
# 查看版本
dataset.__version__

'1.6.2'

1 使用說明
- 1.1 數據庫操作
- 1.2 表格操作
- 1.3 事務操作
2 參考

1 使用說明

1.1 數據庫操作

數據庫創建

connect函數是dataset庫中用于連接到數據庫并返回一個數據庫對象。它的使用方式如下：

# 創建一個連接到名為mydatabase.db的SQLite數據庫的對象
# 如果要連接的數據庫不存在，則自動創建
db = dataset.connect('sqlite:///mydatabase.db')
# 如果要連接到MySQL數據庫（需要安裝相應組件），可以使用以下語法：
# db = dataset.connect('mysql://user:password@localhost/mydatabase')
# 如果要連接到PostgreSQL數據庫（需要安裝相應組件），可以使用以下語法：
# db = dataset.connect('postgresql://user:password@localhost/mydatabase')

表格查詢與創建

dataset提供get_table函數或直接表名索引來創建或加載一個表格。同時，dataset也提供create_table函數創建一個新的表格，也可以使用load_table函數加載現有的表格。

# 使用db['table_name']語法獲取指定表的引用，其中table_name是表的名稱
# 如果表不存在，dataset庫將自動創建它，但是只有數據插入才會保存
table = db['mytable']
# 或者使用如下函數：
# table = db.get_table('mytable')

# 創建一個新表格，原有表格會被覆蓋
# 該表沒有自定義的主鍵或類型。Dataset庫會默認創建一個名為 'id' 的整數類型的主鍵。
table = db.create_table('mytable')

此外也可以指定表格創建時的各種參數，dataset主要支持類型有：

db.types.integer：表示整數類型。
db.types.float：表示浮點數類型。
db.types.boolean：表示布爾類型。
db.types.datetime：表示日期時間類型。
db.types.date：表示日期類型。
db.types.text：表示文本字符串類型。

# 指定了自定義主鍵為 'age'
# db.create_table('mytable', 'age')
# 指定了自定義的主鍵為 'city'，并且指定該主鍵的類型為文本類型
# db.create_table('mytable', primary_id='city', primary_type=db.types.text)
# 指定沒有主鍵
# db.create_table('mytable', primary_id=False)

# 加載現有表格
table = db.load_table('mytable')
# 如果表格不存在，則會報錯
# table = db.load_table('user')

# 也可以通過has_table函數查詢是否存在某個表
db.has_table('user')

False

# 查看當前數據庫下所有表格
db.tables

[]

直接在數據庫上運行sql查詢語句

dataset庫提供query函數用于過濾和查詢數據集。它接受一個字符串參數，該參數表示一條 SQL 查詢語句，用于選擇符合條件的數據。

# statement = 'SELECT * FROM mytable'
# for row in db.query(statement):
#    print(row)

1.2 表格操作

插入數據

# 使用db['table_name']語法獲取指定表的引用，其中table_name是表的名稱
# 如果表不存在，dataset庫將自動創建它
table = db['mytable']

insert用于插入數據。

# 插入一條包含名字、年齡和電子郵件的記錄
data = {'name': 'John', 'age': 25, 'email': 'john@example.com'}
table.insert(data)
# 插入一條包含名字、年齡,電子郵件的記錄為空
table.insert(dict(name='Alice', age=25))

insert_ignore用于插入一行數據，但只有在keys中列的值不重復時才會插入。

table.insert_ignore(dict(name='Alice', age=25),keys=['name'])

False

insert_many函數用于向數據庫表中批量插入數據。它接受一個列表作為參數，該列表中的每個元素都是一個字典，表示要插入的一行數據。使用該函數要比逐個添加快很多。

data = [
    {'name': 'John', 'age': 25, 'city': 'New York'},
    {'name': 'Alice', 'age': 30, 'city': 'Los Angeles'},
    {'name': 'Bob', 'age': 35, 'city': 'San Francisco'}
]
table.insert_many(data)

插入列

# 插入名為 key 的列，數據類型為文本（text），并且設置了唯一約束（unique=True）
# table.create_column('key', db.types.text,unique=True)
# 插入名為 food 的列，數據類型為文本（text）
table.create_column('food', db.types.text)

# 創建一個列，第一個值為列名，第二個值用于自動猜測該列的類型
table.create_column_by_example('length', 4.2)

查看數據

# 查看表格行數
len(table)

# 獲取表中的所有數據
data = list(table.all())

# 打印數據
for row in data:
    print(row)

OrderedDict([('id', 1), ('name', 'John'), ('age', 25), ('email', 'john@example.com'), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 2), ('name', 'Alice'), ('age', 25), ('email', None), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 3), ('name', 'John'), ('age', 25), ('email', None), ('city', 'New York'), ('food', None), ('length', None)])
OrderedDict([('id', 4), ('name', 'Alice'), ('age', 30), ('email', None), ('city', 'Los Angeles'), ('food', None), ('length', None)])
OrderedDict([('id', 5), ('name', 'Bob'), ('age', 35), ('email', None), ('city', 'San Francisco'), ('food', None), ('length', None)])

# 直接打印數據
for row in table:
    print(row)

OrderedDict([('id', 1), ('name', 'John'), ('age', 25), ('email', 'john@example.com'), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 2), ('name', 'Alice'), ('age', 25), ('email', None), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 3), ('name', 'John'), ('age', 25), ('email', None), ('city', 'New York'), ('food', None), ('length', None)])
OrderedDict([('id', 4), ('name', 'Alice'), ('age', 30), ('email', None), ('city', 'Los Angeles'), ('food', None), ('length', None)])
OrderedDict([('id', 5), ('name', 'Bob'), ('age', 35), ('email', None), ('city', 'San Francisco'), ('food', None), ('length', None)])

# 表的列名
table.columns

['id', 'name', 'age', 'email', 'city', 'food', 'length']

# 是否存在某列
table.has_column('sex')

False

查找數據

# 根據字段查找
results = table.find(name='John')
# 遍歷結果
for row in results:
    print(row['name'], row['city'])

John None
John New York

# 多字段查找
results = table.find(name='John',city=None)
# 遍歷結果
for row in results:
    print(row['name'], row['city'])

John None

# 僅返回第一個結果
results =  table.find_one(name='John')
results

OrderedDict([('id', 1),
             ('name', 'John'),
             ('age', 25),
             ('email', 'john@example.com'),
             ('city', None),
             ('food', None),
             ('length', None)])

# 基于id查詢
results = table.find(id=[1, 2, 4])
# 遍歷結果
for row in results:
    print(row['name'], row['city'])

John None
Alice None
Alice Los Angeles

# 找出age大于等于30的結果
results = table.find(age={'>=': 30})
for row in results:
    print(row['name'], row['age'])

Alice 30
Bob 35

# 找出age在21到30之間的結果
results = table.find(age={'between': [21, 30]})
for row in results:
    print(row['name'], row['age'])

John 25
Alice 25
John 25
Alice 30

# 篩選某個字段下的唯一項
results = table.distinct('name')
for row in results:
    print(row)

OrderedDict([('name', 'Alice')])
OrderedDict([('name', 'Bob')])
OrderedDict([('name', 'John')])

dataset庫還支持更豐富的查詢操作，具體可以查看文檔：dataset-query。

更新數據

update函數用于更新單條記錄。

# 第一參數為要更新的字段和對應的新值，如果字段不存在則會添加該字段
# 第二個參數keys用于指定根據哪些列來篩選要更新的對象
# 下面代碼表示將name為'John'的行更新數據
affected_rows = table.update(dict(name= 'John',age=23,email="qq.com"), keys=['name'])
print("受影響的行數：", affected_rows)

受影響的行數： 2

# 獲取表中的所有數據
data = list(table.all())

# 打印數據
for row in data:
    print(row)

OrderedDict([('id', 1), ('name', 'John'), ('age', 23), ('email', 'qq.com'), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 2), ('name', 'Alice'), ('age', 25), ('email', None), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 3), ('name', 'John'), ('age', 23), ('email', 'qq.com'), ('city', 'New York'), ('food', None), ('length', None)])
OrderedDict([('id', 4), ('name', 'Alice'), ('age', 30), ('email', None), ('city', 'Los Angeles'), ('food', None), ('length', None)])
OrderedDict([('id', 5), ('name', 'Bob'), ('age', 35), ('email', None), ('city', 'San Francisco'), ('food', None), ('length', None)])

update_many也被提供用于批量更新數據。

# 更新多行數據
rows = [
    {'name': 'Alice', 'age': 123},
    {'name': 'Bob', 'age': 18}
]
# 該函數沒有返回值
table.update_many(rows,keys=['name'])

此外也可以用upsert和upsert_many插入更新單條或多條記錄，即如果存在具有匹配關鍵字的行，則這些行將被更新，否則將在表中插入新行。具體使用類似update和update_many。

刪除數據

delete函數用于刪除行，如下所示：

for row in table:
    print(row)

OrderedDict([('id', 1), ('name', 'John'), ('age', 23), ('email', 'qq.com'), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 2), ('name', 'Alice'), ('age', 123), ('email', None), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 3), ('name', 'John'), ('age', 23), ('email', 'qq.com'), ('city', 'New York'), ('food', None), ('length', None)])
OrderedDict([('id', 4), ('name', 'Alice'), ('age', 123), ('email', None), ('city', 'Los Angeles'), ('food', None), ('length', None)])
OrderedDict([('id', 5), ('name', 'Bob'), ('age', 18), ('email', None), ('city', 'San Francisco'), ('food', None), ('length', None)])

# 刪除age為23的記錄
table.delete(age=23)
for row in table:
    print(row)

OrderedDict([('id', 2), ('name', 'Alice'), ('age', 123), ('email', None), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 4), ('name', 'Alice'), ('age', 123), ('email', None), ('city', 'Los Angeles'), ('food', None), ('length', None)])
OrderedDict([('id', 5), ('name', 'Bob'), ('age', 18), ('email', None), ('city', 'San Francisco'), ('food', None), ('length', None)])

# 刪除age為30且name為'John'的記錄
table.delete(age=30, name='John')
for row in table:
    print(row)

OrderedDict([('id', 2), ('name', 'Alice'), ('age', 123), ('email', None), ('city', None), ('food', None), ('length', None)])
OrderedDict([('id', 4), ('name', 'Alice'), ('age', 123), ('email', None), ('city', 'Los Angeles'), ('food', None), ('length', None)])
OrderedDict([('id', 5), ('name', 'Bob'), ('age', 18), ('email', None), ('city', 'San Francisco'), ('food', None), ('length', None)])

# 找出age大于等于30的結果
table.delete(age={'>=': 30})
for row in table:
    print(row)

OrderedDict([('id', 5), ('name', 'Bob'), ('age', 18), ('email', None), ('city', 'San Francisco'), ('food', None), ('length', None)])

drop_column函數會從數據集中刪除指定的列，如下所示：

# sqlite不支持
# table.drop_column('age')
# for row in table:
#     print(row)

1.3 事務操作

事務是一組數據庫操作，要么全部成功執行，要么全部回滾。這可以確保數據的一致性和完整性。下面代碼展示了如何創建一個事務對象。在這種情況下，所有更新都會立即提交，或者，在出現異常的情況下，立即回滾所有操作。事務通過上下文管理器得到支持，因此可以通過with語句使用：

with dataset.connect('sqlite:///mydatabase.db') as tx:
    # 在這里執行數據庫操作
    # 如果所有操作都成功，事務會自動提交
    # 如果發生錯誤，事務會自動回滾
    tx['user'].insert(dict(name='John Doe', age=46, country='China'))

當然以上代碼可以顯式調用函數begin、commit和rollback來獲得相同的功能：

db = dataset.connect('sqlite:///mydatabase.db')
db.begin()
try:
    db['user'].insert(dict(name='John Doe', age=46, country='China'))
    db.commit()
except:
    db.rollback()

除此之外，也支持嵌套事務如下所示：

db = dataset.connect()
with db as tx1:
    tx1['user'].insert(dict(name='John Doe', age=46, country='China'))
    with db as tx2:
        tx2['user'].insert(dict(name='Jane Doe', age=37, country='France', gender='female'))

2 參考

dataset
dataset-doc
dataset-query

總結

以上是生活随笔為你收集整理的[python] 基于Dataset库操作数据库的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：测油仪使用方法@恒美介绍
下一篇： python异步编程之asyncio初识