■Cloud SQL Python Connector (Cloud SQL language Connector)
CloudSQL auth proxyのバイナリインストールでないやり方
Cloud SQL Python Connector自体は暗号化しないが、内部IPならサーバレスVPCコネクタで暗号化された通信が使え安全になっている。外部IPアドレスの場合はCloud SQL Auth Proxyで通信を暗号化。
Cloud SQL Python Connector自体は暗号化しないが、内部IPならサーバレスVPCコネクタで暗号化された通信が使え安全になっている。外部IPアドレスの場合はCloud SQL Auth Proxyで通信を暗号化。
Cloud SQL 言語コネクタの概要 | Cloud SQL for MySQL | Google Cloud
GitHub - GoogleCloudPlatform/cloud-sql-python-connector: A Python library for connecting securely to your Cloud SQL instances.
GitHub - GoogleCloudPlatform/cloud-sql-python-connector: A Python library for connecting securely to your Cloud SQL instances.
事前必要(pip install>requirements.txt)
Flask==3.0.3
gunicorn==22.0.0
Werkzeug==3.0.3
google-cloud-bigquery==3.25.0
google-cloud-logging==3.11.1
google-cloud-secret-manager==2.20.2
google-api-python-client==2.141.0
google-auth-httplib2==0.2.0
google-auth-oauthlib==1.2.1
websocket-client==1.8.0
google-cloud-resource-manager==1.12.5
Flask-WTF==1.2.1
cloud-sql-python-connector==1.16.0
pymysql==1.0.3
from flask import Flask, jsonify
from google.cloud.sql.connector import Connector
from google.cloud import secretmanager
import pymysql
# 環境変数の定義
PW_NAME = "sql-pw"
PROJECT_NUM = "1234567890"
DB_INSTANCE = "prj:asia-northeast1:db_instance"
DB_USER = "db-user"
DB_NAME = "db001"
# Secret Manager からパスワードを取得する関数
def get_pw(pw_name, project_num):
client = secretmanager.SecretManagerServiceClient()
resource_name = f"projects/{project_num}/secrets/{pw_name}/versions/latest"
res = client.access_secret_version(name=resource_name)
credentials = res.payload.data.decode("utf-8")
return credentials
# Cloud SQL接続
def sql_getconn(connector):
pw = get_pw(PW_NAME, PROJECT_NUM)
conn = connector.connect(
DB_INSTANCE,
"pymysql",
user=DB_USER,
password=pw,
db=DB_NAME,
ip_type="private",
)
return conn
app = Flask(__name__)
@app.route('/test', methods=['GET'])
def get_table_data():
try:
connector = Connector()
conn = sql_getconn(connector)
cursor = conn.cursor()
# SQLを実行して結果を取得
cursor.execute("SELECT no, name, targetDate FROM test")
rows = cursor.fetchall()
# 結果をJSON形式に変換
result = [
{
"no": row[0],
"name": row[1],
"targetDate": row[2].strftime("%Y-%m-%d %H:%M:%S") if row[2] else None
}
for row in rows
]
cursor.close()
conn.close()
return jsonify(result), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080)
=============
# 追加オプションを使った接続も可
connector = Connector(
ip_type="public", # "private" または "psc" も使用可能
enable_iam_auth=False,
timeout=30,
credentials=None, # 必要ならGoogle認証情報を渡す
refresh_strategy="lazy", # "lazy" または "background"
)
#トランザクション
#カーソル
#トランザクション
try:
conn = sql_getconn(connector)
conn.autocommit = False # トランザクション開始、あるいは conn.begin()
cursor = conn.cursor()
# 挿入するデータを準備
new_data = [
{"no": 4, "name": "新しい名前4", "targetDate": "2024-05-01"},
{"no": 5, "name": "新しい名前5", "targetDate": "2024-05-02"},
]
# INSERT文を構築して実行
for data in new_data:
sql = "INSERT INTO test (no, name, targetDate) VALUES (%s, %s, %s)"
values = (data["no"], data["name"], data["targetDate"])
cursor.execute(sql, values)
conn.commit() # トランザクションをコミット
print("Data inserted successfully.")
except Exception as e:
conn.rollback() # エラーが発生した場合はロールバック
print(f"Transaction rolled back due to an error: {e}")
finally:
cursor.close()
conn.close()
#カーソル
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
params: dict形式で取得#[{'no': 1, 'name': 'Alice',...}, ...]
cursor = conn.cursor(cursor=pymysql.cursors.SSCursor)
大量のデータを効率的に取得するためにストリーミングで結果を処理
cursor.execute(query, params=None)
cursor.execute("SELECT * FROM test WHERE no = %s", (1,))
params: プレースホルダーに対応する値のタブルまたはリスト
cursor.executemany(query, param list)
cursor.executemany("INSERT INTO test (no, name) VALUES (%s, %s)", [(1, 'Alice'), (2, 'Bob')])
param list:繰り返し実行するパラメータのリストまたはタブルのリスト
cursor.fetchone()
row = cursor.fetchone()
#結果があれば (1, 'Alice', "2025-01-01") のような形式で1行のみ取得
cursor.rowcount
print(cursor.rowcount) #影響を受けた行数を返す
■接続検証用コンテナをビルド (内部IPを使うrun用)
gcloud builds submit --tag asia-northeast1-docker.pkg.dev/prj/artifact_reg_name/app_name
■IAM？
Cloud SQL設定にCloud SQL 管理者 (roles/cloudsql.admin)、Cloud SQL インスタンス ユーザー (roles/cloudsql.instanceUser)等のIAMが要る?
■IAM？
Cloud SQL設定にCloud SQL 管理者 (roles/cloudsql.admin)、Cloud SQL インスタンス ユーザー (roles/cloudsql.instanceUser)等のIAMが要る?
IAMユーザならいる、ローカルUserなら不要と思われる、ローカルでもCloud SQL Client (roles/cloudsql.client)等は要る
■Cloud SQL MySQL設定
【開発環境】db_instance01
Enterprise / Sandbox / AsiaNorthEast1 (Tokyo) / Single zone
MySQL ver 8.4
Shared core/1cpu 0.6GB/HDD/10GB(auto increase)
PrivateIP/設定にnwが必要(下記)/Enable private path
Auto daily backup 7days (1-5AM) / Enable point-in-time recovery
Week1 sun 0-1am/Enable query insights
root PW: 69696969
【本番環境】
Enterprise plus? キャッシュ使う?
※CloudSQLはTFファイルに記載がなくてもTFステートファイルにPWを含めてしまうためTF化しない
ユーザの作成 sql-user/82828282
■MySQL
サンプルデータ
ORMapperは面倒なのでSQLを使う
ORM Quick Start — SQLAlchemy 2.0 Documentation
【SQLAlchemy】Generic Typesと各種DBの型 対応表
SQLAlchemyでのテーブル定義 #Python - Qiita
- NW: projects/prj/global/networks/sql-vpc-nw
- Connection name: prj:asia-northeast1 db_instance01
ユーザの作成 sql-user/82828282
PWをコードに入れない、シクレMgrに保存
■MySQL
utf8mb4_ja_0900_as_cs_ksを使う
_ai... アクセントを区別しない (Accent Insensitive)
_as... アクセントを区別する (Accent Sensitive)
_ci... 大文字・小文字を区別しない (Case Insensitive)
_cs... 大文字・小文字を区別する (Case Sensitive)
_ks... カナを区別する (Kana Sensitive)
_bin... バイナリ
データベースとテーブルの作成
CREATE DATABASE db;
USE db;
CREATE TABLE test (
no INT(10) UNSIGNED NOT NULL AUTO_INCREMENT,
name VARCHAR(8) NOT NULL,
targetDate TIMESTAMP NOT NULL,
PRIMARY KEY (no),
INDEX index_name (name),
INDEX index_targetDate (targetDate)
)
ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_ja_0900_as_cs_ks;
サンプルデータ
INSERT INTO `table` (`name`, `date`) VALUES ('aaa', '2002-02-23');
ORMapperは面倒なのでSQLを使う
ORM Quick Start — SQLAlchemy 2.0 Documentation
【SQLAlchemy】Generic Typesと各種DBの型 対応表
SQLAlchemyでのテーブル定義 #Python - Qiita
■データベースフラグ
confが直接変更できなためフラグとしてパラメータを渡せる
confが直接変更できなためフラグとしてパラメータを渡せる
Cloud SQL studio (コンソールでMySQLが使える)
MySQLクライアントを使いたいならAuth proxyが必要
Cloud SQLが内部IPだとサーバレスVPCコネクタ、or 外部IPならSQL + auth proxy
内部IPで良いのでVPCを作る、CloudSQLを内部IPで作る
サーバレスVPCアクセスコネクタを作る
vpc: sql-vpc-nw, subnet: sql-vpc-subnet 192.168.77.0/24
Gateway 192.168.77.1, Private Google Access On
sql-vpc-nw-ip-range 192.168.78.0/24 on cloudSQL
run-serverless-vpc-ac 192.168.79.0/28 on Run
ファイアウォールルールでポート (デフォルトで3306など) を開放
Cloud Run のNW設定で、サーバーレス VPCコネクタを選択、ルートオプションとしてすべてのトラフィックを VPC コネクタ経由で送信を選択
CloudSQLを30分程度掛けて起動、接続＞接続テスト
VPC（例: 10.0.0.0/16）
サブネット（Cloud SQL 用）: 10.10.0.0/24（例: us-central1、VPC内）
サブネット（VPCコネクタ用）: 10.8.0.0/28（RunからVPCへ通信用、VPC外）
VPC コネクタのサブネットは 10.8.0.0/28 のような小さな範囲を使用、VPC外だがrun自体がVPC外だから？
VPC コネクタはリージョン単位なので、Cloud Run と Cloud SQL を同じリージョンに配置するのが望ましい
Google Cloudの内部NW設計によりVPC内の異なるサブネット間でも通信可能
VPC内なら異なるリージョンのサブネットでもOK（VPC自体には範囲を設定なしでサブネットでIPが被らなければOKかと
追加の設定なしで、例えば us-central1 の VM から asia-northeast1 の Cloud SQLに直接アクセス可
外部IPの場合：
アプリがrunならサイドカーコンテナとしてAuth Proxyを追加できる
サイドカーは同Pod内なのでループバックアドレス127.0.0.1あるいはlocalhost:5432 (Auth Proxy起動時に指定したポート) に通信しCloudSQLに接続する
GCEにDLしてAuth proxyインストールでもいい
アプリのコネクタはAuth Proxy動いているGCEのIP:ポート番号を指定に通信しCloudSQLに接続する
FWでポートも開けること
run-sql@prj.iam.gserviceaccount.com に必要な権限
Cloud SQL Client (roles/cloudsql.client)
Run Invoker (roles/run.invoker)
Compute Network User (roles/compute.networkUser) -VPCコネクタを使用する
runを建てるが、InternalIPのため同プロジェクト同VPCのGCE を作成し移動してCURLでテスト
curl -H "Authorization: Bearer $(gcloud auth print-identity-token)" "https://run-sql-test-1212124.asia-northeast1.run.app/test"
■MySQLでUUIDを使うか、連番を使うか? > ULIDを使う
UUIDは連番に対し
セキュリティ上より安全、サーバが異なってもユニーク
パフォーマンスが悪い (UUIDをプライマリキーにすると速度が落ちる場合がある)
連番とUUIDの両方を振り出しておく? > ULIDを使うことにする
