🌏

Python、Streamlit と FastAPI で、ブロックチェーンを実装してみた♪③トランザクションを作成

2025/08/01に公開

今回は、いよいよトランザクションを作成し、プールに保存していきます！

 前回までの記事前々回

https://zenn.dev/animalz/articles/896225dc4b6bef

前回

https://zenn.dev/animalz/articles/2946d829db92c9

 環境Windows10、PyCharam、Python v3.13.2、Streamlit v1.45.1、FastAPI v0.115.12

 ディレクトリ構成（node2 と　node3 は、テスト用ファイルとフォルダ以外同じなので省略してます）
 ディレクトリ構成
streamlit_fastapi
├ node1
| ├ json（JSON データ保存用）
| ├ test_data（テスト･データ保存用）
| ├ app.py
| ├ block_chain.log（ログ保存用）
| ├ log.py
| ├ main.py
| ├ mikoto.py
| ├ test_app_main.py
| └ test_mikoto.py
...

 今回やることまず、前回忘れてた log.py に型ヒントを追加します。

それから、トランザクションを作成（発行、送信）し、ブロードキャストして、検証後、プールに保存します。

 log.py に型ヒントを追加 log.py
...
def get_logger(name: str) -> logging.Logger:
    return logging.getLogger(name)

def basic_config(level: str, filename: str = None) -> None:
    ...

def log_debug(logger: logging.Logger, message: str) -> None:
    return logger.debug(message)

def log_error(logger: logging.Logger, message: str) -> None:
    return logger.error(message)

 トランザクションを作成（発行、送信）まず、トランザクションを作成する関数を実装してテストし、その後、フロントエンド、バックエンドを実装していきます。

mikoto.py、test_mikoto.py、app.py、main.py の順にファイルを作成します。
 mikoto.py
...
def make_thanks_transaction(
        sender_secret_key_str: str, sender_public_key_str: str,
        receiver_public_key_str: str, mik: int) -> dict:
    thanks_transaction = {
        'time': dt.datetime.now().isoformat(),
        'sender': sender_public_key_str,
        'receiver': receiver_public_key_str,
        'MIK': mik,
    }
    signature = make_signature_str(
        thanks_transaction, sender_secret_key_str)
    thanks_transaction['signature'] = signature
    return thanks_transaction

def make_mikoto_transaction(receiver_public_key_str: str, mik: int) -> dict:
    mikoto_transaction = {
        'time': dt.datetime.now().isoformat(),
        'sender': 'mikoto_project',
        'receiver': receiver_public_key_str,
        'MIK': mik,
        'signature': 'mikoto_project',
    }
    return mikoto_transaction
...
テストフィアルを作成します。
 test_mikoto.py
...
def test_make_thanks_transaction():
    """ test: type, keys, conversion, value, unhexlify """
    thanks_transaction = mik.make_thanks_transaction(
        co_secret_key_str, co_public_key_str,
        co2_public_key_str, 100
    )
    assert isinstance(thanks_transaction, dict)
    keys = list(thanks_transaction.keys())
    assert keys == ['time', 'sender', 'receiver', 'MIK', 'signature']
    assert dt.datetime.fromisoformat(thanks_transaction['time'])
    assert thanks_transaction['sender'] == co_public_key_str
    assert thanks_transaction['receiver'] == co2_public_key_str
    assert isinstance(thanks_transaction['MIK'], int)
    assert isinstance(thanks_transaction['signature'], str)
    assert binascii.unhexlify(thanks_transaction['signature'])

def test_make_mikoto_trasaction():
    """ test: type, keys, value """
    mikoto_transaction = mik.make_mikoto_transaction(
        co_public_key_str, 100
    )
    assert isinstance(mikoto_transaction, dict)
    assert list(mikoto_transaction.keys()) == ['time', 'sender', 'receiver', 'MIK', 'signature']
    assert dt.datetime.fromisoformat(mikoto_transaction['time'])
    assert mikoto_transaction['sender'] == 'mikoto_project'
    assert mikoto_transaction['receiver'] == co_public_key_str
    assert isinstance(mikoto_transaction['MIK'], int)
    assert mikoto_transaction['signature'] == 'mikoto_project'
...
テストを実行、無事にパスできました。

フロントエンドを実装します。

ログイン後の実装になります。送る側は、本人か mikoto_project（発行）、送信時には本人には送信できないようにしています。

if 文が多いので、コードはやや見にくいですが、UI の使いやすさ優先で実装しました。
 app.py
...
    if ss.state is True:
        st.markdown('ログイン中')
        menu = ['発行または送信']
        choice = st.selectbox(label='メニュー', options=menu)
        key_data_list = mik.load_json('json/key_data_list.json')
        my_data = mik.load_json('json/.my_data.json')
        if choice == '発行または送信':
            # select に変更（my_name or mikoto_project）
            sender_name = st.selectbox('送る側（mikoto_projectは発行）', [my_data['name'], 'mikoto_project'])

            receiver_name = st.text_input('受け取り側')
            receiver_data = None
            if receiver_name:
                if receiver_name == my_data['name']:
                    st.markdown('自分には送信できません')
                else:
                    for key_data in key_data_list:
                        if key_data['name'] == receiver_name:
                            receiver_data = key_data
                            st.write(receiver_data)
                    if not receiver_data:
                        st.markdown('登録されていないニックネームです')

            mik_value = st.number_input('MIKの量', min_value=0, step=1)

            if receiver_data and mik_value:
                if st.button('実行する'):
                    if sender_name == 'mikoto_project':
                        transaction = mik.make_mikoto_transaction(
                            receiver_data['public_key_str'],
                            mik_value
                        )
                        st.json(transaction)
                    else:
                        transaction = mik.make_thanks_transaction(
                            my_data['secret_key_str'],
                            my_data['public_key_str'],
                            receiver_data['public_key_str'],
                            mik_value
                        )
                        st.json(transaction)

                    url_list = mik.get_url_list()
                    for url in url_list:
                        try:
                            res = mik.post_data(url+'/transaction', transaction)
                            st.json(res.json())
                            log.log_debug(logger, f'{res}: {res.json()}')
                        except:
                            st.json({"message": f"{url}: error"})
                            log.log_error(logger, f"{url}: error")

    else:
        st.markdown('ログアウト状態')
バックエンド側を実装します。発行の場合は、検証しない仕様にしています。
 main.py
...
class Transaction(BaseModel):
    time: str
    sender: str
    receiver: str
    MIK: int
    signature: str
...
@app.post('/transaction')
async def post_transaction(transaction: Transaction):
    transaction = dict(transaction)
    if transaction['sender'] != 'mikoto_project':
        if not mik.verify_data(transaction, transaction['sender']):
            log.log_error(logger, 'transaction invalid')
            return {'message': 'transaction invalid'}
    transaction_pool = mik.load_json('json/transaction_pool.json')
    transaction_pool.append(transaction)
    mik.save_json(transaction_pool, 'json/transaction_pool.json')
    log.log_debug(logger, 'transaction received')
    return {'message': 'transaction received'}
...
動作確認します。

ログ･ファイルを削除し、変更したファイルは、node2、node3 についても更新します。

各サーバーを立ち上げます。（計6つ）
 コマンド・プロンプト
streamlit run app.py --server.port 8501
streamlit run app.py --server.port 8502
streamlit run app.py --server.port 8503
uvicorn main:app --reload --port 8010
uvicorn main:app --reload --port 8011
uvicorn main:app --reload --port 8012
各ノードが 100MIK 自分に発行します（各サーバーから）

animal のサーバー（http://127.0.0.1:8501）

実行（発行）すると、無事に発行できました。



json/transaction_pool.json



block_chain.log



（内容はほぼ同じなので、以降の画像は省略します）
各ノードが他ノードに送信します。

animal(node1) -> dog(node2): 50MIK

dog(node2) -> cat(node3): 30MIK

cat(node3) -> animal(node1): 30MIK

すべて、無事に発行、送信できました

json/transaction_pool.json

（発行）



（送信）



block_chain.log（node1）


 参考文献などSatoshi Nakamoto (2008), ビットコイン：P2P 電子通貨システム,

https://bitcoin.org/files/bitcoin-paper/bitcoin_jp.pdf, 論文

モヤっとデータサイエンティスト (2022), 『Python によるブロックチェーン開発教本』, 電子書籍

安田恒 (2023), 『ブロックチェーンを作る！』, 秀和システム, 書籍

 今回のコードGitHub: https://github.com/Animalyzm/mikoto_project

今回のコードは、block_chain/streamlit_fastapi です。

Git のコミット･メッセージは、block_chain_streamlit_fastapi_3_transaction です。
今回は以上となります、ありがとうございましたー♪