はじめに
前回のビュー編に続いて、今回は文書編です。文書の登録、更新を行っていきます。LotusScriptなどでNotesクラスを使って文書を操作する場合との最大の違いは「新規作成(POST)、全体更新(PUT)、部分更新(PATCH)は違う違うコマンドなので処理を分ける必要がある」という点です。これはREST APIの標準仕様なので、今までSaveコマンドや@Command([FileSave])で済んできた身からすると面倒なのですが、これが今の標準なのでこれを機会に学んでいただければと思います。
Notes文書の取得(テキストのみ)－GETコマンド
文書ID(UNID)から対象文書を取得し、特定フィールドを出力するサンプルコードです。
"""
Overview:
文書の取得(GET) - 添付ファイルなし
"""
import requests
# 定数宣言
HOST_URL = 'https://www.hogehoge.com/'
DB_PATH = 'hoge/hoge.nsf'
USER_NAME = '(任意)'
USER_PW = '(任意)'
def create_session():
"""
Summary:
セッション認証を行い、セッションとレスポンスを返す
Args:
(なし)
Returns:
session :セッション
response :レスポンス
"""
# ログインペイロードの作成
payload = {
'username': USER_NAME,
'password': USER_PW
}
# ログインリクエストを送信してセッションを確立
session = requests.Session()
response = session.post(HOST_URL + '?Login', data=payload)
return session, response
def get_document(arg_session, arg_doc_unid):
"""
Summary:
文書を取得
Args:
arg_session :ログイン済みセッション
arg_doc_unid :文書ID
Returns:
(なし)
"""
# 文書を取得
api_endpoint = f'{HOST_URL}{DB_PATH}/api/data/documents/unid/{arg_doc_unid}'
data_response = arg_session.get(api_endpoint)
# 応答からフィールドと文書IDを取得
if data_response.status_code == 200:
data = data_response.json()
print(data['Field02'], data['@unid'])
else:
print('Access failed: ', data_response.status_code)
def main():
# セッション認証を行う
session, response = create_session()
# レスポンスコードから成否を判定
if response.status_code == 200:
doc_unid = '(文書ID)'
get_document(session, doc_unid)
else:
print('Login failed: ', response.status_code)
if __name__ == '__main__':
main()
Notes文書の取得(リッチテキストに画像あり)－GETコマンド
文書ID(UNID)から対象文書を取得し、特定フィールドを出力、リッチテキストフィールドに添付された画像をローカルフォルダに保存するサンプルコードです。画像の扱いは他のREST APIでもBase64でエンコード・デコードするので、Domino Data AccessではないREST APIにも適用できるかと思います。
"""
Overview:
文書の取得(GET) - 添付ファイル画像のファイル化
"""
import requests
import base64
# 定数宣言
HOST_URL = 'https://www.hogehoge.com/'
DB_PATH = 'hoge/hoge.nsf'
USER_NAME = '(任意)'
USER_PW = '(任意)'
TEMP_PATH = 'c:\\tmp\\'
def create_session():
"""
Summary:
セッション認証を行い、セッションとレスポンスを返す
Args:
(なし)
Returns:
session :セッション
response :レスポンス
"""
# ログインペイロードの作成
payload = {
'username': USER_NAME,
'password': USER_PW
}
# ログインリクエストを送信してセッションを確立
session = requests.Session()
response = session.post(HOST_URL + '?Login', data=payload)
return session, response
def save_image_file(arg_file_path, arg_image_binary):
"""
Summary:
イメージファイルを保存
Args:
arg_file_path :保存先ファイルパス
arg_image_binary :画像バイナリ
Returns:
(なし)
"""
with open(arg_file_path, 'wb') as file:
file.write(arg_image_binary)
def extract_image_data_and_filename(arg_data):
"""
Summary:
リッチテキストフィールドからファイル名、Base64エンコードイメージを取得し、戻り値として返す
Base64イメージはデコード
Args:
arg_data :リッチテキストフィールド
Returns:
images(file_name,image_binary) :ファイル名、デコード済み画像バイナリ
"""
images = [] # 戻り値用配列
if not arg_data or arg_data['type'] != 'multipart':
raise Exception("Invalid image body data")
for part in arg_data.get('content', []):
if part['contentType'].startswith('image/') and 'data' in part:
# base64文字列取得とデコード
image_data = part['data']
if part.get('contentTransferEncoding') == 'base64':
image_binary = base64.b64decode(image_data)
else:
image_binary = image_data.encode() # If the data is not Base64 encoded
# ファイル名取得
file_name = part.get('name') or part.get(
'contentDisposition', '').split('filename=')[-1].strip('"')
if not file_name:
raise Exception('Filename not found in the response')
images.append((file_name, image_binary))
if not images:
raise Exception('No image data found in the response')
return images
def get_document(arg_session, arg_doc_unid):
"""
Summary:
文書を取得
Args:
arg_session :ログイン済みセッション
arg_doc_unid :文書ID
Returns:
(なし)
"""
# 文書を取得
api_endpoint = f'{HOST_URL}{DB_PATH}/api/data/documents/unid/{arg_doc_unid}'
data_response = arg_session.get(api_endpoint)
# 応答からフィールドと文書IDを取得、添付ファイル(画像)をローカルに保存
if data_response.status_code == 200:
data = data_response.json()
print(data['Field02'], data['@unid'])
images = extract_image_data_and_filename(data['ImageBody'])
for file_name, image_binary in images:
save_image_file(TEMP_PATH + file_name, image_binary)
print(f'Image saved to {file_name}')
else:
print('Access failed: ', data_response.status_code)
def main():
# セッション認証を行う
session, response = create_session()
# レスポンスコードから成否を判定
if response.status_code == 200:
doc_unid = '(文書ID)'
get_document(session, doc_unid)
else:
print('Login failed: ', response.status_code)
if __name__ == '__main__':
main()
Notes文書の新規作成(テキストのみ)－POSTコマンド
文書の新規登録を行います。
"""
Overview:
文書の新規作成(POST) - 添付ファイルなし
"""
import requests
# 定数宣言
HOST_URL = 'https://www.hogehoge.com/'
DB_PATH = 'hoge/hoge.nsf'
USER_NAME = '(任意)'
USER_PW = '(任意)'
def create_session():
"""
Summary:
セッション認証を行い、セッションとレスポンスを返す
Args:
(なし)
Returns:
session :セッション
response :レスポンス
"""
# ログインペイロードの作成
payload = {
'username': USER_NAME,
'password': USER_PW
}
# ログインリクエストを送信してセッションを確立
session = requests.Session()
response = session.post(HOST_URL + '?Login', data=payload)
return session, response
def create_document(arg_session):
"""
Summary:
新規文書を作成し、フォームを使って計算後、保存
Args:
arg_session :ログイン済みセッション
Returns:
(なし)
"""
# 文書を作成
form = 'frmCsvImport' # フォーム名
computewithform = 'true' # 保存時の文書計算
api_endpoint = f'{HOST_URL}{DB_PATH}/api/data/documents?form={form}&computewithform={computewithform}'
headers = {
'Content-Type': 'application/json'
}
payload = {
'Field01': '001',
'Field02': '値2',
'Field03': '値3',
'Field04': '値4',
'Field05': '値5'
}
try:
data_response = arg_session.post(
api_endpoint, headers=headers, json=payload)
if data_response.status_code in [200, 201]:
print('Success!:', data_response.status_code)
else:
print('Access failed: ', data_response.status_code)
except Exception as e:
print('An error occurred:', e)
def main():
# セッション認証を行う
session, response = create_session()
# レスポンスコードから成否を判定
if response.status_code == 200:
create_document(session)
else:
print('Login failed: ', response.status_code)
if __name__ == '__main__':
main()
Notes文書の新規作成(リッチテキストに画像あり)－POSTコマンド
文書の新規登録を行い、リッチテキストフィールドに指定した画像を添付します。画像のPOSTもBase64でエンコード・デコードするのですが、ヘッダーなどの指定が公式文書には見つからず、試行錯誤しました。
"""
Overview:
文書の新規作成(POST) - 添付ファイルあり
"""
import requests
import base64
import os
from email.header import Header
# 定数宣言
HOST_URL = 'https://www.hogehoge.com/'
DB_PATH = 'hoge/hoge.nsf'
USER_NAME = '(任意)'
USER_PW = '(任意)'
def create_session():
"""
Summary:
セッション認証を行い、セッションとレスポンスを返す
Args:
(なし)
Returns:
session :セッション
response :レスポンス
"""
# ログインペイロードの作成
payload = {
'username': USER_NAME,
'password': USER_PW
}
# ログインリクエストを送信してセッションを確立
session = requests.Session()
response = session.post(HOST_URL + '?Login', data=payload)
return session, response
def create_document_image(arg_session, arg_image_path):
"""
Summary:
新規文書を作成し、フォームを使って計算後、保存
Args:
arg_session :ログイン済みセッション
Returns:
(なし)
"""
# 文書を作成
form = 'frmCsvImport' # フォーム名
computewithform = 'true' # 保存時の文書計算
api_endpoint = f'{HOST_URL}{DB_PATH}/api/data/documents?form={form}&computewithform={computewithform}'
headers = {
'Content-Type': 'application/json'
}
# 画像ファイル読み込みとBase64エンコード
with open(arg_image_path, 'rb') as image_file:
image_data = image_file.read()
encoded_image = base64.b64encode(image_data).decode('utf-8')
file_name = os.path.basename(arg_image_path) # ファイル名取得
header = Header(file_name, 'iso-2022-jp') # 全角ファイル名エンコード
encoded_file_name = header.encode()
payload = {
'Field01': '001',
'Field02': '値2',
'Field03': '値3',
'Field04': '値4',
'Field05': '値5',
'ImageBody': {
'type': 'multipart',
'content': [
{
'contentType': 'image/jpeg',
'contentDisposition': f'attachment; filename=\"{encoded_file_name}\"',
'contentTransferEncoding': 'base64',
'data': encoded_image
}
]
}
}
try:
data_response = arg_session.post(
api_endpoint, headers=headers, json=payload)
if data_response.status_code in [200, 201]:
print('Success!:', data_response.status_code)
else:
print('Access failed:', data_response.status_code)
print(data_response.text)
except Exception as e:
print('An error occurred:', e)
def main():
# セッション認証を行う
session, response = create_session()
# レスポンスコードから成否を判定
if response.status_code == 200:
image_path = '(画像ファイルパス)'
create_document_image(session, image_path)
else:
print('Login failed:', response.status_code)
if __name__ == '__main__':
main()
Notes文書の更新(全体更新)－PUTコマンド
文書ID(UNID)で指定した文書の更新を行います。PUTコマンドでは、文書全体が更新されるため、JSONに含まれないフィールドは文書から削除されます。既存文書を更新するには用途を選ぶコマンドです。
"""
Overview:
文書の更新(PUT) - すべてのフィールドを更新
※ JSONで指定しなかったフィールドはフィールド自体が消える
"""
import requests
# 定数宣言
HOST_URL = 'https://www.hogehoge.com/'
DB_PATH = 'hoge/hoge.nsf'
USER_NAME = '(任意)'
USER_PW = '(任意)'
def create_session():
"""
Summary:
セッション認証を行い、セッションとレスポンスを返す
Args:
(なし)
Returns:
session :セッション
response :レスポンス
"""
# ログインペイロードの作成
payload = {
'username': USER_NAME,
'password': USER_PW
}
# ログインリクエストを送信してセッションを確立
session = requests.Session()
response = session.post(HOST_URL + '?Login', data=payload)
return session, response
def put_document(arg_session, arg_doc_unid):
"""
Summary:
文書を更新-JSONに含まれないフィールドは消える
Args:
arg_session :ログイン済みセッション
arg_doc_unid :文書ID
Returns:
(なし)
"""
# 文書を更新
api_endpoint = f'{HOST_URL}{DB_PATH}/api/data/documents/unid/{arg_doc_unid}'
headers = {
'Content-Type': 'application/json'
}
payload = {
'Field02': '値2 put',
'Field03': '値3 put',
'Field04': '値4 put',
'Field05': '値5 put'
}
try:
data_response = arg_session.put(
api_endpoint, headers=headers, json=payload)
if data_response.status_code in [200, 201]:
print('Success!:', data_response.status_code)
else:
print('Access failed: ', data_response.status_code)
except Exception as e:
print('An error occurred:', e)
def main():
# セッション認証を行う
session, response = create_session()
# レスポンスコードから成否を判定
if response.status_code == 200:
doc_unid = '(文書ID)'
put_document(session, doc_unid)
else:
print('Login failed:', response.status_code)
if __name__ == '__main__':
main()
Notes文書の更新(部分更新)－PATCHコマンド
文書ID(UNID)で指定した文書の更新を行います。PATCHコマンドは部分更新です。JSONに含まれないフィールドは保持されます。
"""
Overview:
文書の更新(PATCH) - 指定項目のみ更新
"""
import requests
# 定数宣言
HOST_URL = 'https://www.hogehoge.com/'
DB_PATH = 'hoge/hoge.nsf'
USER_NAME = '(任意)'
USER_PW = '(任意)'
def create_session():
"""
Summary:
セッション認証を行い、セッションとレスポンスを返す
Args:
(なし)
Returns:
session :セッション
response :レスポンス
"""
# ログインペイロードの作成
payload = {
'username': USER_NAME,
'password': USER_PW
}
# ログインリクエストを送信してセッションを確立
session = requests.Session()
response = session.post(HOST_URL + '?Login', data=payload)
return session, response
def patch_document(arg_session, arg_doc_unid):
"""
Summary:
文書を取得
Args:
arg_session :ログイン済みセッション
arg_doc_unid :文書ID
Returns:
(なし)
"""
# 文書を更新
api_endpoint = f'{HOST_URL}{DB_PATH}/api/data/documents/unid/{arg_doc_unid}'
headers = {
'Content-Type': 'application/json'
}
payload = {
'Field03': '値3 patch'
}
try:
data_response = arg_session.patch(
api_endpoint, headers=headers, json=payload)
if data_response.status_code in [200, 201]:
print('Success!:', data_response.status_code)
else:
print('Access failed: ', data_response.status_code)
except Exception as e:
print('An error occurred:', e)
def main():
# セッション認証を行う
session, response = create_session()
# レスポンスコードから成否を判定
if response.status_code == 200:
doc_unid = '(文書ID)'
patch_document(session, doc_unid)
else:
print('Login failed:', response.status_code)
if __name__ == '__main__':
main()
Notes文書の削除－DELETEコマンド
文書ID(UNID)で指定した文書の削除を行います。
"""
Overview:
文書の削除(DELETE)
"""
import requests
# 定数宣言
HOST_URL = 'https://www.hogehoge.com/'
DB_PATH = 'hoge/hoge.nsf'
USER_NAME = '(任意)'
USER_PW = '(任意)'
def create_session():
"""
Summary:
セッション認証を行い、セッションとレスポンスを返す
Args:
(なし)
Returns:
session :セッション
response :レスポンス
"""
# ログインペイロードの作成
payload = {
'username': USER_NAME,
'password': USER_PW
}
# ログインリクエストを送信してセッションを確立
session = requests.Session()
response = session.post(HOST_URL + '?Login', data=payload)
return session, response
def delete_document(arg_session, arg_doc_unid):
"""
Summary:
文書を削除
Args:
arg_session :ログイン済みセッション
arg_doc_unid :文書ID
Returns:
(なし)
"""
# 文書を取得
api_endpoint = f'{HOST_URL}{DB_PATH}/api/data/documents/unid/{arg_doc_unid}'
data_response = arg_session.delete(api_endpoint)
# 応答からDBタイトルを取得
if data_response.status_code == 200:
print('Success!:', data_response.status_code)
else:
print('Access failed:', data_response.status_code)
def main():
# セッション認証を行う
session, response = create_session()
# レスポンスコードから成否を判定
if response.status_code == 200:
doc_unid = '(文書ID)'
delete_document(session, doc_unid)
else:
print('Login failed:', response.status_code)
if __name__ == '__main__':
main()
最後に
現在はPython、TypeScript(JavaScript)全盛期なので、世の中のサンプルコードと言えばどちらかです。PythonはExcelファイルにアクセスするライブラリや、チャート描画ライブラリ、統計計算ライブラリなどフリーのライブラリ、サンプルコードがネット上にあふれています。既存のDominoサーバーを生かして、新しい機能の拡張、価値創造、自分のレベルアップをPythonで行ってみてはいかがですか？新しい世界が見えるかもしれません。
