heguro/cosmosdb-cli-ja.mjs

## cosmosdb-cli-ja.mjs
// Licensed with CC0-1.0
import { promises as fs } from "fs";
import { CosmosClient } from "@azure/cosmos";

// デフォルトSQL
const DEFAULT_SELECT_SQL = "SELECT TOP 100 * FROM c ORDER BY c._ts DESC";

// 同時挿入件数 (max 100)。これ以上の件数は分割する
const BULK_UPSERT_COUNT = 20;

// Azure Cosmos DB の設定
const endpoint = "";
const key = "";
const databaseId = "";

const cosmosClient = new CosmosClient({ endpoint, key });

const dateFormatter = new Intl.DateTimeFormat("ja-JP", {
  year: "numeric",
  month: "2-digit",
  day: "2-digit",
  hour: "2-digit",
  minute: "2-digit",
  second: "2-digit",
});

/** ヘルプ表示 */
function showHelp() {
  console.log(
    `
使い方:
  # ヘルプを表示
  node db.mjs --help

  # コンテナリストを表示
  node db.mjs

  # コンテナ内のレコードをテーブルで表示
  # (デフォルトSQL: "${DEFAULT_SELECT_SQL}")
  # (table は省略可)
  node db.mjs <コンテナ名> select table
  node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" table

  # JSON で表示
  node db.mjs <コンテナ名> select json
  node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" json
  node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" json index=0,1,2

  # コンテナ内でSELECT文を実行し、結果をファイルに保存
  node db.mjs <コンテナ名> select json > <保存ファイルパス>
  node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" json > <保存ファイルパス>
  node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" json index=0,1,2 > <保存ファイルパス>

  # テーブルに表示する列を追加
  # (列名リスト省略の場合、idとパーティションキーを表示)
  node db.mjs <コンテナ名> select "<SELECT文>" table [カンマ区切りの列名リスト]

  # コンテナ内でSELECT文を実行した結果から削除
  node db.mjs <コンテナ名> select "<SELECT文>" delete all
  node db.mjs <コンテナ名> select "<SELECT文>" delete index=0,1,2

  # コンテナ内のレコードを挿入/更新 (ファイル内容は配列または単体レコード)
  node db.mjs <コンテナ名> upsert <JSONファイルパス>
`,
  );
}

/** コンテナリスト表示 */
async function showContainerList() {
  console.warn(`${dbCommandPrefix} --help でヘルプを表示\n`);

  const database = cosmosClient.database(databaseId);
  const containers = await database.containers.readAll().fetchAll();

  for (const containerDef of containers.resources) {
    console.warn("container name:", [containerDef.id]);
    console.warn("パーティションキー:", containerDef.partitionKey.paths);

    // get container items
    const container = database.container(containerDef.id);
    const { resources: itemsCount } = await container.items
      .query("SELECT VALUE COUNT(1) FROM c")
      .fetchAll();
    console.warn("item count:", itemsCount);
    console.warn("");
  }
}

/** SELECT実行 */
async function executeQuery(containerName, sql) {
  const database = cosmosClient.database(databaseId);
  const container = database.container(containerName);

  if (sql.toLowerCase().startsWith("where ")) {
    sql = `SELECT * FROM c ${sql}`;
  }
  console.warn(sql);

  const { resources } = await container.items.query(sql).fetchAll();
  return resources;
}

/** 削除実行 */
async function deleteItems(containerName, sql) {
  console.warn("削除用アイテム取得中");
  const items = await executeQuery(containerName, sql);

  if (items.length === 0) {
    console.error("対象0件");
    return;
  }

  console.warn(
    `${items.length} 件削除します。 キャンセルしたい場合、3秒以内に Ctrl+C を押してください`,
  );
  await wait(3000);

  console.warn("削除中");
  const database = cosmosClient.database(databaseId);
  const container = database.container(containerName);

  const containerInfo = await container.read();
  const partitionKeyName = containerInfo.resource.partitionKey.paths[0].replace(
    "/",
    "",
  );

  let deletedCount = 0;
  let totalRUs = 0;

  // bulk delete
  /** @type {import("@azure/cosmos").OperationInput[]} */
  const operations = items.map((item) => ({
    operationType: "Delete",
    id: item.id,
    partitionKey: item[partitionKeyName],
  }));

  for (let i = 0; i < operations.length; i += 100) {
    const chunk = operations.slice(i, i + 100);
    const results = await container.items.bulk(chunk);
    for (const result of results) {
      if (result.statusCode === 204) {
        deletedCount++;
        totalRUs += result.requestCharge;
      } else {
        console.error(`削除エラー (${result.statusCode})`);
      }
    }
  }

  console.warn(`\n削除完了:`);
  console.warn(`- 削除数: ${deletedCount}`);
  console.warn(`- RU: ${totalRUs.toFixed(2)}`);
}

/** Upsert実行 */
async function upsertItems(containerName, filePath) {
  console.warn("読込中");
  const fileContent = await fs.readFile(filePath, "utf8");
  let items;

  try {
    items = JSON.parse(fileContent);
    if (!Array.isArray(items)) {
      items = [items]; // 単体レコードの場合は配列に変換
    }
  } catch (error) {
    console.error("JSONファイル解釈失敗:", error.message);
    return;
  }

  console.warn(
    `${items.length} 件 Upsert します。 キャンセルしたい場合、3秒以内に Ctrl+C を押してください`,
  );
  await wait(3000);

  console.warn(`${items.length} 件の Upsert 開始`);

  const database = cosmosClient.database(databaseId);
  const container = database.container(containerName);

  let upsertedCount = 0;
  let totalRUs = 0;

  // バルクアップサート用の操作リストを作成
  const bulkOperations = items.map((item) => ({
    operationType: "Upsert",
    resourceBody: item,
  }));

  // BULK_UPSERT_COUNT 件ずつに分割してバルク処理
  for (let i = 0; i < bulkOperations.length; i += BULK_UPSERT_COUNT) {
    const chunk = bulkOperations.slice(i, i + BULK_UPSERT_COUNT);
    try {
      const results = await container.items.bulk(chunk);

      for (const result of results) {
        if (result.statusCode === 200 || result.statusCode === 201) {
          upsertedCount++;
          totalRUs += result.requestCharge;
        } else {
          console.error(
            `index ${i + results.indexOf(result)} の Upsert に失敗:`,
            result.statusCode,
          );
        }
      }

      console.warn(
        `${Math.min(i + BULK_UPSERT_COUNT, items.length)}/${items.length} items...`,
      );

      // CosmosDBのレート制限を考慮して少し待機
      await wait(1000);
    } catch (error) {
      console.error(
        `Failed to process chunk starting at index ${i}:`,
        error.message,
      );
    }
  }

  console.warn(`\nUpsert 完了:`);
  console.warn(`- Upsert 数: ${upsertedCount}`);
  console.warn(`- RU: ${totalRUs.toFixed(2)}`);
}

/** メイン処理 */
async function run() {
  const args = process.argv.slice(2);

  if (!args[0]) {
    await showContainerList();
    return;
  }
  if (args[0] === "--help") {
    showHelp();
    return;
  }

  const [containerName, operation, ...params] = args;

  try {
    switch (operation) {
      case "select": {
        const sqlDefined =
          params[0]?.toLowerCase().startsWith("select ") ||
          params[0]?.toLowerCase().startsWith("where ");
        if (!sqlDefined) {
          // params[0] を空文字で挿入
          params.unshift("");
        }
        const sql = params[0] || DEFAULT_SELECT_SQL;
        const selectMode = params[1] || "table";
        let results = await executeQuery(containerName, sql);

        const indexDefined = params[2]?.startsWith("index=");
        if (indexDefined) {
          const indexes = params[2].slice(6).split(",").map(Number);
          results = indexes.map((i) => results[i]);
          console.warn(`index=${indexes.join(",")}`);
        }

        if (selectMode === "delete") {
          if (params[2] !== "all" && !indexDefined) {
            throw new Error(
              "delete all または delete index=数値 を指定してください",
            );
          }
          await deleteItems(containerName, sql);
        } else if (selectMode === "table") {
          if (indexDefined) {
            throw new Error("table では、 index=数値 は使用できません");
          }
          const columns = params[2] ? params[2].split(",") : null;

          // コンテナ情報を取得してパーティションキーを特定
          const database = cosmosClient.database(databaseId);
          const container = database.container(containerName);
          const containerInfo = await container.read();
          const partitionKeyName =
            containerInfo.resource.partitionKey.paths[0].replace("/", "");

          // 表示するデータを整形
          const tableData = results.map((item) => {
            let row = {};
            if (item?._ts) {
              row["__更新日"] = dateFormatter.format(
                new Date(item._ts * 1000),
              );
            }
            if (item?.id !== undefined) {
              // 通常のデータ構成
              row.id = item.id;
              row[partitionKeyName] = item[partitionKeyName];

              // 指定された追加の列があれば追加
              if (columns) {
                columns.forEach((col) => {
                  if (col !== "id" && col !== partitionKeyName) {
                    row[col] = item[col];
                  }
                });
              }
            } else {
              // SELECT VALUE などの場合、そのまま表示
              row = item;
            }

            return row;
          });

          console.table(tableData);
        } else if (selectMode === "json") {
          console.log(
            JSON.stringify(
              results,
              // `_` から始まるキーを除外
              (key, val) =>
                key.startsWith("_") && !key.startsWith("__") ? undefined : val,
              2,
            ),
          );
        } else {
          throw new Error(`不明な操作: ${selectMode}`);
        }
        break;
      }

      case "upsert": {
        if (!params[0]) {
          throw new Error("JSONファイルパスを指定してください");
        }
        await upsertItems(containerName, params[0]);
        break;
      }

      default:
        throw new Error(`不明な操作: ${operation}`);
    }
  } catch (error) {
    console.error("Error:", error.message);
  }
}

run()
  .catch((e) => {
    console.error(e);
  })
  .finally(() => {
    cosmosClient.dispose();
    process.exit();
  });

const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
	// Licensed with CC0-1.0
	import { promises as fs } from "fs";
	import { CosmosClient } from "@azure/cosmos";

	// デフォルトSQL
	const DEFAULT_SELECT_SQL = "SELECT TOP 100 * FROM c ORDER BY c._ts DESC";

	// 同時挿入件数 (max 100)。これ以上の件数は分割する
	const BULK_UPSERT_COUNT = 20;

	// Azure Cosmos DB の設定
	const endpoint = "";
	const key = "";
	const databaseId = "";

	const cosmosClient = new CosmosClient({ endpoint, key });

	const dateFormatter = new Intl.DateTimeFormat("ja-JP", {
	year: "numeric",
	month: "2-digit",
	day: "2-digit",
	hour: "2-digit",
	minute: "2-digit",
	second: "2-digit",
	});

	/** ヘルプ表示 */
	function showHelp() {
	console.log(
	`
	使い方:
	# ヘルプを表示
	node db.mjs --help

	# コンテナリストを表示
	node db.mjs

	# コンテナ内のレコードをテーブルで表示
	# (デフォルトSQL: "${DEFAULT_SELECT_SQL}")
	# (table は省略可)
	node db.mjs <コンテナ名> select table
	node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" table

	# JSON で表示
	node db.mjs <コンテナ名> select json
	node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" json
	node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" json index=0,1,2

	# コンテナ内でSELECT文を実行し、結果をファイルに保存
	node db.mjs <コンテナ名> select json > <保存ファイルパス>
	node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" json > <保存ファイルパス>
	node db.mjs <コンテナ名> select "<SELECT文またはWHERE文>" json index=0,1,2 > <保存ファイルパス>

	# テーブルに表示する列を追加
	# (列名リスト省略の場合、idとパーティションキーを表示)
	node db.mjs <コンテナ名> select "<SELECT文>" table [カンマ区切りの列名リスト]

	# コンテナ内でSELECT文を実行した結果から削除
	node db.mjs <コンテナ名> select "<SELECT文>" delete all
	node db.mjs <コンテナ名> select "<SELECT文>" delete index=0,1,2

	# コンテナ内のレコードを挿入/更新 (ファイル内容は配列または単体レコード)
	node db.mjs <コンテナ名> upsert <JSONファイルパス>
	`,
	);
	}

	/** コンテナリスト表示 */
	async function showContainerList() {
	console.warn(`${dbCommandPrefix} --help でヘルプを表示\n`);

	const database = cosmosClient.database(databaseId);
	const containers = await database.containers.readAll().fetchAll();

	for (const containerDef of containers.resources) {
	console.warn("container name:", [containerDef.id]);
	console.warn("パーティションキー:", containerDef.partitionKey.paths);

	// get container items
	const container = database.container(containerDef.id);
	const { resources: itemsCount } = await container.items
	.query("SELECT VALUE COUNT(1) FROM c")
	.fetchAll();
	console.warn("item count:", itemsCount);
	console.warn("");
	}
	}

	/** SELECT実行 */
	async function executeQuery(containerName, sql) {
	const database = cosmosClient.database(databaseId);
	const container = database.container(containerName);

	if (sql.toLowerCase().startsWith("where ")) {
	sql = `SELECT * FROM c ${sql}`;
	}
	console.warn(sql);

	const { resources } = await container.items.query(sql).fetchAll();
	return resources;
	}

	/** 削除実行 */
	async function deleteItems(containerName, sql) {
	console.warn("削除用アイテム取得中");
	const items = await executeQuery(containerName, sql);

	if (items.length === 0) {
	console.error("対象0件");
	return;
	}

	console.warn(
	`${items.length} 件削除します。キャンセルしたい場合、3秒以内に Ctrl+C を押してください`,
	);
	await wait(3000);

	console.warn("削除中");
	const database = cosmosClient.database(databaseId);
	const container = database.container(containerName);

	const containerInfo = await container.read();
	const partitionKeyName = containerInfo.resource.partitionKey.paths[0].replace(
	"/",
	"",
	);

	let deletedCount = 0;
	let totalRUs = 0;

	// bulk delete
	/** @type {import("@azure/cosmos").OperationInput[]} */
	const operations = items.map((item) => ({
	operationType: "Delete",
	id: item.id,
	partitionKey: item[partitionKeyName],
	}));

	for (let i = 0; i < operations.length; i += 100) {
	const chunk = operations.slice(i, i + 100);
	const results = await container.items.bulk(chunk);
	for (const result of results) {
	if (result.statusCode === 204) {
	deletedCount++;
	totalRUs += result.requestCharge;
	} else {
	console.error(`削除エラー (${result.statusCode})`);
	}
	}
	}

	console.warn(`\n削除完了:`);
	console.warn(`- 削除数: ${deletedCount}`);
	console.warn(`- RU: ${totalRUs.toFixed(2)}`);
	}

	/** Upsert実行 */
	async function upsertItems(containerName, filePath) {
	console.warn("読込中");
	const fileContent = await fs.readFile(filePath, "utf8");
	let items;

	try {
	items = JSON.parse(fileContent);
	if (!Array.isArray(items)) {
	items = [items]; // 単体レコードの場合は配列に変換
	}
	} catch (error) {
	console.error("JSONファイル解釈失敗:", error.message);
	return;
	}

	console.warn(
	`${items.length} 件 Upsert します。キャンセルしたい場合、3秒以内に Ctrl+C を押してください`,
	);
	await wait(3000);

	console.warn(`${items.length} 件の Upsert 開始`);

	const database = cosmosClient.database(databaseId);
	const container = database.container(containerName);

	let upsertedCount = 0;
	let totalRUs = 0;

	// バルクアップサート用の操作リストを作成
	const bulkOperations = items.map((item) => ({
	operationType: "Upsert",
	resourceBody: item,
	}));

	// BULK_UPSERT_COUNT 件ずつに分割してバルク処理
	for (let i = 0; i < bulkOperations.length; i += BULK_UPSERT_COUNT) {
	const chunk = bulkOperations.slice(i, i + BULK_UPSERT_COUNT);
	try {
	const results = await container.items.bulk(chunk);

	for (const result of results) {
	if (result.statusCode === 200 \|\| result.statusCode === 201) {
	upsertedCount++;
	totalRUs += result.requestCharge;
	} else {
	console.error(
	`index ${i + results.indexOf(result)} の Upsert に失敗:`,
	result.statusCode,
	);
	}
	}

	console.warn(
	`${Math.min(i + BULK_UPSERT_COUNT, items.length)}/${items.length} items...`,
	);

	// CosmosDBのレート制限を考慮して少し待機
	await wait(1000);
	} catch (error) {
	console.error(
	`Failed to process chunk starting at index ${i}:`,
	error.message,
	);
	}
	}

	console.warn(`\nUpsert 完了:`);
	console.warn(`- Upsert 数: ${upsertedCount}`);
	console.warn(`- RU: ${totalRUs.toFixed(2)}`);
	}

	/** メイン処理 */
	async function run() {
	const args = process.argv.slice(2);

	if (!args[0]) {
	await showContainerList();
	return;
	}
	if (args[0] === "--help") {
	showHelp();
	return;
	}

	const [containerName, operation, ...params] = args;

	try {
	switch (operation) {
	case "select": {
	const sqlDefined =
	params[0]?.toLowerCase().startsWith("select ") \|\|
	params[0]?.toLowerCase().startsWith("where ");
	if (!sqlDefined) {
	// params[0] を空文字で挿入
	params.unshift("");
	}
	const sql = params[0] \|\| DEFAULT_SELECT_SQL;
	const selectMode = params[1] \|\| "table";
	let results = await executeQuery(containerName, sql);

	const indexDefined = params[2]?.startsWith("index=");
	if (indexDefined) {
	const indexes = params[2].slice(6).split(",").map(Number);
	results = indexes.map((i) => results[i]);
	console.warn(`index=${indexes.join(",")}`);
	}

	if (selectMode === "delete") {
	if (params[2] !== "all" && !indexDefined) {
	throw new Error(
	"delete all または delete index=数値を指定してください",
	);
	}
	await deleteItems(containerName, sql);
	} else if (selectMode === "table") {
	if (indexDefined) {
	throw new Error("table では、 index=数値は使用できません");
	}
	const columns = params[2] ? params[2].split(",") : null;

	// コンテナ情報を取得してパーティションキーを特定
	const database = cosmosClient.database(databaseId);
	const container = database.container(containerName);
	const containerInfo = await container.read();
	const partitionKeyName =
	containerInfo.resource.partitionKey.paths[0].replace("/", "");

	// 表示するデータを整形
	const tableData = results.map((item) => {
	let row = {};
	if (item?._ts) {
	row["__更新日"] = dateFormatter.format(
	new Date(item._ts * 1000),
	);
	}
	if (item?.id !== undefined) {
	// 通常のデータ構成
	row.id = item.id;
	row[partitionKeyName] = item[partitionKeyName];

	// 指定された追加の列があれば追加
	if (columns) {
	columns.forEach((col) => {
	if (col !== "id" && col !== partitionKeyName) {
	row[col] = item[col];
	}
	});
	}
	} else {
	// SELECT VALUE などの場合、そのまま表示
	row = item;
	}

	return row;
	});

	console.table(tableData);
	} else if (selectMode === "json") {
	console.log(
	JSON.stringify(
	results,
	// `_` から始まるキーを除外
	(key, val) =>
	key.startsWith("_") && !key.startsWith("__") ? undefined : val,
	2,
	),
	);
	} else {
	throw new Error(`不明な操作: ${selectMode}`);
	}
	break;
	}

	case "upsert": {
	if (!params[0]) {
	throw new Error("JSONファイルパスを指定してください");
	}
	await upsertItems(containerName, params[0]);
	break;
	}

	default:
	throw new Error(`不明な操作: ${operation}`);
	}
	} catch (error) {
	console.error("Error:", error.message);
	}
	}

	run()
	.catch((e) => {
	console.error(e);
	})
	.finally(() => {
	cosmosClient.dispose();
	process.exit();
	});

	const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
No results found