LoginSignup
4
1

【UiPath】Automation CloudのData ServiceにOpenAPIライブラリを使ってPythonからアクセスする

Last updated at Posted at 2024-01-25

この記事の内容は?

本稿ではUiPath Automation CloudのData ServiceにPythonのOpenAPIライブラリ(openapi-python-client)を使ってアクセスする方法をご紹介します。

OpenAPIとはWeb APIの仕様を記述するための標準規格です。

Data Serviceはこの規格に準じているため、Data Serviceが提供するAPIの仕様を元にクライアントライブラリを自動的に生成することができます。

Data ServiceのAPIに関するガイドは次のものです。

この記事ではPythonのライブラリであるopenapi-python-clientを使用し、Data Serviceのクライアントライブラリを自動生成し、それを使った操作の例を記載します。

このライブラリを使用する上での多少の学習コストはありますが、Automation Cloudの認証およびData Serviceの操作をライブラリを使って行えるため、直接HTTPリクエストを操作する方法や、独自にライブラリを生成する場合と比べ、より高速に目的の処理を記述することができます。

環境の準備

次の環境を用意ください。

  • Data Serviceが使用できるAutomation Cloudの組織
  • Python
  • Pythonのパッケージマネージャーのpipコマンド

お手元にPythonを実行できる環境がない場合、Google ColabのようなJupyter Notebookサービスを使用するのが最もお手軽だと思います。

Automation Cloudの設定

まずはData Serviceにアクセスし、適当なエンティティを作成します。
ここではTestEntityとし、フィールドとしてText型のTestFieldを追加しました。

image.png

エンティティの作成が終わったら、Automation Cloudに外部アプリケーションを追加し、このエンティティのみか、あるいはData Service全体へのアクセスを作成した機密アプリケーションに与えます。

この部分の設定は次の記事に詳細に記載されているため、手順はこちらを参照してください。

一例として、今回は次のような権限をアプリケーションに与えました。
image.png

手順の実施後、機密アプリケーションのIDとシークレットが手元にあることを確認します。

DataService.jsonのダウンロード

次のスクリーンショットを参考に、[その他のオプション] > [APIアクセス] > [OpenAPI ファイル (JSON) をダウンロード]より、DataService.jsonファイルをダウンロードします。
このファイルはOpenAPIに準じたものでData ServiceのAPIの仕様が記載されています。
image.png

image.png

依存ライブラリのインストール

次のコマンドで必要となるライブラリをインストールします。

pip install openapi-python-client oauthlib requests requests_oauthlib

注意: Google ColabなどのJupyter Notebook環境の場合、Notebook内でpip、mv、rmなどのコマンドを実行するには次のようにコマンドの最初に!を記載する必要があります。

!pip install openapi-python-client oauthlib requests requests_oauthlib

クライアントコードの生成

DataService.jsonをopenapi-python-clientに与え、クライアントのコードを生成します。

openapi-python-client generate --path DataService.json

カレントディレクトリ以下にentity-store-defaulttenant-clientというフォルダが生成され、その中にライブラリの格納されたフォルダ(entity-store-defaulttenant-client/entity_store_defaulttenant_client)が存在します。
カレントディレクトリのPythonコードから参照するため、このフォルダをカレントディレクトリに移動します。

mv ./entity-store-defaulttenant-client/entity_store_defaulttenant_client .

Pythonコードの実装

さてここまででPythonコードを書いていく準備ができました。

認証

まずはoauthlibを使い、機密アプリケーションのクライアントIDとシークレットによって認証を行います。
ここではスコープとしてDataService.Defaultを指定する必要があります。

from oauthlib.oauth2 import BackendApplicationClient
from requests_oauthlib import OAuth2Session

TOKEN_URL = 'https://cloud.uipath.com/identity_/connect/token'
CLIENT_ID = '<機密アプリケーションのクライアントID>'
CLIENT_SECRET = '<機密アプリケーションのシークレット>'
CLIENT_SCOPE = 'DataService.Default'

ba_client = BackendApplicationClient(client_id=CLIENT_ID)
oauth = OAuth2Session(client=ba_client)

token = oauth.fetch_token(
    token_url=TOKEN_URL,
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET,
    response_type='client_credentials',
    scope=CLIENT_SCOPE
)
access_token = token['access_token']

クライアントの準備

次のコードでDataServiceのAPIへアクセスするためのクライアントを準備します。
このクライアントはこの後のコード全てで使用します。

from entity_store_defaulttenant_client import AuthenticatedClient
client = AuthenticatedClient(
    base_url="https://cloud.uipath.com/<組織名>/<テナント名>/dataservice_/api/",
    token=access_token
)

データを追加する

まずは次のコードを眺めてみてください。
実行するとTestEntityというエンティティにデータを追加することができます。

from entity_store_defaulttenant_client.models import TestEntity
from entity_store_defaulttenant_client.api.test_entity import add_test_entity

data = TestEntity(
    test_field = 'test data',
)
res = add_test_entity.sync_detailed(client=client, body=data)

res.status_codeが示すHTTPステータスコードが200なら成功です。
追加されたデータはDataServiceの[エンティティ] > [TestEntity]内の[データ]タブから確認できます。
image.png

権限が不足していると例えば次のようなエラーが返ります。

403 You don't have permission to access the entity, field or record or you are using an unsupported robot type. Please contact your administrator for necessary permissions.

ここでentity_store_defaulttenant_clientというライブラリをインポートしていますが、これは先ほどopenapi-python-clientによって生成したものです。
フォルダ構成としては次のようになっています。

パス コンテンツ
entity_store_defaulttenant_client/models データのモデル
entity_store_defaulttenant_client/api/test_entity 各エンティティのAPI操作に対応するクラス

各エンティティのAPI操作に対応するクラスや、そのクラスが使用するデータのモデルは自動的に生成されますが、実際にコード上で使用するためには自分で適切なファイルを選び、インポートする必要があります。

では次の例に行きましょう。

データを検索する

from entity_store_defaulttenant_client.models import QueryRequest, QueryFilter, QueryFilterGroup
from entity_store_defaulttenant_client.api.test_entity import query_test_entity

query_filter = QueryFilter(
    field_name='TestField',
    operator='=', 
    value='test data',
)
query_filter_group = QueryFilterGroup(
    logical_operator=0, # 0 for ALL, 1 for OR
    query_filters=[query_filter],
)
query_request = QueryRequest(
    selected_fields=['Id', 'TestField'],
    filter_group=query_filter_group,
)
res = query_test_entity.sync(client=client, body=query_request)
record_id = res.value[0].id

record_idとしてUUIDが返れば成功です。
リクエストに必要なデータが全てPythonのクラスとして生成されているため、色々なクラスを組み合わせると複雑なクエリも実現できますが、ある程度の慣れが必要かもしれません。

データを取得する

record_idを取得できれば色々な操作ができます。
ここではデータを取得する例を示します。

from entity_store_defaulttenant_client.models import TestEntity
from entity_store_defaulttenant_client.api.test_entity import get_test_entity

res = get_test_entity.sync_detailed(client=client, id=record_id)

バッチ系APIを使用する

データ1個ごとにAPIへ1回リクエストを送信するのは非効率なので、DataServiceはbatchでリクエストを送信するためのAPIを提供しています。
データ追加をまとめて行うためにはここではbatch_add_test_entityクラスを使用します。

from entity_store_defaulttenant_client.models import TestEntity
from entity_store_defaulttenant_client.api.test_entity import batch_add_test_entity

def to_entity(text):
    return TestEntity(
        test_field = text,
    )

data = [to_entity(x) for x in ['test data 1', 'test data2']]

res = batch_add_test_entity.sync_detailed(client=client, body=data)

上記コードでうまくいくはずで、実際にウェブインターフェイス上でエンティティは生成されているものの、2024-01-22時点では次のエラーが発生します。
注記: この問題は将来的に修正される予定です。

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
Cell In[17], line 11
      5     return TestEntity(
      6         test_field = text,
      7     )
      9 data = [to_entity(x) for x in ['test data 1', 'test data2']]
---> 11 res = batch_add_test_entity.sync_detailed(client=client, body=data)
     12 res

File ~/work/2024-01-22_qiita/entity_store_defaulttenant_client/api/test_entity/batch_add_test_entity.py:109, in sync_detailed(client, body, expansion_level, fail_on_first)
     99 kwargs = _get_kwargs(
    100     body=body,
    101     expansion_level=expansion_level,
    102     fail_on_first=fail_on_first,
    103 )
    105 response = client.get_httpx_client().request(
    106     **kwargs,
    107 )
--> 109 return _build_response(client=client, response=response)

File ~/work/2024-01-22_qiita/entity_store_defaulttenant_client/api/test_entity/batch_add_test_entity.py:73, in _build_response(client, response)
     66 def _build_response(
     67     *, client: Union[AuthenticatedClient, Client], response: httpx.Response
     68 ) -> Response[Union[Any, TestEntityBatchResponse]]:
     69     return Response(
     70         status_code=HTTPStatus(response.status_code),
     71         content=response.content,
     72         headers=response.headers,
---> 73         parsed=_parse_response(client=client, response=response),
     74     )

File ~/work/2024-01-22_qiita/entity_store_defaulttenant_client/api/test_entity/batch_add_test_entity.py:51, in _parse_response(client, response)
     47 def _parse_response(
     48     *, client: Union[AuthenticatedClient, Client], response: httpx.Response
     49 ) -> Optional[Union[Any, TestEntityBatchResponse]]:
     50     if response.status_code == HTTPStatus.OK:
---> 51         response_200 = TestEntityBatchResponse.from_dict(response.json())
     53         return response_200
     54     if response.status_code == HTTPStatus.UNAUTHORIZED:

File ~/work/2024-01-22_qiita/entity_store_defaulttenant_client/models/test_entity_batch_response.py:68, in TestEntityBatchResponse.from_dict(cls, src_dict)
     66     failure_records = UNSET
     67 else:
---> 68     failure_records = TestEntityBatchResponseFailureRecords.from_dict(_failure_records)
     70 test_entity_batch_response = cls(
     71     success_records=success_records,
     72     failure_records=failure_records,
     73 )
     75 test_entity_batch_response.additional_properties = d

File ~/work/2024-01-22_qiita/entity_store_defaulttenant_client/models/test_entity_batch_response_failure_records.py:49, in TestEntityBatchResponseFailureRecords.from_dict(cls, src_dict)
     46 from ..models.test_entity import TestEntity
     48 d = src_dict.copy()
---> 49 error = d.pop("error", UNSET)
     51 _record = d.pop("record", UNSET)
     52 record: Union[Unset, TestEntity]

TypeError: pop expected at most 1 argument, got 2

原因を追っていくと、どうやらDataService.jsonに問題があるようでした。
バッチ系APIが返すデータのfailureRecordsは配列ですが、OpenAPIの定義ではobjectになっています。
例えばTestEntityBatchResponseを見てみると、次のようになっていますが、正しくはその次に示すスキーマになります。

問題のあるスキーマ

      "TestEntityBatchResponse": {
        "type": "object",
        "properties": {
          "successRecords": {
            "type": "array",
            "items": {
              "$ref": "#/components/schemas/TestEntity"
            }
          },
          "failureRecords": {
            "type": "object",
            "properties": {
              "error": {
                "type": "string"
              },
              "record": {
                "$ref": "#/components/schemas/TestEntity"
              }
            }
          }
        }
      },

修正したスキーマ

      "TestEntityBatchResponse": {
        "type": "object",
        "properties": {
          "successRecords": {
            "type": "array",
            "items": {
              "$ref": "#/components/schemas/TestEntity"
            }
          },
          "failureRecords": {
			"type": "array",
            "items": {
              "type": "object",
              "properties": {
                "error": {
                  "type": "string"
                },
                "record": {
                  "$ref": "#/components/schemas/TestEntity"
              }
            }
          }
        }
      },

修正方法

この問題を修正するためには、まず上記のようにfailureRecordsの定義を修正します。

そして次の2つのフォルダを削除します。

rm -rf ./entity-store-defaulttenant-client
rm -rf ./entity_store_defaulttenant_client

上記のステップを忘れると次のエラーになります。

Unable to generate the client Directory already exists. Delete it or use the update command.

最後にDataService.jsonを上記の通り修正し、再度openapi-python-client generate --path DataService.jsonコマンドを実行してクライアントライブラリを生成します。

4
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
4
1