本文介绍如何在 Python 开发环境连接并访问 ByteHouse 云数仓。
ByteHouse 兼容开源 MySQL Python Driver 程序： PyMySQL (非 MySQL 官方驱动，本文已在程序 1.1.1 版本下验证）

访问 ByteHouse 所需的连接信息，请参考 获取连接信息 获取。

从 PyPI 安装

可以通过如下命令，获取最新发布版本的 PyMySQL。

python3 -m pip install PyMySQL

本章节介绍通过 PyMySQL 程序连接 ByteHouse 的基本用法，您可以在程序 Github 主页 获取最新的文档和发布版本信息。

设置连接信息

可参考下面代码样例设置 ByteHouse 连接信息，具体内容可以根据 前提条件中获取的信息填写。

db_config = {
    'host': '{HOST}', # 填写HOST地址
    'port': 3306,
    'user': 'bytehouse', 
    'password': '{API_KEY}', # API_key密钥
    'database': '{DATABASE_NAME}', # 数据库名称
    'charset': 'utf8mb4',        
}

self.connection: pymysql.Connection = pymysql.connect(**db_config)

基本用法示例

import unittest
import pymysql
from mysql.connector import Error

class TestPyMySQL(unittest.TestCase):

    def setUp(self):
        db_config = {
            'host': '{HOST}', # 填写HOST地址
            'port': 3306,
            'user': 'bytehouse', 
            'password': '{API_KEY}', # API_key密钥
            'database': '{DATABASE_NAME}', # 数据库名称
            'charset': 'utf8mb4',        
        }

        self.connection: pymysql.Connection = pymysql.connect(**db_config)

    def tearDown(self):
        self.connection.close()

    def test_show_databases(self):
        try:
            cursor = self.connection.cursor()
            cursor.execute("SHOW DATABASES")
            rows = cursor.fetchall()

            if cursor.description is not None:
                columns = [i[0] for i in cursor.description]
                print(",".join(columns))

            if rows is not None:
                for row in rows:
                    print(",".join(row))

            cursor.close()
        except Error as e:
            self.fail(f"Failed to execute SHOW DATABASES: {e}")