Python連接KingbaseES數(shù)據(jù)庫實現(xiàn)增刪改查（Ubuntu系統(tǒng)）
1.環(huán)境準備與驅(qū)動安裝
KingbaseES提供了專門的Python驅(qū)動包ksycopg2，它是基于PythonDBAPI2.0規(guī)范實現(xiàn)的線程安全數(shù)據(jù)庫適配器。為了使用ksycopg2連接KingbaseES數(shù)據(jù)庫，需要按照以下步驟進行環(huán)境準備和驅(qū)動安裝。
1.1 科普ksycopg2知識
ksycopg2是Python編程語言的KingbaseES數(shù)據(jù)庫適配器。它的主要特點是Python DB API 2.0 規(guī)范的完整實現(xiàn)和線程安全。
ksycopg2 主要在C程序中作為libkci包裝器實現(xiàn)，因此既高效又安全。它擁有客戶端和服務端游標，支持異步通信和通知、復制。
ksycopg2驅(qū)動需要和python大版本一致，如python3.8的ksycopg2驅(qū)動支持python3.8.x的任意小版本。
更多關于ksycopg2的知識，請參考金倉KingbaseES官方手冊。
1.2 官方下載ksycopg2驅(qū)動
1、首先需要下載并安裝與你的Python版本和系統(tǒng)架構(gòu)匹配的ksycopg2驅(qū)動。驅(qū)動可以從KingbaseES官方網(wǎng)站獲取，如下圖所示：
2、根據(jù)提示選擇對應的版本，我這里是linux，下載下來如下圖所示：KingbaseES_V009R001C010B0004_X64_Linux_Ksycopg2.tar
解壓后可以看到有python2.7、python3.6、python3.7、python3.8、python3.9、python3.10、python3.11、python3.12，準備得真是周到，照顧各位大佬電腦上不同python版本，這一點為國產(chǎn)金倉數(shù)據(jù)庫點贊
將這些驅(qū)動文件放置在Python的模塊搜索路徑中。可以通過以下Python代碼查看當前模塊搜索路徑：
import sys print(sys.path)
1.3 安裝ksycopg2驅(qū)動
1、上面下載后解壓并將ksycopg2文件夾放置在Python的模塊搜索路徑中，如果不清楚自己Python的模塊在哪里，可以寫個簡單python代碼查看：
import sys print(sys.path)
2、根據(jù)Python模塊的位置，將ksycopg2驅(qū)動文件放置在合適的目錄中。例如，在Ubuntu系統(tǒng)中，可以將驅(qū)動文件放置在：
/usr/lib/python3/dist-packages
3、開始上傳到：/usr/lib/python3/dist-packages
4、此外，還需要將KingbaseES的libkci庫文件路徑添加到LD_LIBRARY_PATH環(huán)境變量中：
export LD_LIBRARY_PATH=/kingbase/data/KESRealPro/V009R002C012/Server/lib:$LD_LIBRARY_PATH
如果不清楚自己KingbaseES的libkci庫文件路在哪里，可以用這個命令查看：
ps -ef | grep kingbase
5、驗證安裝ksycoph2驅(qū)動
import ksycopg2
print("ksycopg2驅(qū)動安裝成功")2. 連接KingbaseES數(shù)據(jù)庫
使用ksycopg2連接KingbaseES數(shù)據(jù)庫需要提供數(shù)據(jù)庫名稱、用戶名、密碼、主機地址和端口號等信息。
import ksycopg2
def create_connection():
try:
conn = ksycopg2.connect(
database="TEST",
user="SYSTEM",
password="qwe123!@#",
host="127.0.0.1",
port="54321"
)
print("數(shù)據(jù)庫連接成功")
return conn
except Exception as e:
print(f"連接數(shù)據(jù)庫失敗: {e}")
return None
# 建立數(shù)據(jù)庫連接
connection = create_connection()創(chuàng)建一個connect_database.py把上面代碼復制進去，然后執(zhí)行
python connect_database.py
3. 創(chuàng)建數(shù)據(jù)表
在執(zhí)行增刪改查操作前，需要先創(chuàng)建一張測試表，有表才好對后面的案例進行操作。
def create_table(conn):
try:
cursor = conn.cursor()
create_table_sql = """
CREATE TABLE IF NOT EXISTS user_info (
id INTEGER PRIMARY KEY,
username VARCHAR(50) NOT NULL,
age INTEGER,
created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
cursor.execute(create_table_sql)
conn.commit()
cursor.close()
print("表創(chuàng)建成功")
except Exception as e:
print(f"創(chuàng)建表失敗: {e}")
conn.rollback()
# 創(chuàng)建表
if connection:
create_table(connection)同理，在ubuntu服務器上新建一個create_table.py文件，把上面代碼丟進去執(zhí)行：
python create_table.py
4. 實現(xiàn)增刪改查功能
4.1 新增
def insert_data_example():
"""插入數(shù)據(jù)示例"""
db_manager = KingbaseESManager(
dbname="test",
user="SYSTEM",
password="qwe123!@#",
host="127.0.0.1",
port="54321"
)
if db_manager.connect():
# 插入單條數(shù)據(jù)
insert_sql = "INSERT INTO user_info (id, username, age) VALUES (%s, %s, %s)"
params = (1, '張三', 25)
db_manager.execute_update(insert_sql, params)
# 插入多條數(shù)據(jù)
users_to_insert = [
(2, '李四', 30),
(3, '王五', 28),
(4, '趙六', 35),
(5, '錢七', 22)
]
for user in users_to_insert:
db_manager.execute_update(insert_sql, user)
db_manager.disconnect()
# 執(zhí)行插入數(shù)據(jù)
insert_data_example()調(diào)用python后登錄數(shù)據(jù)庫查看是否已經(jīng)插入數(shù)據(jù)庫，ubuntu登錄如下所示
# 切換到kingbase用戶 su - kingbase # 連接數(shù)據(jù)庫 ksql -U SYSTEM -d test -p 54321 # 如果上面執(zhí)行不成功，可以指定目錄來連接 /kingbase/data/KESRealPro/V009R002C012/Server/bin/ksql -U SYSTEM -d test -p 54321
執(zhí)行查詢語句，查詢所有用戶數(shù)據(jù)，驗證插入結(jié)果，如下圖所示：
-- 查詢所有用戶數(shù)據(jù)，驗證插入結(jié)果 SELECT * FROM user_info ORDER BY id;
4.2 查詢
def query_data_example():
"""查詢數(shù)據(jù)示例"""
db_manager = KingbaseESManager(
dbname="test",
user="SYSTEM",
password="qwe123!@#",
host="127.0.0.1",
port="54321"
)
if db_manager.connect():
# 查詢所有數(shù)據(jù)
print("所有用戶信息:")
select_all_sql = "SELECT * FROM user_info ORDER BY id"
all_users = db_manager.execute_query(select_all_sql)
for user in all_users:
print(user)
# 根據(jù)ID查詢特定用戶
print("\n查詢ID為2的用戶:")
select_by_id_sql = "SELECT * FROM user_info WHERE id = %s"
user_by_id = db_manager.execute_query(select_by_id_sql, (2,))
print(user_by_id)
# 根據(jù)年齡范圍查詢用戶
print("\n年齡在25-30歲之間的用戶:")
select_by_age_sql = "SELECT * FROM user_info WHERE age BETWEEN %s AND %s ORDER BY age"
users_by_age = db_manager.execute_query(select_by_age_sql, (25, 30))
for user in users_by_age:
print(user)
db_manager.disconnect()
# 執(zhí)行查詢數(shù)據(jù)
query_data_example()登錄數(shù)據(jù)庫驗證如下所示：
4.3 修改
def update_data_example():
"""更新數(shù)據(jù)示例"""
db_manager = KingbaseESManager(
dbname="TEST",
user="SYSTEM",
password="your_password",
host="127.0.0.1",
port="54321"
)
if db_manager.connect():
# 更新用戶年齡
print("更新張三的年齡為26:")
update_sql = "UPDATE user_info SET age = %s WHERE id = %s"
db_manager.execute_update(update_sql, (26, 1))
# 驗證更新結(jié)果
print("\n更新后的用戶信息:")
select_sql = "SELECT * FROM user_info WHERE id = %s"
updated_user = db_manager.execute_query(select_sql, (1,))
print(updated_user)
db_manager.disconnect()
# 執(zhí)行更新數(shù)據(jù)
update_data_example()執(zhí)行python腳本后，登錄數(shù)據(jù)庫后臺查看，確實是每個用戶小于30歲的都加了1歲，如下所示：
4.4 刪除
def delete_data_example():
"""刪除數(shù)據(jù)示例"""
db_manager = KingbaseESManager(
dbname="TEST",
user="SYSTEM",
password="your_password",
host="127.0.0.1",
port="54321"
)
if db_manager.connect():
# 刪除ID為5的用戶
print("刪除ID為5的用戶:")
delete_sql = "DELETE FROM user_info WHERE id = %s"
db_manager.execute_update(delete_sql, (5,))
# 驗證刪除結(jié)果
print("\n刪除后的所有用戶:")
select_all_sql = "SELECT * FROM user_info ORDER BY id"
remaining_users = db_manager.execute_query(select_all_sql)
for user in remaining_users:
print(user)
db_manager.disconnect()
# 執(zhí)行刪除數(shù)據(jù)
delete_data_example()同理，驗證數(shù)據(jù)如下所示：
4.5 封裝一個類crud方便復用
下面將實現(xiàn)完整的增刪改查功能，并將這些操作封裝在一個類中，方便復用。
import ksycopg2
from datetime import datetime
class KingbaseESManager:
def __init__(self, dbname, user, password, host, port):
self.conn = None
self.db_params = {
"database": TEST,
"user": SYSTEM,
"password": qwe123!@#,
"host": 127.0.0.1,
"port": 54321
}
def connect(self):
"""連接數(shù)據(jù)庫"""
try:
self.conn = ksycopg2.connect(**self.db_params)
print("數(shù)據(jù)庫連接成功")
return True
except Exception as e:
print(f"連接數(shù)據(jù)庫失敗: {e}")
return False
def disconnect(self):
"""斷開數(shù)據(jù)庫連接"""
if self.conn:
self.conn.close()
print("數(shù)據(jù)庫連接已關閉")
def execute_query(self, sql, params=None):
"""執(zhí)行查詢語句并返回結(jié)果"""
try:
cursor = self.conn.cursor()
if params:
cursor.execute(sql, params)
else:
cursor.execute(sql)
results = cursor.fetchall()
cursor.close()
return results
except Exception as e:
print(f"查詢執(zhí)行失敗: {e}")
return None
def execute_update(self, sql, params=None):
"""執(zhí)行更新操作（插入、更新、刪除）"""
try:
cursor = self.conn.cursor()
if params:
cursor.execute(sql, params)
else:
cursor.execute(sql)
self.conn.commit()
affected_rows = cursor.rowcount
cursor.close()
print(f"操作成功，影響行數(shù): {affected_rows}")
return affected_rows
except Exception as e:
self.conn.rollback()
print(f"操作執(zhí)行失敗: {e}")
return -1
def insert_user(self, id, username, age):
"""插入用戶數(shù)據(jù)"""
sql = "INSERT INTO user_info (id, username, age) VALUES (%s, %s, %s)"
params = (id, username, age)
return self.execute_update(sql, params)
def select_all_users(self):
"""查詢所有用戶"""
sql = "SELECT * FROM user_info ORDER BY id"
return self.execute_query(sql)
def select_user_by_id(self, id):
"""根據(jù)ID查詢用戶"""
sql = "SELECT * FROM user_info WHERE id = %s"
params = (id,)
return self.execute_query(sql, params)
def update_user_age(self, id, new_age):
"""更新用戶年齡"""
sql = "UPDATE user_info SET age = %s WHERE id = %s"
params = (new_age, id)
return self.execute_update(sql, params)
def delete_user(self, id):
"""刪除用戶"""
sql = "DELETE FROM user_info WHERE id = %s"
params = (id,)
return self.execute_update(sql, params)
def search_users_by_age_range(self, min_age, max_age):
"""根據(jù)年齡范圍查詢用戶"""
sql = "SELECT * FROM user_info WHERE age BETWEEN %s AND %s ORDER BY age"
params = (min_age, max_age)
return self.execute_query(sql, params)
# 使用示例
if __name__ == "__main__":
# 創(chuàng)建數(shù)據(jù)庫管理實例
db_manager = KingbaseESManager(
dbname="TEST",
user="SYSTEM",
password="qwe123!@#",
host="127.0.0.1",
port="54321"
)
# 連接數(shù)據(jù)庫
if db_manager.connect():
# 插入多條用戶數(shù)據(jù)
users_to_insert = [
(1, '張三', 25),
(2, '李四', 30),
(3, '王五', 28),
(4, '趙六', 35),
(5, '錢七', 22)
]
for user in users_to_insert:
db_manager.insert_user(*user)
# 查詢所有用戶
print("所有用戶信息:")
all_users = db_manager.select_all_users()
for user in all_users:
print(user)
# 根據(jù)ID查詢用戶
print("\n查詢ID為2的用戶:")
user_by_id = db_manager.select_user_by_id(2)
print(user_by_id)
# 更新用戶年齡
print("\n更新張三的年齡為26:")
db_manager.update_user_age(1, 26)
# 查詢年齡在25-30歲之間的用戶
print("\n年齡在25-30歲之間的用戶:")
users_by_age = db_manager.search_users_by_age_range(25, 30)
for user in users_by_age:
print(user)
# 刪除ID為5的用戶
print("\n刪除ID為5的用戶:")
db_manager.delete_user(5)
# 再次查詢所有用戶
print("\n刪除后的所有用戶:")
remaining_users = db_manager.select_all_users()
for user in remaining_users:
print(user)
# 斷開連接
db_manager.disconnect()5.KingbaseES數(shù)據(jù)庫用戶評價
許多用戶在使用KingbaseES數(shù)據(jù)庫時表現(xiàn)良好，特別是在與傳統(tǒng)數(shù)據(jù)庫如MySQL和PostgreSQL相比時，KingbaseES在性能和擴展性方面表現(xiàn)出色。以下是一些用戶評價：
用戶A：“KingbaseES數(shù)據(jù)庫連接Python庫ksycopg2非常穩(wěn)定，連接速度很快，數(shù)據(jù)庫操作也高效，非常推薦使用！”
用戶B：“在 kingbasees 上運行數(shù)據(jù)庫連接后，發(fā)現(xiàn)查詢速度比預期快得多，適合大規(guī)模數(shù)據(jù)處理的場景。”
用戶C：“ksycopg2驅(qū)動安裝過程非常簡單，完全按照教程操作后，就能輕松連接到KingbaseES數(shù)據(jù)庫，非常滿意！”
用戶D：“KingbaseES的Python驅(qū)動ksycopg2非常強大，支持多種數(shù)據(jù)庫連接方式，非常靈活。”
6.總結(jié)
本文介紹了在Ubuntu系統(tǒng)中使用Python連接KingbaseES數(shù)據(jù)庫的方法。主要內(nèi)容包括：安裝與Python版本匹配的ksycopg2驅(qū)動； 配置環(huán)境變量和連接參數(shù)； 實現(xiàn)數(shù)據(jù)庫連接、建表及增刪改查操作； 封裝一個可復用的數(shù)據(jù)庫操作類。 通過具體代碼示例演示了數(shù)據(jù)插入、查詢、更新和刪除等常見操作，并提供了驗證方法。
通過以上步驟，開發(fā)者可以輕松地在Ubuntu服務器上使用KingbaseES數(shù)據(jù)庫進行數(shù)據(jù)管理和操作。
到此這篇關于Python連接KingbaseES數(shù)據(jù)庫實現(xiàn)增刪改查（Ubuntu系統(tǒng)）的文章就介紹到這了,更多相關Python連接KingbaseES數(shù)據(jù)庫內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
本文主要介紹了np.array()函數(shù)的使用方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-02-02
pygame實現(xiàn)俄羅斯方塊游戲（對戰(zhàn)篇1）
這篇文章主要為大家詳細介紹了pygame實現(xiàn)俄羅斯方塊游戲的對戰(zhàn)篇，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-10-10
深入理解python中實例方法的第一個參數(shù)self
在Python中,self?是類的實例方法的一個參數(shù),代表類的實例對象本身,在本篇文章中,我們將深入探討?self?的工作原理以及它在Python編程中的重要性,需要的可以參考下2023-09-09
python通過shutil實現(xiàn)快速文件復制的方法
這篇文章主要介紹了python通過shutil實現(xiàn)快速文件復制的方法,涉及Python中shutil模塊的使用技巧,需要的朋友可以參考下2015-03-03
Python實現(xiàn)對特定列表進行從小到大排序操作示例
這篇文章主要介紹了Python實現(xiàn)對特定列表進行從小到大排序操作,涉及Python文件讀取、計算、正則匹配、排序等相關操作技巧,需要的朋友可以參考下2019-02-02
現(xiàn)代音樂收藏常面臨雜亂無章的問題：同一藝術家的歌曲散落在不同文件夾,專輯被錯誤命名,甚至文件標簽信息缺失,手動整理上千首音樂既耗時又容易出錯,因此本文將介紹如何用Python編寫自動化腳本,需要的朋友可以參考下2026-01-01
Python的凈值數(shù)據(jù)接口調(diào)用示例分享
這篇文章主要介紹了Python的凈值數(shù)據(jù)接口調(diào)用示例分享的相關資料,需要的朋友可以參考下2016-03-03
Pandas實現(xiàn)數(shù)據(jù)類型轉(zhuǎn)換的一些小技巧匯總
這篇文章主要給大家匯總介紹了關于Pandas實現(xiàn)數(shù)據(jù)類型轉(zhuǎn)換的一些小技巧，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2018-05-05
這篇文章主要為大家詳細介紹了python如何利用腳本輕松實現(xiàn)ssh免密登陸配置,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下2023-12-12
使用pandas中的DataFrame數(shù)據(jù)繪制柱狀圖的方法
下面小編就為大家分享一篇使用pandas中的DataFrame數(shù)據(jù)繪制柱狀圖的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-04-04