import sys
print(sys.path)

import sys
print(sys.path)

export LD_LIBRARY_PATH=/kingbase/data/KESRealPro/V009R002C012/Server/lib:$LD_LIBRARY_PATH

ps -ef | grep kingbase

import ksycopg2
print("ksycopg2驅(qū)動(dòng)安裝成功")

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ù)庫(kù)連接成功")
        return conn
    except Exception as e:
        print(f"連接數(shù)據(jù)庫(kù)失敗: {e}")
        return None

# 建立數(shù)據(jù)庫(kù)連接
connection = create_connection()

python connect_database.py

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)

python create_table.py

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()

# 切換到kingbase用戶
su - kingbase
 
# 連接數(shù)據(jù)庫(kù)
ksql -U SYSTEM -d test -p 54321

# 如果上面執(zhí)行不成功，可以指定目錄來(lái)連接
/kingbase/data/KESRealPro/V009R002C012/Server/bin/ksql -U SYSTEM -d test -p 54321

-- 查詢所有用戶數(shù)據(jù)，驗(yàn)證插入結(jié)果
SELECT * FROM user_info ORDER BY id;

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()

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))
        
        # 驗(yàn)證更新結(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()

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,))
        
        # 驗(yàn)證刪除結(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()

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ù)庫(kù)"""
        try:
            self.conn = ksycopg2.connect(**self.db_params)
            print("數(shù)據(jù)庫(kù)連接成功")
            return True
        except Exception as e:
            print(f"連接數(shù)據(jù)庫(kù)失敗: {e}")
            return False
    
    def disconnect(self):
        """斷開數(shù)據(jù)庫(kù)連接"""
        if self.conn:
            self.conn.close()
            print("數(shù)據(jù)庫(kù)連接已關(guān)閉")
    
    def execute_query(self, sql, params=None):
        """執(zhí)行查詢語(yǔ)句并返回結(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ù)庫(kù)管理實(shí)例
    db_manager = KingbaseESManager(
        dbname="TEST",
        user="SYSTEM",
        password="qwe123!@#",
        host="127.0.0.1",
        port="54321"
    )
    
    # 連接數(shù)據(jù)庫(kù)
    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()