Lindorm：Menggunakan DBUtils untuk mengembangkan aplikasi

Persiapan

Jalankan kode berikut untuk menginstal phoenixdb V1.2.0 dan DBUtils V3.0.2:

pip install phoenixdb==1.2.0
pip install DBUtils==3.0.2

Kode contoh

#!/usr/bin/python3

from dbutils.pooled_db import PooledDB
import importlib


class DBUtilsDemo:

    def __init__(self, url, user, password, database):
        config = {
            'url': url,
            'lindorm_user': user,
            'lindorm_password': password,
            'database': database,
            'autocommit': True
        }
        db_creator = importlib.import_module("phoenixdb")
        # Buat kumpulan koneksi berdasarkan DBUtils.
        self.pooled = PooledDB(db_creator,
                               maxcached=10,
                               # Tentukan jumlah maksimum koneksi idle dalam kumpulan koneksi berdasarkan kebutuhan bisnis.
                               maxconnections=50,
                               # Tentukan jumlah maksimum koneksi dalam kumpulan koneksi berdasarkan kebutuhan bisnis.
                               blocking=True,
                               # Tentukan apakah klien menunggu jika tidak ada koneksi idle yang tersedia dalam kumpulan koneksi. True menunjukkan bahwa klien menunggu hingga koneksi tersedia. False menunjukkan bahwa klien tidak menunggu koneksi idle. 
                               ping=1,
                               # Periksa apakah server dapat diakses.
                               **config)

    # Dapatkan koneksi dari kumpulan koneksi.
    def _connect(self):
        try:
            r = self.pooled.connection()
            return r
        except Exception as e:
            print("Gagal terhubung:" + str(e))

    # Kembalikan koneksi ke kumpulan koneksi.
    def _close(self, conn, stmt):
        if stmt:
            stmt.close()
        if conn:
            conn.close()

    # Query satu baris data.
    def select_row(self, sql):
        connection = self._connect()
        statement = None
        try:
            statement = connection.cursor()
            statement.execute(sql)
            row = statement.fetchone()
            return row
        except Exception as e:
            print(e)
        finally:
            self._close(connection, statement)

    # Query beberapa baris data.
    def select_rows(self, sql):
        connection = self._connect()
        statement = None
        try:
            statement = connection.cursor()
            print(sql)
            statement.execute(sql)
            rows = statement.fetchall()
            return rows
        except Exception as e:
            print(e)
        finally:
            self._close(connection, statement)

    # Perbarui dan sisipkan data ke tabel.
    def upsert_data(self, sql_upsert):
        connection = self._connect()
        statement = None
        try:
            statement = connection.cursor()
            statement.execute(sql_upsert)
            connection.commit()
        except Exception as e:
            print(e)
        finally:
            self._close(connection, statement)

    # Perbarui dan sisipkan data ke tabel dengan menentukan parameter.
    def upsert_data_prams(self, sql_upsert, prams):
        connection = self._connect()
        statement = None
        try:
            statement = connection.cursor()
            statement.execute(sql_upsert, prams)
            connection.commit()
        except Exception as e:
            print(e)
        finally:
            self._close(connection, statement)

if __name__ == '__main__':
    # Tentukan titik akhir yang digunakan untuk terhubung ke LindormTable menggunakan SQL. 
    url = 'http://ld-bp1p7e07ohamf****-proxy-lindorm-pub.lindorm.rds.aliyuncs.com:30060'
    # Tentukan nama pengguna aktual yang digunakan untuk terhubung ke LindormTable. Anda dapat melihat nama pengguna dan kata sandi yang digunakan untuk terhubung ke LindormTable di sistem manajemen kluster LindormTable. 
    user = 'root'
    # Tentukan kata sandi aktual yang digunakan untuk terhubung ke LindormTable.
    password = 'root'
    # Tentukan nama database yang ingin Anda hubungkan.
    database = 'test'

    poolUtils = DBUtilsDemo(url, user, password, database)
    poolUtils.upsert_data("upsert into tb(id,name,address) values ('i001','n001','a001')")

    params = ['i002', 'n002', 'a002']
    poolUtils.upsert_data_prams("upsert into tb(id,name,address) values (?,?,?)", params)

    rows = poolUtils.select_rows("select * from tb")
    print(rows)

    row = poolUtils.select_row("select * from tb limit 1")
    print(row)

    row = poolUtils.select_row("select * from tb where id = 'i001' limit 1")
    print(row)