【Python、Golang】NotionのページのプロパティをCloud SQLのPostgreSQLに記録する（第1回）

Notion APIと相性が悪い。#

Notion APIから取得できるJSONをGoで取得して構造体に反映させるのって、けっこう面倒くさいんですよね。煩雑です。そのために、「go-notion」というライブラリを使ったんですけど、「ID」カテゴリのプロパティを取得することが出来ない。（これは自分で直せばいい気もしますが・・・。）

GitHub

GitHub - dstotijn/go-notion: Go client for the Notion API.

Go client for the Notion API. Contribute to dstotijn/go-notion development by creating an account on GitHub.

github.com

しかしさらに、この「go-notion」ライブラリの更新が鈍重なんですよね。メインリポジトリの更新日が、2023-03-01と・・・、あまり活発なリポジトリではないことが窺えます。（2024-11-25に確認しました。）

まあでも、大きなリポジトリの運営って絶対大変でしょうからね・・・。そこはしょうがない・・・。

SQLと相性が悪い。#

GoでPostgreSQLを叩いていこうと思いました。しかし、DBのカラムが増えたり名前を変えたりして、DBへのクエリを変えると同時に構造体も編集するのはいちいち面倒くさいですね。なので、今回は「GORM」ライブラリでDBのテーブルと構造体のObject-Relation-Mappinngをしていきたいと思います。

GORM

GORM

The fantastic ORM library for Golang aims to be developer friendly.

gorm.io

しかし、今回はこれが上手くいきませんでした。

まず厄介なことに、GoのSQLドライバーは、DBのカラムの中に配列を入れることが出来ないらしいです。

GitHub

Question about sql error "unsupported type []string, a slice of string" · Issue #175 · DATA-DOG/go-sqlmock

Help needed Please see the example code and the error message at the bottom. I'm not sure what is causing the error. My insert function works fine. I'm using this code to insert to a clickhouse dat...

github.com

GORMライブラリでやろうとすると、DBに配列を持てません。なので、例えばカラムをテキスト型にしてCSVを入れようとすると、SELECT文でテキスト型のカラムを参照する時に構造体の配列に変換できずにエラーが発生します。つまり、構造体に配列の属性を持てない。

こうなってしまうと、DBを参照して配列に変換するための構造体まで準備しなくてはなりません。元々、リレーションと構造体の両方を治すのが面倒だからORMを使用したいというのに、これでは本末転倒です。全然開発が進まない・・・。

unsupported type []string, a slice of string

SQLAlchemyを使う。#

今回、Cloud SQLを使っていくので、公式のチュートリアルを確認すると、SQLAlchemyを使っていました。

なので、Cloud SQLにDBを作る前に、ローカルでSQLAlchemyを使ってPostgreSQLに記録する処理を作りたいと思います。

Google Cloud Documentation

Cloud SQL Python Connector を使用して接続を作成する | Cloud SQL for PostgreSQL | Google Cloud Documentation

Cloud SQL Python Connector を使用して Cloud SQL for PostgreSQL への接続を開きます。

docs.cloud.google.com

SQLAlchemyでInsert文を作る。#

SQLAlchemyでPostgreSQLのDBにデータを格納していきます。

今回は例として、Create文とInsert文だけ。

1
from sqlalchemy import create_engine, text
2

3
# PostgreSQL接続情報
4
DATABASE_URL = "postgresql+psycopg2://username:password@localhost:5432/dbname"
5

6
# SQLAlchemyエンジンのセットアップ
7
engine = create_engine(DATABASE_URL)
8

9
# テーブル作成クエリ
10
create_table_query = """
11
CREATE TABLE IF NOT EXISTS records (
12
    id SERIAL PRIMARY KEY,
13
    name VARCHAR(255) NOT NULL,
14
    number INTEGER NOT NULL,
15
    is_active BOOLEAN NOT NULL,
16
    account_number INTEGER NOT NULL,
17
    is_verified BOOLEAN NOT NULL,
18
    score VARCHAR(50) NOT NULL,
19
    url TEXT,
20
    extra_info TEXT
21
);
22
"""
23

24
# データ挿入クエリ
25
insert_query = """
26
INSERT INTO records (name, number, is_active, account_number, is_verified, score, url, extra_info)
27
VALUES (:name, :number, :is_active, :account_number, :is_verified, :score, :url, :extra_info);
28
"""
29

30
# データ（複数レコード）
31
record = ["tanaka", 10, True, 67890, False, "5", "https://www.google.co.jp/", None]
32

33
# 実行
34
with engine.connect() as connection:
35
    # テーブル作成
36
    connection.execute(text(create_table_query))
37
    print("Table created or already exists.")
38

39
    # データを挿入
40
    connection.execute(text(insert_query), record)
41

42
    print("Multiple records inserted successfully!")

SQLAlchemyで一気にInsertできるクエリを作る。#

SQLAlchemyを使う際のトピックとしては、Insert文の作り方ですね。

Insertする際に、DBに1文ずつ渡すのでは時間が掛かり過ぎです。なので、SQLAlchemyのInsert実行のメソッドは1度だけ実行するようにしたい・・・。しかし、SQLAlchemyでググった際に、SQLAlchemyのメソッド内でInsertを一気に実行している記事が見られました。

まあそれでもいいんですけど・・・、でも個人的には、Insert実行のメソッドにはクエリを渡すようにしたいんですよね・・・。DBを操作するのに、自分が書いていない関数でやるのって、なんかスキルが身に付いている感じがしない・・・。だからクエリを渡す方法でやりたい・・・。

そこでまたSQLAlchemyでググったのですが・・・、Insert文をクエリの中に1つしか書いていない情報しか見つけられませんでした。なのでここに動いたものを書いておきます。

1
from sqlalchemy import create_engine, text
2

3
# PostgreSQL接続情報
4
DATABASE_URL = "postgresql+psycopg2://username:password@localhost:5432/dbname"
5

6
# SQLAlchemyエンジンのセットアップ
7
engine = create_engine(DATABASE_URL)
8

9
# テーブル作成クエリ
10
create_table_query = """
11
CREATE TABLE IF NOT EXISTS records (
12
    id SERIAL PRIMARY KEY,
13
    name VARCHAR(255) NOT NULL,
14
    number INTEGER NOT NULL,
15
    is_active BOOLEAN NOT NULL,
16
    account_number INTEGER NOT NULL,
17
    is_verified BOOLEAN NOT NULL,
18
    score VARCHAR(50) NOT NULL,
19
    url TEXT,
20
    extra_info TEXT
21
);
22
"""
23

24
# データ挿入クエリ
25
insert_query = """
26
INSERT INTO records (name, number, is_active, account_number, is_verified, score, url, extra_info)
27
VALUES (:name, :number, :is_active, :account_number, :is_verified, :score, :url, :extra_info);
28
"""
29

30
# データ（複数レコード）
31
records = [
32
    ["tanaka", 10, True, 67890, False, "5", "https://www.google.co.jp/", None],
33
    ["suzuki", 25, False, 12345, True, "8", "https://www.example.com/", "extra data"],
34
    ["yamada", 30, True, 67891, False, "3", "https://www.test.com/", None],
35
]
36

37
# 実行
38
with engine.connect() as connection:
39
    # テーブル作成
40
    connection.execute(text(create_table_query))
41
    print("Table created or already exists.")
42

43
    # 複数レコードのデータを挿入
44
    connection.execute(
45
        text(insert_query),
46
        [
47
            {
48
                "name": record[0],
49
                "number": record[1],
50
                "is_active": record[2],
51
                "account_number": record[3],
52
                "is_verified": record[4],
53
                "score": record[5],
54
                "url": record[6],
55
                "extra_info": record[7],
56
            }
57
            for record in records
58
        ]
59
    )
60
    print("Multiple records inserted successfully!")

上記の処理はChatGPTに出してもらったんですけど、（だからもう少し頑張ってググれば記事はあるのだろうか。）

僕はこの処理をさらに関数化して、その関数にはクエリとレコードを渡すようにして、正規表現でクエリ内に書いてあるカラム名を抽出して、リスト内包表記で辞書を作成する処理にしました。

1
from pprint import pprint
2
from typing import TypedDict, Final, List
3
import re
4
import sqlalchemy
5
import inspect
6

7
def execute_db_ops(engine: sqlalchemy.engine.base.Engine, query: str, records: List[List[any]] = None):
8
  """
9

10
  """
11
  def extract_columns(query):
12
    match = re.search(r"\((.*?)\)", query.text)
13
    if match:
14
        columns = match.group(1).split(",")
15
        return [column.strip() for column in columns] # remove spaces
16
    return []
17

18
  func_name = inspect.currentframe().f_code.co_name
19
  pprint(f"{func_name}: start processing into PostgreSQL...")
20
  with engine.begin() as conn:
21
    if records:
22
      pprint(f"{func_name}: processing with records into PostgreSQL...")
23
      columns = extract_columns(query)
24
      columns = [c.replace(":", "") for c in columns]
25
      pprint(f"{func_name}: columns:")
26
      pprint(columns)
27
      mapped_records = [{column: record[i] for i, column in enumerate(columns)} for record in records]
28
      pprint(f"{func_name}: mapped_records:")
29
      pprint(mapped_records)
30
      conn.execute(query, mapped_records)
31
      pprint(f"{func_name}: completed processing with records into PostgreSQL.")
32
    else:
33
      conn.execute(query)
34
  pprint(f"{func_name}: completed processing into PostgreSQL.")

また、ここでもう一つ気を付けたのが、このSQLAlchemy実行用の関数の引数に辞書型の変数を渡さないようにしたことです。

今後このツールを改造して、このPythonのソースコードをコンパイル出来るようになったら、スタックメモリを使える処理が出てくるかもしれません。しかし、辞書型を渡す処理は間違いなくヒープメモリを使うことになります。そして、ヒープメモリを使う方がメモリ使用量が増えるんですよね多分。あと処理も遅くなります。

あと、今回使うCloud Functionsでは、最大メモリ使用サイズによって費用が変わるので、なるべくスタックメモリを使って余計なリソースを使わないで節約したいです。もしかしたらCloud Functions以外のサービスでデプロイする時にもコストを抑えられるかもしれません。

日本語ロケールの設定#

DBに日本語の情報を使う際に、利用するサーバに日本語ロケールを設定します。こうしないとDBで日本語を使えなかったりします。

1
sudo apt-get install language-pack-ja
2
sudo dpkg-reconfigure locales

そしたら、PostgreSQLを起動して、コンソールに入ります。

1
sudo /sbin/service postgresql start
2
sudo su - postgres

SQLコンソールに入る。

1
psql -d postgres

ユーザーとDBを作ります。DBのロケールを日本語に設定します。

1
CREATE USER myuser WITH PASSWORD 'mypassword';
2
CREATE DATABASE testdb OWNER myuser ENCODING 'UTF8' lc_collate 'ja_JP.UTF-8' lc_ctype 'ja_JP.UTF-8' template 'template0';
3
GRANT ALL PRIVILEGES ON DATABASE testdb TO myuser; GRANT USAGE, CREATE ON SCHEMA PUBLIC TO myuser;

PostgreSQLへの格納が完了。#

とりあえずはPostgreSQLへの格納が終わったので、次にこのDBをCloud SQL上に作っていきたいと思います。

Cloud SQL インスタンスを作成する#

Google Cloud コンソールで、Cloud SQL の「インスタンス」 ページに移動します。

「インスタンスを作成」をクリックしてインスタンス作成画面に移動しようとすると「Compute Engine API」を有効にしておく必要があるので有効にします。有効にすると、インスタンスの作成画面が表示されます。 Cloud SQLのエディションには「Enterprise plus」と「Enterprise」の2種類あるらしく、「Enterprise」の方が安くなりそうです。今回は、「Enterprise」にしておきます。

エディションのプリセットは「サンドボックス」にします。そしてインスタンスの情報は以下の通りにしておきます。

• データベースのバージョン：PostgreSQL 16
• インスタンス-ID：testdb-01
• パスワード：（任意の文字列）

あと、「リージョンとゾーンの可用性の選択」です。 「サンドボックス」エディションでは、「シングルゾーン」しか選べないようなのですが、ここではなんかゴニョゴニョ編集できましたね・・・。「複数のゾーン」にして作成すると引っかかるのか？

「マシンの構成」やら「ストレージ」やら、色々と設定できるみたいですが、今回はデフォルト設定にしておきます。

「接続」では、デフォルト設定の「パブリックIP」を選択しておきます。新しいネットワークには今回特に入力しません。

そしたら、インスタンスを作成します。作成が始まると、インスタンスの概要を閲覧できる画面が表示されます。

Cloud SQL Auth Proxy クライアントをインストールする。#

インスタンスを作成したら、そのインスタンスに認証できるツール「Cloud SQL Auth Proxy」のクライアントをインストールします。 まずは、「APIとサービス」より「APIとサービスを有効にする」をクリックして、「Cloud SQL Admin API」を有効にします。

先程Pythonで開発したツールが置いてある場所に、Cloud SQL Auth Proxy クライアントを設置します。

OSがDebianもしくはUbuntuであれば、以下のコマンドでダウンロードします。

1
curl -o cloud-sql-proxy https://storage.googleapis.com/cloud-sql-connectors/cloud-sql-proxy/v2.13.0/cloud-sql-proxy.linux.amd64

Cloud SQL Auth Proxy を実行可能にします。

1
chmod +x cloud-sql-proxy

Cloud SQL Auth Proxyで認証する。#

それでは次に、Cloud SQL Auth Proxyで認証していきます。

まずは、Google Cloud コンソール内でDBインスタンスの画面から、「このインスタンスと接続」セクションで「接続名」をコピーします。接続名の形式は projectID:region:instanceID です。

OSがDebianもしくはUbuntuであれば、以下のコマンドでログインします。

1
./cloud-sql-proxy projectID:region:instanceID

ログインしようとすると、認証の作業が入ります。僕の場合はCloud Shell上で作業していたのでこんな感じになりました。

こんな感じのメッセージが表示されると、Cloud SQLが外部からのリクエストをListenしている状態になります。

1
2024/11/29 02:19:18 The proxy has started successfully and is ready for new connections!

そしたら今度は、Cloud SQL上のPostgreSQLに接続します。 その接続をする際に、必ずパスワードの入力が求められるので、Cloud SQLのコンソール上でサイドバーから「ユーザー」を選択して、表示された画面内でスーパーユーザー「postgres」のパスワードを設定しておきます。

設定したら、その「postgres」で先程作ったPostgreSQLインスタンス内にデフォルトで作成される「postgres」DBにログインします。

1
psql "host=127.0.0.1 port=5432 sslmode=disable dbname=postgres user=postgres"

ちなみに、Cloud SQLのコンソール上でサイドバーにある「データベース」をクリックすると、現在存在するデータベースを確認できます。

ユーザー、ロールおよびデータベースを作成する。#

DBに接続できたら、まずはユーザーを作成します。

1
CREATE USER myuser WITH PASSWORD 'mypassword';

このSQLコマンドを実行して、現在のユーザーのロールと権限を確認できます。

1
SELECT * FROM pg_roles WHERE rolname = 'myuser';

接続中のユーザーがsuperuser相当の状態で、myuserにロールを付与します。 次のようなSQLコマンドを実行して、ロールを付与できます。

1
GRANT myuser TO current_user;

そして、テーブルを作成します。日本語ロケールで作ります。

1
CREATE DATABASE lmakdb OWNER myuser ENCODING 'UTF8' lc_collate 'ja_JP.UTF-8' lc_ctype 'ja_JP.UTF-8' template 'template0';

DBを使いやすくします。

1
GRANT ALL PRIVILEGES ON DATABASE testdb TO myuser; GRANT USAGE, CREATE ON SCHEMA PUBLIC TO myuser;

これにて、データベースの作成は完了です。 先程、GRANT myuser TO current_user;を実行した理由は、実行しないでテーブルを作成しようとした時に以下のエラーが発生したためです。

1
ERROR:  must be member of role "test"

今回作成したユーザーである「myuser」でそのエラーを回避するためには、どうやら「myuser」をsuperuser相当のユーザーからロールを付与されたユーザーにする必要があるみたいです。そうしないとテーブルを作成できません。 これはPostgreSQLの仕様のようです。

Qiita

RDS (PostgreSQL) でデータベースを作成するときの注意点 - Qiita

RDS(PostgreSQL)でデータベース作成をしようとしてハマったのでメモ。 新規ユーザが所有者のデータベースを作りたい マスタユーザでログインして、 ユーザ（ロール）作成 上で作成したロールがOWNERなデータベース作成 をしようとすると、以下のようなエラーが出...

qiita.com

そして、既存のテーブルおよび作成されたテーブルを「\l」で確認してみると、

1
     Name      |       Owner       | Encoding | Locale Provider |   Collate   |    Ctype    | ICU Locale | ICU Rules |            Access privileges
2
---------------+-------------------+----------+-----------------+-------------+-------------+------------+-----------+-----------------------------------------
3
 cloudsqladmin | cloudsqladmin     | UTF8     | libc            | en_US.UTF8  | en_US.UTF8  |            |           |
4
 testdb        | myuser            | UTF8     | libc            | ja_JP.UTF-8 | ja_JP.UTF-8 |            |           | =Tc/myuser                             +
5
               |                   |          |                 |             |             |            |           | myuser=CTc/myuser
6
 postgres      | cloudsqlsuperuser | UTF8     | libc            | en_US.UTF8  | en_US.UTF8  |            |           |
7
 template0     | cloudsqladmin     | UTF8     | libc            | en_US.UTF8  | en_US.UTF8  |            |           | =c/cloudsqladmin                       +
8
               |                   |          |                 |             |             |            |           | cloudsqladmin=CTc/cloudsqladmin
9
 template1     | cloudsqlsuperuser | UTF8     | libc            | en_US.UTF8  | en_US.UTF8  |            |           | =c/cloudsqlsuperuser                   +
10
               |                   |          |                 |             |             |            |           | cloudsqlsuperuser=CTc/cloudsqlsuperuser
11
(5 rows)

ちなみに、先程作成したユーザーとロールは、Cloud SQLへのセッションを閉じても、Cloud SQL上のインスタンスを停止しても消えません。

また、hostをPythonでホストに設定していた部分を「localhost」からデータベースが置いてあるIPアドレスを指定すれば、これまでと同じようにテーブルの作成、レコードの追加などを行うことが出来ます。

Query InsightsでDBの状況を確認する。#

テーブルを作成できたら、DBホストのアドレスを変えること以外は、先程ローカルで作成したPostgreSQLデータベースに対する処理と何も変わらないので、とりあえずNotionの情報をテーブルに格納しておきます。

そして、Cloud SQLのコンソール上でサイドバーにある「Query Insights」をクリックして、Query Insightsを有効にします。

Query Insightsを有効にすると、このようなダッシュボードが表示されました。

認証に失敗するとエラー扱いになるようなので、赤いチャートが目立っています。

クエリのレイテンシー、CPU使用率を確認できて、ディスクストレージがあとどれぐらいで満杯になるかどうかを確認することも出来ます。

その他にも色々な指標を確認することが出来ます。個人的には、もう少し大量のデータを参照して、「オペレーション別の処理行数」がどのように推移していくのかを把握しておきたいですね。

Cloud SQL Studioでテーブルを参照する。#

それでは次は、Cloud SQL Studioでテーブルの中身を見ていきます。

Cloud SQLのコンソール上でサイドバーにあるCloud SQL Studioをクリックして開きます。開くとログイン画面になるのでログインして、データベースの中身を確認できるようになります。

列やシーケンスを確認できます。

そして、SQLを打ち込めるコンソールもあるので、SELECT文を打てば先程テーブルに格納したレコードを確認することが出来ました。

DBインスタンスを片付ける。#

今回、お試しに作っただけなので、課金されないように片付けたいと思います。

インスタンスが不要になったら削除したいところですが、削除する前にインスタンスの削除保護を解除する必要があります。

Google Cloud Documentation

インスタンスの削除を防止する | Cloud SQL for MySQL | Google Cloud Documentation

docs.cloud.google.com

このインスタンスの削除保護の解除は、インスタンスが起動している間にしか行うことが出来ません。なので、停止する前に設定します。 片付けたいDBインスタンスを表示して編集をクリックします。

インスタンスのカスタマイズセクションにて、データの保護の項目で、削除からの保護の有効化のチェックボックスをオフにします。

その状態で保存すると、設定したインスタンスを削除できるようになっています。そしたらインスタンスを削除します。

削除できました。削除するのに1分ほど掛かりました。