任意の SaaS から AWS へのデータ連携を Amazon AppFlow カスタムコネクタで実現 !

{
  "Version": "2012-10-17",
  "Id": "default",
  "Statement": [
    {
      "Sid": "<任意の名前>",
      "Effect": "Allow",
      "Principal": {
        "Service": "appflow.amazonaws.com"
      },
      "Action": "lambda:InvokeFunction",
      "Resource": "arn:aws:lambda:<リージョン名>:<アカウントID>:function:<Lambda関数名>",
      "Condition": {
        "ArnLike": {
          "AWS:SourceArn": "arn:aws:appflow:<リージョン名>:<アカウントID>:*"
        }
      }
    }
  ]
}

authentication_config = auth.AuthenticationConfig(
    is_custom_auth_supported=True,
    custom_auth_config=[
        auth.CustomAuthConfig(
            authentication_type="CUSTOM",
            auth_parameters=[
                auth.AuthParameter(
                    key="Key名",
                    required=True,
                    label="ラベル",
                    description="説明",
                    sensitive_field=True,
                    connector_supplied_values=None,
                )
            ],
        )
    ],
)

runtime_setting = settings.ConnectorRuntimeSetting(
    key="Key名",
    data_type=settings.ConnectorRuntimeSettingDataType.String,
    required=True,
    label="ラベル",
    description="説明",
    scope=settings.ConnectorRuntimeSettingScope.CONNECTOR_PROFILE,
)

import custom_connector_sdk.connector.settings as settings
import custom_connector_sdk.lambda_handler.requests as requests
import custom_connector_sdk.lambda_handler.responses as responses
import custom_connector_sdk.connector.auth as auth
import custom_connector_sdk.connector.configuration as config

def describe_connector_configuration(
        self, request: requests.DescribeConnectorConfigurationRequest
    ) -> responses.DescribeConnectorConfigurationResponse:
    
    # 認証方法 (OAuth2, Basic認証, APIキー, カスタム認証) の設定
    authentication_config = auth.AuthenticationConfig(
        is_custom_auth_supported=True,
        custom_auth_config=[
            auth.CustomAuthConfig(
                authentication_type="CUSTOM",
                auth_parameters=[
                    auth.AuthParameter(
                        key="Key名",
                        required=True,
                        label="ラベル",
                        description="説明",
                        sensitive_field=True,
                        connector_supplied_values=None,
                    )
                ],
            )
        ],
    )
   
    # ランタイムの設定 
    runtime_setting = settings.ConnectorRuntimeSetting(
        key="Key名",
        data_type=settings.ConnectorRuntimeSettingDataType.String,
        required=True,
        label="ラベル",
        description="説明",
        scope=settings.ConnectorRuntimeSettingScope.CONNECTOR_PROFILE,
    )
    
    return responses.DescribeConnectorConfigurationResponse(
        is_success=True,
        connector_owner="本カスタムコネクタの所有者",
        connector_name="本カスタムコネクタ名",
        authentication_config=authentication_config, # 認証方法の設定
        connector_version="本カスタムコネクタのバージョン", 
        supported_api_versions=["サポートしている連携元のデータ API のバージョン"],
        connector_modes=[config.ConnectorModes.SOURCE], # モードの設定 (SOURCE, DESTINATION)
        connector_runtime_setting=[runtime_setting], # ランタイムの設定
    )

import custom_connector_sdk.lambda_handler.requests as requests
import custom_connector_sdk.lambda_handler.responses as responses

# スキーマ定義
sample_entity = context.Entity(
    entity_identifier="ID",
    label="ラベル",
    has_nested_entities=False, # サブオブジェクトの有無
    description="説明",
)

def list_entities(
    self, request: requests.ListEntitiesRequest
) -> responses.ListEntitiesResponse:

    return responses.ListEntitiesResponse(is_success=True, entities=[sample_entity])

import custom_connector_sdk.lambda_handler.requests as requests
import custom_connector_sdk.lambda_handler.responses as responses
import custom_connector_sdk.connector.context as context
import custom_connector_sdk.connector.fields as fields

# スキーマ定義
sample_entity = context.Entity(
    entity_identifier="ID",
    label="ラベル",
    has_nested_entities=False, # サブオブジェクトの有無
    description="説明",
)

# フロー作成のデータフィールドマッピングで呼び出され、フィールドの定義を返す
def describe_entity(
    self, request: requests.DescribeEntityRequest
) -> responses.DescribeEntityResponse:

    sample_field = context.FieldDefinition(
        field_name="フィールド名",
        data_type=fields.FieldDataType.String,
        data_type_label="string",
        label="ラベル",
        description="説明",
        default_value="デフォルト値",
        is_primary_key=True,
        # データの送信元に指定する場合に設定するフィールドの型定義
        read_properties=fields.ReadOperationProperty(
            is_queryable=True, # データ取得時にフィルタとして指定できる項目か
            is_retrievable=True, # データ取得時に表示されるか
            is_nullable=False, # null 値を許容するか
            is_timestamp_field_for_incremental_queries=False, # Date か DateTime 型か
        ),
        write_properties=None, # データの送信先に指定する場合に設定するフィールドの型定義
    )

    entity_definition = context.EntityDefinition(
        entity=sample_entity,
        fields=[sample_field],
    )

    return responses.DescribeEntityResponse(
        is_success=True, entity_definition=entity_definition
    )

import custom_connector_sdk.lambda_handler.requests as requests
import custom_connector_sdk.lambda_handler.responses as responses

# スキーマ定義
sample_entity = context.Entity(
    entity_identifier="ID",
    label="ラベル",
    has_nested_entities=False, # サブオブジェクトの有無
    description="説明",
)

def query_data(
    self, request: requests.QueryDataRequest
) -> responses.QueryDataResponse:

    # JSON オブジェクトの文字列の配列
    # フロー作成時にマッピングしたフィールドのみ送信される
    record_list = [
        '{"DateTime": "2022-01-01 00:00:00", "ScreenPageViews": 1}',
        '{"DateTime": "2022-01-01 00:01:00", "ScreenPageViews": 2}',
    ]

    return responses.QueryDataResponse(is_success=True, records=record_list)