INNER CODE UNIT · Python

clickhouse_query

Altinity/clickhouse-operator · tools/restore_shared_to_replicated.py:376

def clickhouse_query(
    args: argparse.Namespace,
    sql: str,
    *,
    database: str | None = None,
) -> str:
    params = {
        "user": args.clickhouse_user,
        "password": args.clickhouse_password,
        "database": database or args.clickhouse_database,
    }
    url = args.clickhouse_url.rstrip("/") + "/?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, data=sql.encode("utf-8"), method="POST")
    req.add_header("Content-Type", "text/plain; charset=utf-8")
    try:
        with urllib.request.urlopen(req) as resp:
            return resp.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as e:

View source record →