TIPCommon.adapters

class TIPCommon.adapters.pubsub.pubsub.PubSubAdapter

class TIPCommon.adapters.pubsub.pubsub.PubSubAdapter(session, project_id=None, logger=None, region=None)

基地：object

用於管理 Google Cloud 專案 Pub/Sub 主題和訂閱項目的介面卡類別。

ack

ack(sub_name, ack_ids)

確認 PubSubAdapter.pull() 傳回的 AcknowledgeRequest 回應中，與 ackIDs 相關聯的訊息。

參數
sub_name	(str) 訂閱項目名稱。
ack_ids	(list[str]) 要確認的訊息確認 ID 清單。Pub/Sub 系統會在 PubSubAdapter.pull() 回應中傳回清單。

static build_pubsub_message

static build_pubsub_message(message_content, encoding='utf-8', ordering_key=None, **attr)

建立 PubSubMessage 物件。

參數
message_content	(str) 訊息內容。
encoding	(str) 用於編碼或解碼訊息文字的編碼類型。 預設值為 UTF-8。
ordering_key	選填 (str) 如果使用這個參數，系統會識別所有相關訊息，這些訊息必須遵循發布順序。
**attr	選用 (str) 要以訊息物件屬性形式傳遞的屬性。

傳回

TIPCommon.adapters.pubsub.PubSubMessage 物件。

傳回類型

PubSubMessage

create_subscription

create_subscription(sub_name, topic, **attr)

為指定主題建立 Pub/Sub 訂閱項目。

參數
sub_name	(str) 訂閱項目名稱。
topic	(str) 要建立訂閱項目的 Pub/Sub 主題名稱。
**attr	要傳遞至訂閱要求的其他參數。

傳回

所建立訂閱項目的 TIPCommon.adapters.pubsub.Subscription 物件。

傳回類型

Subscription

create_topic

create_topic(topic_name)

在 Google Cloud 專案中建立 Pub/Sub 主題。

參數
topic_name	(str) 主題名稱。名稱必須符合 Google Cloud 資源名稱規則。

傳回

所建立主題的 TIPCommon.adapters.pubsub.Topic 物件。

傳回類型

Topic

delete_subscription

delete_subscription(sub_name)

從 Google Cloud 專案中刪除 Pub/Sub 訂閱項目。

參數
sub_name	(str) 要移除的訂閱項目名稱。

delete_topic

delete_topic(topic_name)

從 Google Cloud 專案中刪除 Pub/Sub 主題。

參數
topic_name	(str) 要移除的主題。

from_credentials static

static from_credentials(credentials, project_id=None, verify_ssl=True, quota_project=None, logger=None, region=None)

從 google.oauth2.credentials.Credentials 物件建立 PubSubAdapter 物件。

參數
credentials	(google.oauth2.credentials.Credentials) google.oauth2.credentials.Credentials 物件。
project_id	選填 (str) 專案 ID。 Google Cloud 如未提供，函式會嘗試使用憑證物件中設定的專案 (預設)。
verify_ssl	選填 (bool) 指定是否為 HTTP 工作階段啟用 SSL 憑證驗證。
quota_project	選填 (str) 用於配額和帳單的專案。
logger	選填 (SiemplifyLogger) SiemplifyLogger 物件。
region	(str) Pub/Sub 的運作區域。

傳回

PubSubAdapter 物件。

傳回類型

PubSubAdapter

from_service_account_info 的 static

static subscription_name(project_id, sub_name)

從 service_account JSON 建立 PubSubAdapter 物件。

參數
user_service_account	(str) 文字格式的 Google Cloud 服務帳戶 JSON。
project_id	選填 (str) 專案 ID。 Google Cloud 如未提供，函式會嘗試使用憑證物件中設定的專案 (預設)。
verify_ssl	選填 (bool) 指定是否為 HTTP 工作階段啟用 SSL 憑證驗證。
quota_project	選填 (str) 用於配額和帳單的專案。
logger	選填 (SiemplifyLogger) SiemplifyLogger 物件。

傳回

PubSubAdapter 物件。

傳回類型

PubSubAdapter

get_subscription

get_subscription(sub_name, topic=None, create_if_not_exist=False, **attr)

擷取 Pub/Sub 訂閱項目。

參數
sub_name	(str) 訂閱項目名稱。
topic	(str) 要建立訂閱項目的 Pub/Sub 主題名稱。 如果 create_if_not_exist 為 True，則為必要欄位。
create_if_not_exist	如果 Google Cloud 中沒有 Pub/Sub 訂閱項目，則會建立該項目。
**attr	要傳遞至訂閱項目建立要求的其他參數。

傳回

擷取訂閱項目的 TIPCommon.adapters.pubsub.Subscription 物件。

傳回類型

Subscription

get_topic

get_topic(topic_name, create_if_not_exist=False)

從已設定的 Google Cloud專案中擷取 Pub/Sub 主題物件。

參數
topic_name	(str) 主題名稱 (簡化版，不含 projects/PROJECT_ID/topics/ 前置字串)。
create_if_not_exist	bool 如果不存在，則在 Google Cloud中建立 Pub/Sub 主題。

傳回

接收主題的 TIPCommon.adapters.pubsub.Topic 物件。

傳回類型

Topic

patch_subscription

patch_subscription(sub_name, topic_name, push_config=None, bigquery_config=None, cloud_storage_config=None, ack_deadline_seconds=None, retain_acked_messages=None, retention_duration=None, labels=None, enable_message_ordering=None, expiration_policy=None, query_filter=None, dead_letter_policy=None, return_policy=None, detached=None, enable_once_delivery=None)

更新現有訂閱項目。

注意： 部分訂閱屬性無法修改。

參數
sub_name	(str) 訂閱項目名稱。
topic_name	(str) 這個訂閱項目接收訊息的主題名稱。

傳回

收到的訂閱項目 TIPCommon.adapters.pubsub.Subscription 物件。

傳回類型

Subscription

patch_topic

patch_topic(topic_name, labels=None, message_storage_policy=None, kms_key_name=None, schema_settings=None, satisfies_pzs=None, retention_duration=None)

更新現有主題。

注意： 部分主題屬性無法修改。

參數
topic_name	(str) 主題名稱。

傳回

接收主題的 TIPCommon.adapters.pubsub.Topic 物件。

傳回類型

Topic

publish

publish(topic_name, messages)

將 PubSubMessage 物件清單發布至主題。

參數
topic_name	(str) 要發布訊息的主題名稱。
messages	list PubSubMessage 物件清單。 您可以使用 PubSubAdapter.build_message() 靜態方法建立清單。

傳回

訊息 ID 清單。

傳回類型

list[str]

提取

pull(sub_name, limit, timeout=60, encoding='utf-8')

從 Pub/Sub 訂閱項目提取訊息。

參數
sub_name	(str) 訂閱項目名稱。
limit	int 這項要求傳回的訊息數量上限。
timeout	int HTTP 要求逾時時間 (以秒為單位)。 預設值為 60 秒。
encoding	(str) Pub/Sub 訊息編碼。預設值為 utf-8。

傳回

TIPCommon.adapters.pubsub.ReceivedMessage 物件清單。

傳回類型

list[ReceivedMessage]

static subscription_name

static subscription_name(project_id, sub_name)

以「projects/project_id/subscriptions/subscription_name」格式擷取完整訂閱項目名稱。

參數
project_id	(str) 包含資源的專案名稱。
sub_name	(str) Pub/Sub 訂閱項目名稱。

傳回

完整訂閱名稱，格式如下：projects/project_id/subscriptions/subscription_name。

傳回類型

str

static topic_name

static topic_name(project_id, topic)

擷取 projects/project_id/topics/topic_name。

參數
project_id	(str) 包含資源的專案名稱。
topic	(str) Pub/Sub 主題名稱。

傳回

完整主題名稱：projects/project_id/topics/topic_name。

傳回類型

str

class TIPCommon.adapters.pubsub.data_models.PubSubMessage

class TIPCommon.adapters.pubsub.data_models.PubSubMessage(raw_data: 'dict', data: 'str' = None, attributes: 'dict' = None, message_id: 'str' = None, publish_time: 'int' = None, ordering_key: 'str' = None)

基底：object

attributes: dict= None

data: str= None

json()

message_id: str= None

ordering_key: str= None

publish_time: int= None

raw_data: dict

class TIPCommon.adapters.pubsub.data_models.ReceivedMessage

class TIPCommon.adapters.pubsub.data_models.ReceivedMessage(raw_data: 'dict', ack_id: 'str', message: 'PubSubMessage', delivery_attempt: 'int')

基底：object

ack_id: str

delivery_attempt: int

json()

message: PubSubMessage

raw_data: dict

class TIPCommon.adapters.pubsub.data_models.SchemaSettings

class TIPCommon.adapters.pubsub.data_models.SchemaSettings(raw_data: 'dict', schema: 'str', encoding: 'str' = None, first_revision_id: 'str' = None, last_revision_id: 'str' = None)

基底：object

encoding: str= None

first_revision_id: str= None

json()

last_revision_id: str= None

raw_data: dict

schema: str

class TIPCommon.adapters.pubsub.data_models.Subscription

class TIPCommon.adapters.pubsub.data_models.Subscription(raw_data: 'dict', name: 'str', identifier: 'str', topic_identifier: 'str', state: 'str', ack_deadline_secs: 'int' = None, retain_ack_messages: 'bool' = None, message_retention_duration: 'int' = None, labels: 'dict' = None, message_ordering: 'bool' = None, query_filter: 'str' = None, topic_message_retention_duration: 'int' = None)

基底：object

ack_deadline_secs: int= None

identifier: str

json()

labels: dict= None

message_ordering: bool= None

message_retention_duration: int= None

name: str

query_filter: str= None

raw_data: dict

retain_ack_messages: bool= None

state: str

topic_identifier: str

topic_message_retention_duration: int= None

class TIPCommon.adapters.pubsub.data_models.Topic

class TIPCommon.adapters.pubsub.data_models.Topic(raw_data: 'dict', name: 'str', identifier: 'str', labels: 'dict' = None, schema_settings: 'SchemaSettings' = None, message_retention_duration: 'int' = None)

基底：object

identifier: str

json()

labels: dict= None

message_retention_duration: int= None

name: str

raw_data: dict

schema_settings: SchemaSettings= None

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。