airflow.providers.google.cloud.hooks.cloud_sql¶

此模組包含一個 Google Cloud SQL Hook。

屬性¶

`UNIX_PATH_MAX`
`TIME_TO_SLEEP_IN_SECONDS`
`CLOUD_SQL_PROXY_VERSION_REGEX`
`CLOUD_SQL_PROXY_DOWNLOAD_URL`
`CLOUD_SQL_PROXY_VERSION_DOWNLOAD_URL`
`CONNECTION_URIS`
`CLOUD_SQL_VALID_DATABASE_TYPES`

類¶

`CloudSqlOperationStatus`	操作狀態的輔助類。
`CloudSQLHook`	Google Cloud SQL API 的 Hook。
`CloudSQLAsyncHook`	獲取 Google Cloud SQL 非同步 Hook 的類。
`CloudSqlProxyRunner`	下載 cloud-sql-proxy 並將其作為 Python 程序的子程序執行。
`CloudSQLDatabaseHook`	為 Google Cloud SQL 提供資料庫連線配置（gcpcloudsqldb:// 型別的連線）。

模組內容¶

airflow.providers.google.cloud.hooks.cloud_sql.UNIX_PATH_MAX = 108[原始碼]¶

airflow.providers.google.cloud.hooks.cloud_sql.TIME_TO_SLEEP_IN_SECONDS = 20[原始碼]¶

airflow.providers.google.cloud.hooks.cloud_sql.CLOUD_SQL_PROXY_VERSION_REGEX[原始碼]¶

class airflow.providers.google.cloud.hooks.cloud_sql.CloudSqlOperationStatus[原始碼]¶

操作狀態的輔助類。

PENDING = 'PENDING'[原始碼]¶

RUNNING = 'RUNNING'[原始碼]¶

DONE = 'DONE'[原始碼]¶

UNKNOWN = 'UNKNOWN'[原始碼]¶

class airflow.providers.google.cloud.hooks.cloud_sql.CloudSQLHook(api_version, gcp_conn_id=default_conn_name, impersonation_chain=None, **kwargs)[原始碼]¶

繼承自: airflow.providers.google.common.hooks.base_google.GoogleBaseHook

Google Cloud SQL API 的 Hook。

Hook 中所有使用 project_id 的方法必須使用關鍵字引數而非位置引數呼叫。

引數：

api_version (str) – API 版本。
gcp_conn_id (str) – 用於 GCP 憑據的 Airflow 連線。
impersonation_chain (str | collections.abc.Sequence[str] | None) – 這是用於模擬的可選服務賬號，使用短期憑據。

conn_name_attr = 'gcp_conn_id'[原始碼]¶

default_conn_name = 'google_cloud_sql_default'[原始碼]¶

conn_type = 'gcpcloudsql'[原始碼]¶

hook_name = 'Google Cloud SQL'[原始碼]¶

api_version[原始碼]¶

get_conn()[原始碼]¶

獲取 Cloud SQL 的連線。

返回：: Google Cloud SQL 服務物件。
返回型別：: googleapiclient.discovery.Resource

get_instance(instance, project_id)[原始碼]¶

獲取包含 Cloud SQL 例項資訊的資源。

引數：

instance (str) – 資料庫例項 ID。不包含專案 ID。
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

一個 Cloud SQL 例項資源。

返回型別：

dict

create_instance(body, project_id)[原始碼]¶

建立新的 Cloud SQL 例項。

引數：

body (dict) – Cloud SQL insert API 所需的請求體，如 https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/insert#request-body 中所述。
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

None

返回型別：

None

patch_instance(body, instance, project_id)[原始碼]¶

更新 Cloud SQL 例項的設定。

注意：這不是部分更新，因此必須包含所有要保留的設定的值。

引數：

body (dict) – Cloud SQL patch API 所需的請求體，如 https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/patch#request-body 中所述。
instance (str) – Cloud SQL 例項 ID。不包含專案 ID。
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

None

返回型別：

None

delete_instance(instance, project_id)[原始碼]¶

刪除 Cloud SQL 例項。

引數：

project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。
instance (str) – Cloud SQL 例項 ID。不包含專案 ID。

返回：

None

返回型別：

None

get_database(instance, database, project_id)[原始碼]¶

從 Cloud SQL 例項獲取資料庫資源。

引數：

instance (str) – 資料庫例項 ID。不包含專案 ID。
database (str) – 例項中的資料庫名稱。
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

一個 Cloud SQL 資料庫資源，如 https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/databases#resource 中所述。

返回型別：

dict

create_database(instance, body, project_id)[原始碼]¶

在 Cloud SQL 例項內建立新的資料庫。

引數：

instance (str) – 資料庫例項 ID。不包含專案 ID。
body (dict) – 請求體，如 https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/databases/insert#request-body 中所述。
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

None

返回型別：

None

patch_database(instance, database, body, project_id)[原始碼]¶

更新 Cloud SQL 例項內的資料庫資源。

此方法支援 patch 語義。請參閱 https://cloud.google.com/sql/docs/mysql/admin-api/how-tos/performance#patch。

引數：

instance (str) – 資料庫例項 ID。不包含專案 ID。
database (str) – 要在該例項中更新的資料庫名稱。
body (dict) – 請求體，如 https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/databases/insert#request-body 中所述。
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

None

返回型別：

None

delete_database(instance, database, project_id)[原始碼]¶

從 Cloud SQL 例項刪除資料庫。

引數：

instance (str) – 資料庫例項 ID。不包含專案 ID。
database (str) – 要在該例項中刪除的資料庫名稱。
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

None

返回型別：

None

export_instance(instance, body, project_id)[原始碼]¶

將資料從 Cloud SQL 例項匯出到 Cloud Storage 儲存桶，格式為 SQL dump 或 CSV 檔案。

引數：

instance (str) – Cloud SQL 例項的資料庫例項 ID。不包含專案 ID。
body (dict) – 請求體，如 https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/export#request-body 中所述
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

None

import_instance(instance, body, project_id)[原始碼]¶

從 Cloud Storage 中的 SQL dump 或 CSV 檔案將資料匯入到 Cloud SQL 例項。

引數：

instance (str) – 資料庫例項 ID。不包含專案 ID。
body (dict) – 請求體，如 https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/import#request-body 中所述
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

None

返回型別：

None

clone_instance(instance, body, project_id)[原始碼]¶

將一個例項克隆到目標例項。

引數：

instance (str) – 要克隆的資料庫例項 ID。不包含專案 ID。
instance – 用於克隆的資料庫例項 ID。不包含專案 ID。
body (dict) – 請求體，如 https://cloud.google.com/sql/docs/mysql/admin-api/rest/v1beta4/instances/clone 中所述
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

None

返回型別：

None

create_ssl_certificate(instance, body, project_id)[原始碼]¶

為 Cloud SQL 例項建立 SSL 證書。

引數：

instance (str) – Cloud SQL 例項 ID。不包含專案 ID。
body (dict) – 請求體，如 https://cloud.google.com/sql/docs/mysql/admin-api/rest/v1/sslCerts/insert#SslCertsInsertRequest 中所述
project_id (str) – 包含該例項的專案的專案 ID。如果設定為 None 或缺失，則使用 Google Cloud 連線中的預設專案 ID。

返回：

SslCert 插入響應。更多詳情請參閱： https://cloud.google.com/sql/docs/mysql/admin-api/rest/v1/sslCerts/insert#response-body

airflow.providers.google.cloud.hooks.cloud_sql.CLOUD_SQL_PROXY_DOWNLOAD_URL = 'https://dl.google.com/cloudsql/cloud_sql_proxy.{}.{}'[原始碼]¶

airflow.providers.google.cloud.hooks.cloud_sql.CLOUD_SQL_PROXY_VERSION_DOWNLOAD_URL = 'https://storage.googleapis.com/cloudsql-proxy/{}/cloud_sql_proxy.{}.{}'[原始碼]¶

class airflow.providers.google.cloud.hooks.cloud_sql.CloudSQLAsyncHook(**kwargs)[原始碼]¶

繼承自: airflow.providers.google.common.hooks.base_google.GoogleBaseAsyncHook

獲取 Google Cloud SQL 非同步 Hook 的類。

sync_hook_class[原始碼]¶

async get_operation_name(project_id, operation_name, session)[原始碼]¶

async get_operation(project_id, operation_name)[原始碼]¶

class airflow.providers.google.cloud.hooks.cloud_sql.CloudSqlProxyRunner(path_prefix, instance_specification, gcp_conn_id='google_cloud_default', project_id=PROVIDE_PROJECT_ID, sql_proxy_version=None, sql_proxy_binary_path=None)[原始碼]¶

繼承自: airflow.utils.log.logging_mixin.LoggingMixin

下載 cloud-sql-proxy 並將其作為 Python 程序的子程序執行。

需要下載並啟動 cloud-sql-proxy，然後才能透過資料庫連線連線到 Google Cloud SQL 例項。它會建立到資料庫的安全隧道連線。它使用透過配置傳遞的 Google Cloud 憑據進行授權。

更多關於此代理的詳情可在此處找到： https://cloud.google.com/sql/docs/mysql/sql-proxy

引數：

path_prefix (str) – 代理將下載到的唯一路徑字首，以及用於建立 unix 套接字目錄的字首。
instance_specification (str) – 要連線代理的例項規格。應以 https://cloud.google.com/sql/docs/mysql/sql-proxy#multiple-instances 中 -instances 引數描述的形式指定（通常對於 UNIX 套接字連線為 <project>:<region>:<instance> 形式，對於 TCP 連線為 <project>:<region>:<instance>=tcp:<port> 形式）。
gcp_conn_id (str) – 用於身份驗證的 Google Cloud 連線 ID
project_id (str) – 要連線到的 Google Cloud 專案的可選 ID - 它會覆蓋從 Google Cloud 連接獲取的預設專案 ID。
sql_proxy_version (str | None) – 要下載的 SQL 代理特定版本（例如 'v1.13'）。預設下載最新版本。
sql_proxy_binary_path (str | None) – 如果指定，則將使用指定路徑下的代理，而不是動態生成。這意味著如果該路徑下不存在二進位制檔案，它也將被下載。

path_prefix[原始碼]¶

sql_proxy_was_downloaded = False[原始碼]¶

sql_proxy_version = None[原始碼]¶

download_sql_proxy_dir = None[原始碼]¶

sql_proxy_process: subprocess.Popen | None = None[原始碼]¶

instance_specification[原始碼]¶

project_id = None[原始碼]¶

gcp_conn_id = 'google_cloud_default'[source]¶

command_line_parameters: list[str] = [][source]¶

cloud_sql_proxy_socket_directory[source]¶

sql_proxy_path = 'Uninferable_cloud_sql_proxy'[source]¶

credentials_path[source]¶

start_proxy()[source]¶

啟動 Cloud SQL Proxy。

如果啟動了 Proxy，請記住停止它！

stop_proxy()[source]¶

停止執行中的 Proxy。

在停止使用 Proxy 後應該停止它。

get_proxy_version()[source]¶

返回 Cloud SQL Proxy 的版本。

get_socket_path()[source]¶

獲取 Cloud SQL Proxy 使用的 UNIX 套接字路徑。

返回：: Proxy 建立的套接字的動態生成路徑。
返回型別：: str

airflow.providers.google.cloud.hooks.cloud_sql.CONNECTION_URIS: dict[str, dict[str, dict[str, str]]][source]¶

airflow.providers.google.cloud.hooks.cloud_sql.CLOUD_SQL_VALID_DATABASE_TYPES = ['postgres', 'mysql'][source]¶

class airflow.providers.google.cloud.hooks.cloud_sql.CloudSQLDatabaseHook(gcp_cloudsql_conn_id='google_cloud_sql_default', gcp_conn_id='google_cloud_default', impersonation_chain=None, default_gcp_project_id=None, sql_proxy_binary_path=None, ssl_cert=None, ssl_key=None, ssl_root_cert=None, ssl_secret_id=None, **kwargs)[source]¶

基類：airflow.hooks.base.BaseHook

為 Google Cloud SQL 提供資料庫連線配置（gcpcloudsqldb:// 型別的連線）。

此 Hook 是一個“元” Hook。它不執行實際的連線。

它的作用是檢索 gcpcloudsql:// 連線中配置的所有引數，根據需要啟動/停止 Cloud SQL Proxy，在資料庫中動態生成 Postgres 或 MySQL 連線，並返回實際的 Postgres 或 MySQL Hook。返回的 Postgres/MySQL Hook 將按照配置使用直接連線或 Cloud SQL Proxy 的套接字/TCP 連線。

Hook 的主要引數從標準 URI 元件中檢索：

user - 用於向資料庫進行身份驗證的使用者名稱（來自 URI 的登入部分）。
password - 用於向資料庫進行身份驗證的密碼（來自 URI 的密碼部分）。
public_ip - 公共連線要連線到的 IP 地址（來自 URI 的主機部分）。
public_port - 公共連線要連線到的埠（來自 URI 的埠部分）。
database - 要連線的資料庫（來自 URI 的 schema 部分）。
sql_proxy_binary_path - Cloud SQL Proxy 二進位制檔案的可選路徑。如果未指定二進位制檔案或二進位制檔案不存在，則會自動下載。

剩餘引數從 extras 中檢索（URI 查詢引數）：

project_id - 可選，Cloud SQL 例項所在的 Google Cloud 專案。
例項存在。如果缺失，則使用傳入的預設專案 ID。
instance - Cloud SQL 資料庫例項的例項名稱。
location - Cloud SQL 例項的位置（例如 europe-west1）。
database_type - 資料庫例項的型別（MySQL 或 Postgres）。
use_proxy - （預設為 False）是否應使用 SQL Proxy 連線到 Cloud SQL DB。
use_ssl - （預設為 False）連線到 Cloud SQL DB 時是否應使用 SSL。不能同時使用 Proxy 和 SSL。
use_iam - （預設為 False）連線到 Cloud SQL DB 時是否應使用 IAM。使用 IAM 時，密碼欄位應為空字串。
sql_proxy_use_tcp - （預設為 False）如果設定為 true，則透過 Proxy 使用 TCP 進行連線，否則使用 UNIX 套接字。
sql_proxy_version - 要下載的 Proxy 的特定版本（例如 v1.13）。如果未指定，則下載最新版本。
sslcert - 使用 SSL 時用於身份驗證的客戶端證書路徑。
sslkey - 使用 SSL 時用於身份驗證的客戶端私鑰路徑。
sslrootcert - 使用 SSL 時用於身份驗證的伺服器證書路徑。

引數：

gcp_cloudsql_conn_id (str) – 連線的 URL
gcp_conn_id (str) – 用於連線 Google Cloud 以進行 cloud-sql-proxy 身份驗證的連線 ID。
default_gcp_project_id (str | None) – 如果連線 URL 中未指定 project_id，則使用的預設專案 ID。
ssl_cert (str | None) – 可選。使用 SSL 時用於身份驗證的客戶端證書路徑。覆蓋連線欄位 sslcert。
ssl_key (str | None) – 可選。使用 SSL 時用於身份驗證的客戶端私鑰路徑。覆蓋連線欄位 sslkey。
ssl_root_cert (str | None) – 可選。使用 SSL 時用於身份驗證的伺服器證書路徑。覆蓋連線欄位 sslrootcert。
ssl_secret_id (str | None) –
可選。Google Cloud Secret Manager 中儲存 SSL 證書的 Secret ID，格式如下：

{‘sslcert’: ‘’,
’sslkey’: ‘’, ‘sslrootcert’: ‘’}

覆蓋連線欄位 sslcert、sslkey、sslrootcert。請注意，根據 Secret Manager 的要求，上述字典應儲存為字串並使用 base64 編碼。請注意，此引數與 ssl_cert、ssl_key、ssl_root_cert 引數不相容。

conn_name_attr = 'gcp_cloudsql_conn_id'[source]¶

default_conn_name = 'google_cloud_sqldb_default'[source]¶

conn_type = 'gcpcloudsqldb'[source]¶

hook_name = 'Google Cloud SQL Database'[source]¶

gcp_conn_id = 'google_cloud_default'[source]¶

gcp_cloudsql_conn_id = 'google_cloud_sql_default'[source]¶

impersonation_chain = None[source]¶

cloudsql_connection[source]¶

extras[source]¶

project_id[source]¶

instance[source]¶

database[source]¶

location[source]¶

database_type[source]¶

use_proxy = False[source]¶

use_ssl = False[source]¶

use_iam = False[source]¶

sql_proxy_use_tcp = False[source]¶

sql_proxy_version[source]¶

sql_proxy_binary_path = None[source]¶

public_ip[source]¶

public_port[source]¶

ssl_cert = None[source]¶

ssl_key = None[source]¶

ssl_root_cert = None[source]¶

ssl_secret_id = None[source]¶

sql_proxy_tcp_port = None[source]¶

sql_proxy_unique_path: str | None = None[source]¶

db_hook: airflow.hooks.base.BaseHook | None = None[source]¶

reserved_tcp_socket: socket.socket | None = None[source]¶

db_conn_id = ''[source]¶

property sslcert: str | None[source]¶

property sslkey: str | None[source]¶

property sslrootcert: str | None[source]¶

validate_ssl_certs()[source]¶

SSL 證書驗證器。

返回：: None
返回型別：: None

validate_socket_path_length()[source]¶

驗證套接字路徑長度。

返回：: None 或丟擲 AirflowException 異常
返回型別：: None

create_connection()[source]¶

建立連線。

連線 ID 將根據是否使用 Proxy、TCP、UNIX 套接字、SSL 隨機生成。

get_sqlproxy_runner()[source]¶

獲取 Cloud SQL Proxy 執行器。

它用於管理每個任務的 Proxy 生命週期。

返回：: Cloud SQL Proxy 執行器。
返回型別：: CloudSqlProxyRunner

get_database_hook(connection)[source]¶

獲取資料庫 Hook。

這是實際的 Postgres 或 MySQL 資料庫 Hook，它使用 Proxy 或直接連線到 Google Cloud SQL 資料庫。

cleanup_database_hook()[source]¶

使用後清理資料庫 Hook。

reserve_free_tcp_port()[source]¶

保留供 Cloud SQL Proxy 使用的空閒 TCP 埠。

free_reserved_port()[source]¶

釋放 TCP 埠。

使其立即可供 Cloud SQL Proxy 使用。