シリコンバレーのDeveloperWeek2023 HackathonでOCIを使ってOverall winnerとSponsor Prizeを獲得しました。
OCIの各種PaaSサービスを使ったサーバレスアーキテクチャを組んだので、まとめておきたいと思います。
https://github.com/Contact-IoT-Digital-Signage

今回の記事の内容

OCI Functionsから、OCIの各種PaaSサービスにアクセスする方法

Vault
NoSQL
Object Storage
Media Flow

環境

OCI https://cloud.oracle.com/
Python SDK https://docs.oracle.com/en-us/iaas/tools/python/2.92.0/

(共通)サイナー取得

PythonSDKでOCIのClientを生成する場合、signerを引数に取ります。
実環境のFunctionsで実行する場合、signerを取得するのは以下の式で大丈夫です。

import oci
signer = oci.auth.signers.get_resource_principals_signer()

Vaultアクセス

Vaultを使うとSDKのキーなどを安全に保管できるだけではなく、コードを変更せずにキーを更新したりできるので便利です。

secret_ocid
Secret InformationのOCIDを指定します。
デコード
get_secret_bundleで取得したシークレットはbase64でエンコードされているので、デコード処理が必要です。
バージョン
この処理で取得したシークレットは、最新バージョンになります。（以下画像でlatestがついてるもの）

oci.secrets.SecretsClient({}, signer=signer)
secret_base64 = secret_client.get_secret_bundle(secret_ocid).data.secret_bundle_content.content.encode('utf-8')
secret = base64.b64decode(secret_base64).decode("utf-8")

NoSQLアクセス

フルマネージドなキーバーリューストアが使えます。
SQLのような操作も提供されていて使いやすかったです。

Insert

TableのOCIDを指定して、update_rowを実行します。

table_ocid = os.getenv("TABLE_OCID")
nosql_client = oci.nosql.NosqlClient({}, signer=signer)
nosql_client.update_row(
    table_name_or_id=table_ocid,
    update_row_details=oci.nosql.models.UpdateRowDetails(
        value={
            'userId': "U123456789",
            'userName': "John Doe"
        }
    )
)

List

SQLライクな書き方ができるみたいです。

compartment_ocid = os.getenv("COMPARTMENT_OCID")
table_name = os.getenv("TABLE_NAME")
nosql_client = oci.nosql.NosqlClient({}, signer=signer)
statement = f'select * from {table_name}'
prepare_statement_response = nosql_client.prepare_statement(compartment_id=compartment_ocid,statement=statement)

query_response = nosql_client.query(
    query_details=oci.nosql.models.QueryDetails(
        compartment_id=compartment_ocid,
        statement=prepare_statement_response.data.statement,
        is_prepared=True,
        consistency="EVENTUAL"
    )
)

return query_response.data

Update

プリペアードステートメントが使えます。これもSQLの書き方ですね。

compartment_ocid = os.getenv("COMPARTMENT_OCID")

nosql_client = oci.nosql.NosqlClient({}, signer=signer)
table_name = os.getenv("TABLE_NAME")
update_statement = f'update {table_name} set isActive = true where tpc = $Userid and isActive = false'

prepare_statement_response = nosql_client.prepare_statement(compartment_id=compartment_ocid,statement=update_statement)

nosql_client.query(
    query_details=oci.nosql.models.QueryDetails(
        compartment_id=compartment_ocid,
        statement=prepare_statement_response.data.statement,
        is_prepared=True,
        consistency="EVENTUAL",
        variables={
            '$Userid': 'U123456789'
        }
    )
)

しかしSQLライクな書き方でsetのところに変数を使ってUpdateを発行してもうまくいかなかったので、一旦GetしてInsertしなおす方法を取りました。要検証・・・。（上の例ではsetの中身はisActive = falseと固定だけど、本当にはisActive = $Isactiveみたいにしたかったんですよね。

table_ocid = os.getenv("TABLE_OCID")
compartment_ocid = os.getenv("COMPARTMENT_OCID")
user_id = "U123456789"

user = nosql_client.get_row(
    table_name_or_id=table_ocid,
    key=[user_id],
    compartment_id=compartment_ocid
)

nosql_client.update_row(
    table_name_or_id=table_ocid,
    update_row_details=oci.nosql.models.UpdateRowDetails(
        value={
            'userId': user_id,
            'userName': user.data.value['userName'],
            'age': '20'
        }
    )
)

Delete

キー指定して消すだけです。

table_ocid = os.getenv("TABLE_OCID")
compartment_ocid = os.getenv("COMPARTMENT_OCID")

nosql_client = oci.nosql.NosqlClient({}, signer=signer)

nosql_client.delete_row(
    table_name_or_id=table_ocid,
    key=["U123456789"],
    compartment_id=compartment_ocid
)

Object Storage

S3互換のオブジェクトストレージです。動画などのメディアファイルからJsonファイルの格納まで、いろいろなことができます。
Functionsには実行環境の中にテンポラリー領域があるので、ファイルを一旦Functionsのメモリに落とすことも可能です。

アップロード

BUCKET_NAME_SPACE = os.getenv("BUCKET_NAME_SPACE")
BUCKET_NAME = os.getenv("BUCKET_NAME")

client = oci.object_storage.ObjectStorageClient(config={}, signer=signer)
with open(tmp_file_path, "rb") as f:
    data = f.read()
    client.put_object(BUCKET_NAME_SPACE, BUCKET_NAME, "sample.mp4", data)

ダウンロード

JSONファイルをダウンロードしてオブジェクトとして扱うのを例にします。

object_name = "ダウンロードしたいオブジェクトの名前"
file_content = client.get_object(BUCKET_NAME_SPACE, BUCKET_NAME, object_name).data.content
json_data = json.loads(source_file)

Media Flow

ジョブ実行です。フロー定義はあらかじめGUIからやっておいて、ジョブ実行のみをスクリプトで行いましたが、ちょっと曲者でした。

media_workflow_idを設定するためには、CreateMediaWorkflowJobDetailsではなくCreateMediaWorkflowJobByIdDetailsを使うこと
workflow_identifier_typeはmedia_workflow_idとは違うということ
パラメーターの上書きはjob実行時のjsonファイルをOCIのGUIから確認しながら設定するとよい(View JSONのところからJSONが見れるので、そこのParametersを確認）

media_flow = oci.media_services.MediaServicesClient({}, signer=signer)

compartment_id = os.environ.get('COMPARTMENT_ID')
media_workflow_id = os.environ.get('MEDIA_WORKFLOW_ID')

job_details = oci.media_services.models.CreateMediaWorkflowJobByIdDetails(
    compartment_id=compartment_id,
    workflow_identifier_type=oci.media_services.models.CreateMediaWorkflowJobDetails.WORKFLOW_IDENTIFIER_TYPE_ID,
    media_workflow_id=media_workflow_id,
    display_name=object_file_name,
    parameters={
        "input": {
            "objectName": object_name, # 実行時のパラメータをここで指定
            "objectNameFilename": object_file_name # 実行時のパラメータをここで指定
        },
    }
)
response = media_flow.create_media_workflow_job(job_details)

まとめ

OCI Functionsを中心にOCIの各種サービスにPython SDKでアクセスする方法をまとめました。
基本的にはPython SDKのドキュメントと戦うことになると思います。
今回OCIを利用するのは初めてでしたが、サンプルコードが少なかったので苦労した点が多く、それを解決すべくこの記事を書きました。

質問、ご指摘歓迎します。