目次

こんなタスクがありました
 pythonで自動化しよう
 下準備
 コード
 まとめ

こんなタスクがありました

もらったエクセルを元にしてDBにデータを追加するお仕事。
エクセルからコピペでSqlServerにデータを追加するという超アナログタスク。
コピペするにはDBのテーブルカラムと列を合わせる必要があるので
新しいエクセルシートを作成して列毎に情報を一つ一つセルに書き込む下準備が必要。
その上、定期的に仕事がやってくるのでメンドくさいorz

pythonで自動化しよう

手動でエクセルを用意してデータ追加するのではなく、プログラムにやらせよう。
作業時間が大幅に短縮されるし、コピペミスも無くなる！

ということで openpyxl, sqlalchemy を使ってタスクの自動化をしていこう。
※openpyxlとは、PythonでExcelファイルを作成できる便利なツールである

下準備

もらったエクセル（例えばdata.xlsx）をデスクトップに置こう。
次に、デスクトップに auto_insert.py ファイルを作成しよう。
ファイルの中身に以下のコードを書いて保存しよう。

auto_insert.py

print("Hello, World!")

次に、ターミナルを開いて「デスクトップ」ディレクトリに移動しよう。コマンドはこんな感じ。

~ $ cd Desktop

移動ができたら、以下を実行しよう！

~/Desktop $ python auto_insert.py

成功すればターミナルに Hello, World! が表示されるよ。
※エラーが出たりわからない人はChatGPTに質問しよう

コード

それではauto_insert.py, models.py, database.pyの３つのファイルをデスクトップに作成しよう。
それぞれのファイルの中身は以下を書いてくれ。

auto_insert.py

import openpyxl
from sqlalchemy import insert
from core.database import SessionLocal
from core.models import Fruits


# もらったExcelファイル(data.xlsx)の最初のシートを読み込む
input_excel_path = 'data.xlsx'
input_wb = openpyxl.load_workbook(input_excel_path)
input_sheet = input_wb.active

# DBにデータを追加する
fruits: List[Any] = []
# もらったExcelファイルの１行目にヘッダーがある場合を想定してmin_row=2にする
for row in input_sheet.iter_rows(min_row=2, values_only=True):
    insert_data = {
        "id": row[0],
        "name": row[1],
        "price": row[2],
        "pieces": row[3],
    }
    fruits.append(insert_data)

# バルクインサート
db = SessionLocal()
try:
    db.execute(insert(Fruits), fruits)
    db.commit()
except Exception as e:
    db.rollback()
finally:
    db.close()

models.py

from sqlalchemy import VARCHAR, Integer
from sqlalchemy.orm import Mapped, mapped_column

from core.database import Base


class Fruits(Base):
    __tablename__ = "FRUITS_TBL"
    __table_args__ = {"extend_existing": True}

    id: Mapped[int] = mapped_column("ID", Integer, primary_key=True)
    name: Mapped[str] = mapped_column("NAME", VARCHAR(50))
    price: Mapped[str] = mapped_column("PRICE", VARCHAR(50))
    pieces: Mapped[str] = mapped_column("PIECES", VARCHAR(50))

database.py

from sqlalchemy import create_engine
from sqlalchemy.engine import URL
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker


connection_url = URL.create(
    "mssql+pyodbc",
    username="DBのユーザ名",
    password="DBのパスワード",
    host="DBのアドレス",
    port="ポート",
    database="データベース名",
    query={
        "driver": "ODBC Driver 17 for SQL Server",
        "autocommit": "True",
        "LongAsMax": "Yes",
    },
)
engine = create_engine(
    connection_url,
    echo=True,
    connect_args={"TrustServerCertificate": "yes"},
    pool_pre_ping=True,
    pool_recycle=3600,
    pool_timeout=30,
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


Base = declarative_base()

準備ができたらもう一度以下を実行してみよう。DBにデータが追加されれば無事成功だ！

~/Desktop $ python auto_insert.py

※openpyxl, sqlalchemyが見つからないエラーが出てきたら pip install でインストールしてあげよう

まとめ

アナログタスクを自動化したおかげで生産性が爆上がりした。
DBへのデータ追加に2時間かかっていた(データが1万件ぐらいあった)のが1分でできるようになったよ。