【Software開発】python実行におけるエントリーポイントのおさらいについて

classファイルの実行について

　普段、基盤関連のシェルスクリプトやJupyterNotebookからインタラクティブにスクリプトを実行することが多く、関数型言語の実行に慣れていた。案件でオブジェクト指向で開発を実施して、スクリプト実行する機会があったので備忘として整理しておく。
インスタンス化やクラスなどの考え方は、前提にします。

簡単な処理の流れについて

下記のようなエントリーポイントを持つモジュールを作成します。
実行時は下記のように実行する。

python src.module.py

python module.py

# ─────────────────────────────────────────────────────
# 共通モジュールのインポートと初期設定
# ─────────────────────────────────────────────────────
import 必要なモジュール類

# 共通ユーティリティやクラスのインポート
import LoggingUtility as log
import ParamsLoader as params_loader
import BusinessDateUtility as date_util
import dir.BusinessLogic as logic

# ─────────────────────────────────────────────────────
# コントロールクラスの定義（処理全体の司令塔）
# ─────────────────────────────────────────────────────
class ControllerClass:
    def __init__(self, logger):
        self.logger = logger
        self.transaction_id = ""
        self.retry_count = 0

    @retry.retry(...)
    def main_process(self, blob_name, params, dates):
        session, self.retry_count, self.transaction_id, params = session_mgmt.create_session_and_transaction(...)
        try:
        logic_instance = logic.BusinessLogic(
            self.logger,
            session,
            current_time,
            business_date
            )
            logic_instance.main(name)
        except Exception as ex:
            session_mgmt.rollback(session)
            self.logger.error(f"エラー発生: {repr(ex)}")
            raise

# ─────────────────────────────────────────────────────
# メイン処理（コントローラーの利用）
# ─────────────────────────────────────────────────────
# 各種パラメータ・日付・ロガーの初期化
params = params_loader.load()
business_date = date_util.get_current_date()
logger = log.setup_logger(...)

# コントロールクラスのインスタンス化
controller = ControllerClass(logger)

try:
controller.main_process(file, params, business_date)
except 特定エラー:
    continue
except Exception as e:
        logger.error(f"処理中にエラー: {repr(e)}")

# ログのアップロードや後処理
log.upload_and_close(logger)

下記のように、dirの下に「BusinessLogic.py」というモジュールを置く。

# ─────────────────────────────────────────────────────
# 必要なモジュールのインポート（仮）
# ─────────────────────────────────────────────────────
import pandas as pd
from snowflake.snowpark.functions import col, lit, concat
# その他必要なモジュールを適宜インポート

# ─────────────────────────────────────────────────────
# BusinessLogicクラス定義（ビジネス処理層）
# ─────────────────────────────────────────────────────
class BusinessLogic:
    def __init__(self, logger, session, current_time, business_date):
        self.logger = logger
        self.session = session

        # 必要なリポジトリ・ユーティリティの初期化
        self.snowpark_repo = ...
        ...

    def 処理1 (self):
        ##　処理内容を記載

    def 処理2 (self):
        ##　処理内容を記載
        
    def main(self, blob_name):

処理の流れとしては下記の通りとなる。

処理の流れ

Control → BusinessLogic の実行フロー（概要）

① Pythonファイルの実行開始

たとえば main_controller.py を次のようにコマンドで実行
python main_controller.py

② モジュール・クラスの読み込み（import）

from dir.BusinessLogic import BusinessLogic
Python は dir/BusinessLogic.py を探し、その中に定義された BusinessLogic クラスをメモリに読み込みます（実行はされない）。

③ if name == "main" もしくは 末尾の実行ブロックが実行される

• 各種初期化
  params = ...
  logger = ...
  controller = ControllerClass(logger) ← ★ここでControllerClassをインスタンス化

• 処理実行
  controller.main_process(...) ← ★ここでmain_process関数が実行される

④ ControllerClass.main_process() が呼ばれる

def main_process(self, blob_name, params, dates):
session, self.retry_count, self.transaction_id, params = ...

# ロジック層のクラスをインスタンス化
logic_instance = BusinessLogic(...)  ← ★ここでBusinessLogicのインスタンスを生成

# ビジネスロジックのmain()を実行
logic_instance.main(blob_name)       ← ★main関数の処理開始

⑤ BusinessLogicクラスの__init__とmain()が実行される

class BusinessLogic:
def init(...): ← ★ここが呼ばれ、初期化される
...

def main(self, blob_name): ← ★ここがビジネス処理のエントリーポイント
    self.処理1()
    self.処理2()

処理の流れのまとめ

main_controller.py 実行
↓
import で BusinessLogic 読み込み
↓
ControllerClass インスタンス化
↓
main_process() 呼び出し
↓
BusinessLogic クラスを new で生成（__init__）
↓
BusinessLogic.main() 呼び出し
↓
必要に応じて 処理1(), 処理2() が呼ばれる

ちょっと疑問点を記載してくおく

import分について改めて整理

import分で作成したpythonモジュールに対して「as」などで短縮した別名をつけて読み込みを行い、その中で定義されているclassを読んでインスタん化する。

logic.BusinessLogic(self.logger, session, params, ...)

それ以外にfromを使用した外部モジュールの使用の仕方もあるので簡単に整理していく。
例えば下記などのものである。下記はもともとpythonの仮想環境を作成する際にsnowflake.snowparkのSDKを読み込んでいます。

from snowflake.snowpark.functions import col, lit, concat

実際には、自身のローカル環境の下記のディレクトリに

${HOME}/miniforge3/envs/my_env/lib/python3.10/site-packages/snowflake/snowpark/

下記のfunction.pyが存在している。

ls -ltr functions.py

そして、下記のようにlitの関数が定義されていることがわかる。

python function.py

def lit(literal: LiteralType) -> Column:
    """
    Creates a :class:`~snowflake.snowpark.Column` expression for a literal value.
    It supports basic Python data types, including ``int``, ``float``, ``str``,
    ``bool``, ``bytes``, ``bytearray``, ``datetime.time``, ``datetime.date``,
    ``datetime.datetime`` and ``decimal.Decimal``. Also, it supports Python structured data types,
    including ``list``, ``tuple`` and ``dict``, but this container must
    be JSON serializable.
    
    Example::
    
        >>> import datetime
        >>> columns = [lit(1), lit("1"), lit(1.0), lit(True), lit(b'snow'), lit(datetime.date(2023, 2, 2)), lit([1, 2]), lit({"snow": "flake"})]
        >>> session.create_dataframe([[]]).select([c.as_(str(i)) for i, c in enumerate(columns)]).show()
        ---------------------------------------------------------------------------------------
        |"0"  |"1"  |"2"  |"3"   |"4"                 |"5"         |"6"   |"7"                |
        ---------------------------------------------------------------------------------------
        |1    |1    |1.0  |True  |bytearray(b'snow')  |2023-02-02  |[     |{                  |
        |     |     |     |      |                    |            |  1,  |  "snow": "flake"  |
        |     |     |     |      |                    |            |  2   |}                  |
/lit

つまり下記では、fromによって使用するpythonモジュールまで読み込み、その中で定義されている関数やクラスをimportして、そのまま使用できるようにしているわけである。

python.py

from snowflake.snowpark.functions import col, lit, concat

まどろっこしいですが、importを使用して書くこともできて下記のように
function.pyを読み込んで(別名をFとする)、その中のcol関数を呼ぶということも可能である。この場合は名前空間(どのパッケージやモジュールに属する関数やクラスなのか)をわかりやすくできるメリットもある。

python.py

import snowflake.snowpark.functions as F
・・・
F.col()

def init(self,xxx )について

　よくpythonの本にも記載されているが

python.py

def __init__(self,xxxx)

は、コンストラクタでありクラスがインスタンス化された際に呼ばれ初期化も処理に使用される。
なお、関数の定義側にselfがあるがこれは自身のインスタンスを指しているものであり、関数の定義側には必須である。
また、インスタンス化するときにclassの引数にあたる

python.py

logic_instance = logic.BusinessLogic(self.logger, session, current_time, business_date)

と、コンストラクタで定義する__init__の引数の対応は揃える必要があることに注意する。__init__側にはselfをつけることに留意しておこう。

python.py

__init__(self, logger, session, current_time, business_date):

関数型のようにpythonファイルを作成することももちろん可能

　下記のようにシェルスクリプトのように関数型に記載して上から下に実行することも可能である。

python test1.py

def main():
    print("これはメイン処理です")

print("Hello")
main()

この場合は、実行するとまず実行ブロックのprintが実行されてmain関数が呼ばれる。

(my_env)% python test.py
Hello
これはメイン処理です

下記の場合も同じである。printが呼ばれて「__name__のif文」が呼ばれてその下のprintが呼ばれる。シェルと同じように関数型処理のように実施されるわけである。

python test2.py

def main():
    print("メイン処理")

print('hello1')
# 推奨される書き方（汎用性が高い）
if __name__ == "__main__":
    main()
print('hello2')