【Python】固定長データを区切りSQLServerにインポートする

きっかけ

当方、趣味と実益を兼ねて、ボートレース予想支援アプリを作ろうと思っています。
ボートレース公式サイトに、ボートレーサーの半年ごとの成績データが固定長データで公開されています。
ボートレーサーの成績データのダウンロード元 | BOAT RACE オフィシャルウェブサイト
これを、扱いやすいように、SQLServerに移行しようと思いました。

環境

• python3.9
• Windows10
• VSCode
• SQLServer2016Express

作るもの

固定長データのレイアウト、SQLServerにおけるテーブル名、型を格納するJSON(以下、型定義JSON)

固定長データの何桁目から何桁目は選手名、保存先テーブル名等を表すJSONを作成します。

例で用いている固定長データのレイアウトはボートレース公式を参照しました。
「モーターボートファン手帳」のレイアウト | BOAT RACE オフィシャルウェブサイト

以下の形式で定義しました。

フィールド名	内容	値例
dbo	データベース	"BoatRace"
table_name	テーブル名	"M_ファン手帳"
fields	下記列情報群の配列	[{field_name: "登番", "start_index"...}, {field_name: "名前漢字", ...}, ...]

列情報群

フィールド名	内容	値例
field_name	列名	"登番"
start_index	固定長データにおける開始位置	0
end_index	固定長データにおける終了位置	4
mssql_datatype	SQLServerでのデータ型(smallint, nvarchar等)	"smallint"
is_primary_key	主キーか否か(主キー:1, 主キーでない:0)	1

データ例

{
  "dbo": "BoatRace", 
  "table_name": "M_ファン手帳", 
  "fields": [
    {"field_name": "登番", "start_index": 0, "end_index": 4, "mssql_datatype": "smallint", "is_primary_key": 1}, 
    {"field_name": "名前漢字", "start_index": 4, "end_index": 12, "mssql_datatype": "nvarchar", "is_primary_key": 0}, 
    {"field_name": "名前カナ", "start_index": 12, "end_index": 27, "mssql_datatype": "nvarchar", "is_primary_key": 0}, 
    {"field_name": "支部", "start_index": 27, "end_index": 29, "mssql_datatype": "nvarchar", "is_primary_key": 0}, 
    {"field_name": "級", "start_index": 29, "end_index": 31, "mssql_datatype": "nvarchar", "is_primary_key": 0}, 
    {"field_name": "年号", "start_index": 31, "end_index": 32, "mssql_datatype": "nvarchar", "is_primary_key": 0}, 
    {"field_name": "生年月日", "start_index": 32, "end_index": 38, "mssql_datatype": "nvarchar", "is_primary_key": 0}, 
    {"field_name": "性別", "start_index": 38, "end_index": 39, "mssql_datatype": "smallint", "is_primary_key": 0}, 
    {"field_name": "年齢", "start_index": 39, "end_index": 41, "mssql_datatype": "smallint", "is_primary_key": 0}, 
    {"field_name": "身長", "start_index": 41, "end_index": 44, "mssql_datatype": "smallint", "is_primary_key": 0}, 
    {"field_name": "体重", "start_index": 44, "end_index": 46, "mssql_datatype": "smallint", "is_primary_key": 0}, 
    ...
}

このファイルを元に、保存先テーブルの作成、固定長データからSQLServerへのエクスポートを行います。

SQLServerにテーブルを作成するクラス DDLController

上記で作成した型定義JSONをから、SQLServerにテーブルを作成するクラス「DDLContrller」を以下のように設計しました。

メンバ説明

• create_table_command

  文字列で下記のようなテーブルを作成するコマンドを保持します。

create table テーブル名(
    登番 smallint , 
    ..., 
    primary key (登番, 年, 期)
)

• drop_table_command
  テーブルを削除するコマンドを保持します。

• connection
  SQLServerへのConnectionオブジェクトを保持します。
  pyodbc.connectionクラスのインスタンスです。

• structrual_settings
  型定義を保持する辞書です。型定義JSONをパースした値です。

メソッド説明

• __init__(connection, structual_settings_path)

  • connecton
    SQLServerへの接続情報、pyodbc.connectionのインスタンスを渡します。
  • structual_settings_path
    型定義JSONのパスを渡します。

• create_table()

  SQLServerにcreate tableコマンドを実行します。

• drop_table()

  SQLServerにdrop tableコマンドを実行します。

• __create_create_table_command()

  型定義JSONからcreate tableコマンドを生成するヘルパー関数です。

コード

import pyodbc
import json

class DDLController:
    def __init__(self, connection, structual_settings_path):
        self.create_table_command = ""
        self.connectoin = connection
        self.structual_settings = {}

        # jsonファイルをリード
        with open(structual_settings_path, 'r', encoding="utf-8") as jfile:
            self.structual_settings = json.load(jfile)

        # コマンドを生成する
        self.create_table_command = self.__create_create_table_command()
        self.drop_table_command = "drop table [" + self.structual_settings["table_name"] + "]"


    def __create_command(self):
        # フィールド定義部分を生成する
        fields = ""
        primary_key_fields = ""

        for field in self.structual_settings["fields"]:
            if fields:
                fields += ", \n"
            
            field_name = field["field_name"]
            data_type = field["mssql_datatype"]
            size = ""
            
            # データ型がchar, varchar, nvarcharの時、フィールドサイズを取得する。
            if "char" in data_type:
                size = str(field["end_index"] - field["start_index"])
            
            fields += "[" + field_name + "] " + data_type
            if size:
                fields += "(" + size + ")"
            
            if field["is_primary_key"]:
                if primary_key_fields:
                    primary_key_fields += ", "
                
                primary_key_fields += "[" + field_name + "]"
        
        command = "create table [" + self.structual_settings["table_name"] + "] (" + fields
        if primary_key_fields:
            command += ", \n primary key (" + primary_key_fields + ")"
        command += ")"

        return command
    
    def create_table(self):
        command = "use [" + self.structual_settings["dbo"] + "]\n"
        command += self.create_table_command
        
        cursor = self.connectoin.cursor()
        cursor.execute(command)
        cursor.commit()

    def drop_table(self):
        command = "use [" + self.structual_settings["dbo"] + "]\n"
        command += self.drop_table_command

        cursor = self.connectoin.cursor()
        cursor.execute(command)
        cursor.commit()

• テスト

  テーブルが作成されるかテストします。
  型定義ファイルは同階層に配置されており、ファイル名は"fannotebookstructure.json"とします。

if __name__ == "__main__":
    # connectionを生成する。
    server = r"(local)\SQLExpress"
    user = "sa"
    password = ""
    db = "CCSDB"

    constr = "DRIVER={SQL Server};SERVER=" + server + ";uid=" + user + ";pwd=" + password + ";DATABASE=" + db

    conn =  pyodbc.connect(constr)
    
    # テーブルを生成する。
    ddlcontroller = DDLController(conn, "./fannotebookstructure.json")

固定長データをリードするクラス FixedDataAnalyzer

型定義JSONと読み込み元固定長データからデータをリードするクラス、FixedDataAnalyzerを以下のように定義しました。
実際にSQLServerにインポートする処理を作成する前に、固定長データをリードできるかテストするために作成しました。

メンバ説明

• base_file_path

  読み込み元固定長データのパスを保持します。

• structual_settings

  型定義を保持する辞書です。型定義JSONをパースした値です。

メソッド説明

• __init__(base_file_path, structual_settings_path)

  • base_file_path
    読み込み元固定長データのパスを渡します。
  • structual_settings_path
    型定義JSONのパスを渡します。

• execute()

  固定長データをリードし、print関数にて表示します。

コード

import json
import pyodbc # テスト時SQLServerに接続するので、インポートする。

class FixedDataAnalyzer:
    
    def __init__(self, base_file_path, structual_settings_path):
        # 読み込み元固定長データのパス
        self.base_file_path = base_file_path
        
        # 設定ファイルを開き設定値をロードする。
        self.structual_settings = {}
        with open(structual_settings_path, 'r', encoding="utf-8") as json_file:
            self.structual_settings = json.load(json_file)
    
    def execute(self):
        # 固定長データをオープン
        with open(self.base_file_path, 'r') as f:
            # 1行ずつ巡回する。
            for line in f.readlines():
                str = ""
                
                # 各列を巡回する。
                for field in self.structual_settings['fields']:
                    if str:
                        str += "\t"
                    str += "{0}:{1}".format(field['field_name'], line[field['start_index']:field['end_index']])
                print(str)
    ```
    
- テスト

    固定長データをリードできるかテストします。
    型定義ファイルは同階層に配置されており、ファイル名は"fann2010.txt"とします。
```python
if __name__ == "__main__":
    fixedDataAnalyzer = FixedDataAnalyzer("./fan2010.txt", "./fannotebookstructure.json")
    fixedDataAnalyzer.execute()
    print("完了")

実行時の図

固定長データをリードしSQLServerにInsertするクラス FixedDataAnalyzerSQL

FixedDataAnalyzerを継承し、SQLServerにインポートするクラスを以下のように定義しました。
※少々、無理やり継承を使ってみた、という感があります。

メソッド説明

• execute(connection)
  • connection
    SQLServerへの接続情報、pyodbc.connectionのインスタンスを渡します。
    固定長データ1行ごとにINSERT INTO文を発行し、SQLServerの対象テーブルにレコードを追加します。
• __create_insert_command(line)
  • line
    1行当たりの固定長データを渡します。
    渡された固定長データから、INSERT INTO文を生成し、返します。

コード

import json
import pyodbc # テスト時SQLServerに接続するので、インポートする。

class FixedDataAnalyzerSQL(FixedDataAnalyzer):
    def __create_insert_command(self, line):
        fields = ""
        values = ""

        for col in self.structual_settings["fields"]:
            start_index = col["start_index"]
            end_index = col["end_index"]

            if fields:
                fields += ", "
                values += ", "

            fields += "[" + col["field_name"] + "]"
            
            if "char" in col["mssql_datatype"] or "date" in col["mssql_datatype"] or "time" in col["mssql_datatype"]:
                values += "'" + line[start_index:end_index] + "'"
            else:
                values += line[start_index:end_index]
            
        table_name = self.structual_settings["table_name"]
        dbo = self.structual_settings["dbo"]

        return "INSERT INTO [" + dbo + "].dbo.[" + table_name + "] (" + fields +") VALUES (" + values + ")"


    def execute(self, connection):
        #cursorを取得する。
        cursor = connection.cursor()

        with open(self.base_file_path, 'r') as f:
            for line in f.readlines():
                #挿入コマンドを取得する。
                sql = self.__create_insert_command(line)
                #コマンドを実行する。
                print("実行中:{0}".format(sql))
                cursor.execute(sql)
                cursor.commit()

• 実行

  型定義ファイルは同階層に配置されており、ファイル名は"fann2010.txt"とします。

if __name__ == "__main__":
    # connectionを生成する。
    server = r"(local)\SQLExpress"
    user = "sa"
    password = ""
    db = "CCSDB"

    constr = "DRIVER={SQL Server};SERVER=" + server + ";uid=" + user + ";pwd=" + password + ";DATABASE=" + db

    conn =  pyodbc.connect(constr)
    
    fixedDataAnalyzer = FixedDataAnalyzerSQL("./fan2010.txt", "./fannotebookstructure.json")
    fixedDataAnalyzer.execute(conn)

実行後、SQLServerManagementStudioでテーブルを確認した図

無事、ファン手帳データを取り込むことができました。

感想

クラスの設計方法や、クラス名の付け方、難しいです。失敗しながら慣れていこうと思います。
クラス名は、google翻訳を使用して得た英語をもとに作成しました。なんとなく、実態より大げさな名前になり、なんだか照れます。
当記事では、ボートレースのファン手帳データを取り込みました。ほかにSQLServerに取り込みたい固定長データがあるとき、型定義JSONファイルを作成し、それをもとに取り込むことができると思います。

参考にさせていただいた記事、サイト

当記事は下記記事、サイトを参考にさせていただきました。

• Python クラスについて - Qiita
• 固定長のファイルをpythonでごりごりCSV化した話。 - Qiita
• pyodbcを使ってPythonでDBを操作しよう　～SQL Server編～ - GIS奮闘記
  ありがとうございました。