Pythonフレームワーク「Flask」を利用したアプリでpyodbcを利用してSQL Serverに接続してみた

pyodbcパッケージを利用すると、データベースに接続することができる。

今回は、Pythonフレームワーク「Flask」を利用したアプリで、pyodbcを利用してSQL Serverに接続してみたので、そのサンプルプログラムを共有する。

前提条件

下記記事のpytestによる単体テストの実装が完了していること。

Python単体テスト用フレームワーク「pytest」を利用して単体テストを実装してみたpytestとは、Python用に設計された単体テスト用のフレームワーク（テスティングフレームワーク）で、Flaskを用いたWebアプリ...

pyodbcのインストール

2024年12月12日現在の最新バージョンのpyodbc(5.2.0)をインストールする。その手順は、以下の通り。

1) コマンドプロンプトで「pip install pyodbc」コマンドを実行し、最新バージョンのpytestをインストールする。

2)「pip list | find “pyodbc”」コマンドを実行すると、以下のように、インストールしたpyodbcが確認できる。

サンプルプログラムの作成

作成したサンプルプログラムの構成は、以下の通り。

なお、上記の赤枠は、前提条件のプログラムから追加・変更したプログラムである。

views.pyの内容は以下の通りで、確認画面で送信ボタンを押下した場合に、登録処理を実行するようにしている。

from demo import app
from flask import render_template, request
from .checks import CheckInputForm
from .regist import RegistInputForm

# 初期表示処理
@app.route("/")
def index():
    return render_template("input.html")

# 入力画面_確認ボタン押下処理
@app.route("/confirm", methods=["POST"])
def confirm():
    # 入力画面の項目チェックでエラーになった場合は、入力画面に遷移する
    chk_rslt = CheckInputForm(request.form).check()
    if not chk_rslt:
        return render_template("input.html", form_data=request.form)
    # 入力画面の項目チェックで正常な場合は、確認画面に遷移する
    return render_template("confirm.html", form_data=request.form)

# 確認画面_送信ボタン押下処理
@app.route("/send", methods=["POST"])
def send():
    # 登録処理を呼び出し、エラーが有る場合は、画面遷移しない
    has_error = RegistInputForm(request.form).regist()
    if has_error:
        return render_template("confirm.html", form_data=request.form)
    # 登録処理を呼び出し、エラーが無い場合は、完了画面に遷移する
    return render_template("complete.html")

# 確認画面_戻るボタン押下処理
@app.route("/back", methods=["POST"])
def back():
    return render_template("input.html", form_data=request.form)

また、regist.pyの内容は以下の通りで、USER_DATAテーブルからid最大値を取得すると共に、USER_DATAテーブルにデータ登録を行うようにしている。また、登録エラー時はエラーメッセージを設定するようにしている。

from flask import flash
import pyodbc

class RegistInputForm:
    # SQL Server接続情報
    DRIVER_NAME = '{ODBC Driver 17 for SQL Server}'
    SERVER_NAME = 'localhost'
    DATABASE_NAME = 'master'
    USER_ID = 'USER01'
    PASSWORD = 'USER01'

    # 初期化処理
    def __init__(self, form):
        self.form = form

    # 登録処理
    def regist(self):
        # エラーの有無を設定する
        has_error = False
        # 各項目値を取得
        name = self.form.get('name')
        birth_year = self.form.get('birthYear')
        birth_month = self.form.get('birthMonth')
        birth_day = self.form.get('birthDay')
        sex = self.form.get('sex')
        memo = self.form.get('memo')
        # SQL Serverに接続
        conn = pyodbc.connect('DRIVER=' + self.DRIVER_NAME 
                   + ';SERVER=' + self.SERVER_NAME
                   + ';DATABASE=' + self.DATABASE_NAME + ';UID=' + self.USER_ID
                   + ';PWD=' + self.PASSWORD)
        conn.autocommit = False
        cursor = conn.cursor()
        try:
            # 登録済のid最大値を取得し設定
            max_id = 0
            cursor.execute('SELECT IsNull(MAX(id), 0) FROM dbo.USER_DATA')
            for row in cursor:
                max_id = row[0]
            # 登録処理
            cursor.execute("INSERT INTO dbo.USER_DATA ("
                 + " id, name, birth_year, birth_month, birth_day, sex, memo "
                 + ") VALUES (" + str(int(max_id) + 1) + ",N'" + name + "'," + birth_year
                 + "," + birth_month + "," + birth_day + ",'" + sex + "',N'" + memo + "')")
            conn.commit()
        except pyodbc.Error as ex:
            print(ex)
            flash('予期せぬエラーが発生しました、管理者に連絡してください。')
            has_error = True
            conn.rollback()
        # データベース接続を終了する
        cursor.close()
        conn.close()
        return has_error

なお、上記ソースコードの「ODBC Driver 17 for SQL Server」は、以下赤枠のODBCドライバーから確認できる。

さらに、確認画面の内容は以下の通りで、登録エラー時のメッセージを画面表示するようにしている。

<!DOCTYPE html>
<html lang="ja">
<head>
  <meta charset="UTF-8">
  <link rel="stylesheet" href="../static/demo.css" type="text/css">
  <script src="../static/demo.js"></script>
  <title>確認画面</title>
</head>
<body>
  入力内容を確認し、問題なければ「送信」ボタンを押下してください。
  <br/><br/>
  <!-- エラーメッセージを表示 -->
  {% with messages = get_flashed_messages() %}
  {% if messages %}
  <ul>
    {% for message in messages %}
    <li class="errorMessage">{{ message }}</li>
    {% endfor %}
  </ul>
  {% endif %}
  {% endwith %}
  <form>
    <table border="0">
      <tr>
        <td align="left" valign="top">名前： </td>
        <td>{{ form_data.name }}</td>
      </tr>
      <tr>
        <td align="left" valign="top">生年月日： </td>
        <td>
          {% if form_data.birthYear != '' and form_data.birthMonth != '' and form_data.birthDay != '' %}
            {{ form_data.birthYear }}年{{ form_data.birthMonth }}月{{ form_data.birthDay }}日
          {% else %}
            不明
          {% endif %}
        </td>
      </tr>
      <tr>
        <td align="left" valign="top">性別： </td>
        <td>
          {% if form_data.sex == '1' %}
            男
          {% elif form_data.sex == '2' %}
            女
          {% else %}
            不明
          {% endif %}
        </td>
      </tr>
      <tr>
        <td align="left" valign="top">メモ： </td>
        <td>
          <!-- テキストエリアの改行コードを変換し、改行された状態で表示 -->
          {{ form_data.memo.replace("\r\n", "<br>") | safe }}
        </td>
      </tr>
      <tr>
        <td align="left" valign="top">確認チェック： </td>
        <td>
          {% if form_data.checked == 'on' %}
            確認済
          {% else %}
            未確認
          {% endif %}
        </td>
      </tr>
    </table>
    <br/><br/>
    <input type="button" value="送信" onclick="send()" />　
    <input type="button" value="戻る" onclick="back()" />
    <!-- 次画面で利用できるよう、hidden属性で、入力画面の各項目値を送信 -->
    <input type="hidden" id="name" name="name" value="{{ form_data.name }}" />
    <input type="hidden" id="birthYear" name="birthYear" value="{{ form_data.birthYear }}" />
    <input type="hidden" id="birthMonth" name="birthMonth" value="{{ form_data.birthMonth }}" />
    <input type="hidden" id="birthDay" name="birthDay" value="{{ form_data.birthDay }}" />
    <input type="hidden" id="sex" name="sex" value="{{ form_data.sex }}" />
    <input type="hidden" id="memo" name="memo" value="{{ form_data.memo }}" />
    <input type="hidden" id="checked" name="checked" value="{{ form_data.checked }}" />
  </form>
</body>
</html>

その他のソースコード内容は、以下のサイトを参照のこと。
https://github.com/purin-it/python/tree/master/make-flask-pyodbc/