اتصال به دیتابیس ClickHouse در برنامه‌های Flask

pip install clickhouse-connect

CLICKHOUSE_HOST=rainier.liara.cloud
CLICKHOUSE_PORT=33273
CLICKHOUSE_DATABASE=default
CLICKHOUSE_USERNAME=root
CLICKHOUSE_PASSWORD=x4y2LJvdFyG5wVO87VbXDrpg

import os
import clickhouse_connect
from dotenv import load_dotenv

load_dotenv()

client = clickhouse_connect.get_client(
  host=os.getenv("CLICKHOUSE_HOST"),
  port=int(os.getenv("CLICKHOUSE_PORT")),
  username=os.getenv("CLICKHOUSE_USERNAME"),
  password=os.getenv("CLICKHOUSE_PASSWORD"),
  database=os.getenv("CLICKHOUSE_DATABASE"),
)

from flask import Flask, jsonify
from clickhouse import client

app = Flask(__name__)


@app.route("/clickhouse-test")
def clickhouse_test():

  try:
      result = client.query("SELECT 1")

      return jsonify({
          "success": True,
          "message": "ClickHouse connection successful",
          "result": result.result_rows
      })

  except Exception as e:

      return jsonify({
          "success": False,
          "message": "ClickHouse connection failed",
          "error": str(e)
      }), 500


if __name__ == "__main__":
  app.run(
      host="0.0.0.0",
      port=5000,
      debug=True
  )