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)

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

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()" />
    
    <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>