CLOVER🍀

That was when it all began.

SQLiteをPythonから使う

これは、なにをしたくて書いたもの?

ちょっとした用途でお手軽にデータベースを使いたいといった文脈の時に、JavaだとH2 Databaseあたりが挙がりますが、
そういえばSQLiteを使ったことないよなと思いまして。

1度、ちゃんと見てみようかなと思って今回使ってみました。

求めているのは、組み込みデータベースで、ある程度クロスプラットフォーム、複数のプログラミング言語で扱えそうなもの、です。

SQLite

SQLiteは、マルチプラットフォームで動作する、オープンソースの組み込み型のRDBMSです。

SQLite Home Page

ドキュメントは、こちら。

SQLite Documentation

SQLiteはC言語で実装されていて、他のRDBMSと比べてサーバープロセスがなく、アプリケーション内でライブラリとして
使われるのが特徴です。

About SQLite

データの保存先はメモリまたはファイルで、多くのプログラミング言語から利用することができます。

データ型や構文に関するページは、こちら。

Datatypes In SQLite Version 3

Query Language Understood by SQLite

データ型は、INTEGER、REAL、TEXT、BLOB、そしてNULLと少なめ。VARCHARみたいな表現が使えないこともないですが、
TEXTに読み替えられます。このあたりは、ドキュメントに記載があります。

組み込み型のデータベースではありますが、CLIも提供されているようです。

Command Line Shell For SQLite

というか、ドキュメントがとてもしっかりしていますね。Androidとかで使われたりすることも大きいのでしょうか?

PythonとSQLite

Pythonは、SQLiteを標準で利用することができます。正確に言うと、SQLite3を使える、ですね。

12.6. sqlite3 --- SQLite データベースに対する DB-API 2.0 インタフェース — Python 3.6.10 ドキュメント

DB-API 2.0に準拠したAPIを提供しているとありますが、DB-APIについて知らなかったので見てみたら、Pythonにおける
データベースアクセスのための標準APIのようですね。

PEP 249 -- Python Database API Specification v2.0 | Python.org

環境

今回の環境は、こちらです。

$ python3
Python 3.6.9 (default, Apr 18 2020, 01:56:04) 
[GCC 8.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import sqlite3
>>> sqlite3.version
'2.6.0'
>>> sqlite3.sqlite_version
'3.22.0'

SQLiteとしては、3.22を使用することになります。

テストコードの雛形を作成する

今回は、お題を書籍にしてテストコードで確認しつつ、かつ外部ライブラリの使用はなしでいってみたいと思います。

つまり、こんな感じのテストクラスを用意して書いていきます。 sqlite_test.py

import sqlite3

import unittest

class SqliteTest(unittest.TestCase):
    # ここに、テストコードを書く!

実行。

$ python3 -m unittest sqlite_test.py

SQLiteをインメモリで使う

では、こちらのページを見ながら、SQLite3モジュールを使っていってみたいと思います。

12.6. sqlite3 --- SQLite データベースに対する DB-API 2.0 インタフェース — Python 3.6.10 ドキュメント

最初は、SQLite3をインメモリで扱ってみたいと思います。

こんな感じで。

    def test_in_memory_database(self):
        with sqlite3.connect(":memory:") as con :
            cur = con.cursor()

            try:
                # DDL
                cur.execute("""
                create table book(
                  isbn text,
                  title text,
                  price integer,
                  primary key(isbn)
                )""")

                books = [
                    {"isbn": "978-4873117386", "title": "入門 Python 3", "price": 4070},
                    {"isbn": "978-4297111113", "title": "Python実践入門 言語の力を引き出し、開発効率を高める (WEB+DB PRESS plusシリーズ)", "price": 3278},
                    {"isbn": "978-4048930611", "title": "エキスパートPythonプログラミング改訂2版", "price": 5256}
                ]

                # insert
                for book in books:
                    cur.execute("insert into book(isbn, title, price) values(:isbn, :title, :price)", book)

                con.commit()

                # select count
                cur.execute("select count(*) from book")
                self.assertEqual(cur.fetchone(), (3,))

                # select all
                cur.execute("select isbn, title, price from book order by price desc")
                results = cur.fetchall()
                self.assertEqual(len(results), 3)
                self.assertEqual(results[0], ("978-4048930611", "エキスパートPythonプログラミング改訂2版", 5256))
                self.assertEqual(results[1], ("978-4873117386", "入門 Python 3", 4070))
                self.assertEqual(results[2], ("978-4297111113", "Python実践入門 言語の力を引き出し、開発効率を高める (WEB+DB PRESS plusシリーズ)", 3278))

                # select filter
                cur.execute("select isbn, title, price from book where isbn = :isbn", {"isbn": "978-4873117386"})
                results = cur.fetchall()
                self.assertEqual(len(results), 1)
                self.assertEqual(results[0], ("978-4873117386", "入門 Python 3", 4070))
            finally:
                cur.close()

最初にConnectionを作成しますが、この時に「:memory:」と指定すると、インメモリデータベースになります。

        with sqlite3.connect(":memory:") as con :

Connectionはコンテキストマネージャーを実装しているので、with構文で扱えます。

次にCursorを取得して、SQLを実行していくことになります。

            cur = con.cursor()

使い終わったCursorは、最後にcloseします。

                cur.close()

Cursorは、コンテキストマネージャーを実装していません。

あとは、DDL、DMLを実行していきましょう。

テーブルの作成。

                # DDL
                cur.execute("""
                create table book(
                  isbn text,
                  title text,
                  price integer,
                  primary key(isbn)
                )""")

データの登録。

                books = [
                    {"isbn": "978-4873117386", "title": "入門 Python 3", "price": 4070},
                    {"isbn": "978-4297111113", "title": "Python実践入門 言語の力を引き出し、開発効率を高める (WEB+DB PRESS plusシリーズ)", "price": 3278},
                    {"isbn": "978-4048930611", "title": "エキスパートPythonプログラミング改訂2版", "price": 5256}
                ]

                # insert
                for book in books:
                    cur.execute("insert into book(isbn, title, price) values(:isbn, :title, :price)", book)

プレースホルダーは「?」と「:名前」の2種類が使えるようなのですが、今回は名前を使ったスタイルでいきます。名前を使った
スタイルの場合、パラメーターは辞書で指定することになります。

今回は使いませんが、「?」を使った場合はパラメーターはタプルでの指定となります。

コミットやロールバックは、Connectionに対して行います。

                con.commit()

select count。

                # select count
                cur.execute("select count(*) from book")
                self.assertEqual(cur.fetchone(), (3,))

1件のみの場合は、Cursor#fetchoneで良さそうです。結果は、タプルで返ってきます。

全件取得や、where句で絞り込んだ場合。

                # select all
                cur.execute("select isbn, title, price from book order by price desc")
                results = cur.fetchall()
                self.assertEqual(len(results), 3)
                self.assertEqual(results[0], ("978-4048930611", "エキスパートPythonプログラミング改訂2版", 5256))
                self.assertEqual(results[1], ("978-4873117386", "入門 Python 3", 4070))
                self.assertEqual(results[2], ("978-4297111113", "Python実践入門 言語の力を引き出し、開発効率を高める (WEB+DB PRESS plusシリーズ)", 3278))

                # select filter
                cur.execute("select isbn, title, price from book where isbn = :isbn", {"isbn": "978-4873117386"})
                results = cur.fetchall()
                self.assertEqual(len(results), 1)
                self.assertEqual(results[0], ("978-4873117386", "入門 Python 3", 4070))

結果はリストで返ってきますが、各行に対応するデータはタプルで返ってくるようです。

結果をRowで受け取る

Cursor#executeの結果はタプルとして返ってきていましたが、これをRowに変更することが可能です。

Rowは、タプルによく似ているものの、カラム名でのアクセスが可能になったりします。

Rowを使ったソースコードはこちら。

    def test_in_memory_database_using_row(self):
        with sqlite3.connect(":memory:") as con :
            con.row_factory = sqlite3.Row
            cur = con.cursor()

            try:
                # DDL
                cur.execute("""
                create table book(
                  isbn text,
                  title text,
                  price integer,
                  primary key(isbn)
                )""")

                books = [
                    {"isbn": "978-4873117386", "title": "入門 Python 3", "price": 4070},
                    {"isbn": "978-4297111113", "title": "Python実践入門 言語の力を引き出し、開発効率を高める (WEB+DB PRESS plusシリーズ)", "price": 3278},
                    {"isbn": "978-4048930611", "title": "エキスパートPythonプログラミング改訂2版", "price": 5256}
                ]

                # insert
                for book in books:
                    cur.execute("insert into book(isbn, title, price) values(:isbn, :title, :price)", book)

                con.commit()

                # select count
                cur.execute("select count(*) from book")
                self.assertEqual(cur.fetchone()[0], 3)

                # select filter
                cur.execute("select isbn, title, price from book where isbn = :isbn", {"isbn": "978-4873117386"})
                results = cur.fetchall()
                self.assertEqual(len(results), 1)
                row = results[0]
                self.assertEqual(tuple(row), ("978-4873117386", "入門 Python 3", 4070))
                self.assertEqual(row["isbn"], "978-4873117386")
                self.assertEqual(row["title"], "入門 Python 3")
                self.assertEqual(row["price"], 4070)
            finally:
                cur.close()

Connectionのrow_factoryに、sqlite3.Rowを設定することで、Cursor#executeの結果が変わります。

        with sqlite3.connect(":memory:") as con :
            con.row_factory = sqlite3.Row
            cur = con.cursor()

select countの結果は、添字アクセスで取得してみました。最初のコードのように、タプルでは比較できません。

                # select count
                cur.execute("select count(*) from book")
                self.assertEqual(cur.fetchone()[0], 3)

カラム名でのアクセスの例。

                # select filter
                cur.execute("select isbn, title, price from book where isbn = :isbn", {"isbn": "978-4873117386"})
                results = cur.fetchall()
                self.assertEqual(len(results), 1)
                row = results[0]
                self.assertEqual(tuple(row), ("978-4873117386", "入門 Python 3", 4070))
                self.assertEqual(row["isbn"], "978-4873117386")
                self.assertEqual(row["title"], "入門 Python 3")
                self.assertEqual(row["price"], 4070)

また、Rowはタプルに変換することもできます。

                self.assertEqual(tuple(row), ("978-4873117386", "入門 Python 3", 4070))

データをファイルに保存する

先ほどはインメモリにデータを保存していましたが、今度はファイルに保存してみましょう。

で、書いたのがこちら。データをファイルに保存する以外にも、ちょっとだけ変更しています。

    def test_file_database(self):
        with sqlite3.connect("./sqlite_example.db") as con :
            cur = con.cursor()

            try:
                # DDL
                cur.execute("drop table if exists book")
                
                cur.execute("""
                create table book(
                  isbn text,
                  title text,
                  price integer,
                  primary key(isbn)
                )""")

                books = [
                    {"isbn": "978-4873117386", "title": "入門 Python 3", "price": 4070},
                    {"isbn": "978-4297111113", "title": "Python実践入門 言語の力を引き出し、開発効率を高める (WEB+DB PRESS plusシリーズ)", "price": 3278},
                    {"isbn": "978-4048930611", "title": "エキスパートPythonプログラミング改訂2版", "price": 5256}
                ]

                # insert many
                cur.executemany("insert into book(isbn, title, price) values(:isbn, :title, :price)", books)

                con.commit()

                # select count
                cur.execute("select count(*) from book")
                self.assertEqual(cur.fetchone(), (3,))

                # select all
                cur.execute("select isbn, title, price from book order by price desc")
                results = cur.fetchall()
                self.assertEqual(len(results), 3)
                self.assertEqual(results[0], ("978-4048930611", "エキスパートPythonプログラミング改訂2版", 5256))
                self.assertEqual(results[1], ("978-4873117386", "入門 Python 3", 4070))
                self.assertEqual(results[2], ("978-4297111113", "Python実践入門 言語の力を引き出し、開発効率を高める (WEB+DB PRESS plusシリーズ)", 3278))

                # select filter
                cur.execute("select isbn, title, price from book where isbn = :isbn", {"isbn": "978-4873117386"})
                results = cur.fetchall()
                self.assertEqual(len(results), 1)
                self.assertEqual(results[0], ("978-4873117386", "入門 Python 3", 4070))
            except sqlite3.DatabaseError as e:
                # ロールバック
                con.rollback()

                self.fail(f"unexpected error: {e}")
            finally:
                cur.close()

まず、データをファイルに保存するにはConnectionを作成する際に、ファイルパスを渡せばOKです。指定されたファイルパスに、
データベースファイルが作成されます。

       with sqlite3.connect("./sqlite_example.db") as con :
            cur = con.cursor()

データベースが永続化されてしまうので、テストを繰り返し実行できるようにするために、DROP&CREATEに変更。

                # DDL
                cur.execute("drop table if exists book")
                
                cur.execute("""
                create table book(
                  isbn text,
                  title text,
                  price integer,
                  primary key(isbn)
                )""")

データの登録は、Cursor#executemanyで一括で登録するようにしました。この場合、パラメーターをリストで渡すことになります。

                books = [
                    {"isbn": "978-4873117386", "title": "入門 Python 3", "price": 4070},
                    {"isbn": "978-4297111113", "title": "Python実践入門 言語の力を引き出し、開発効率を高める (WEB+DB PRESS plusシリーズ)", "price": 3278},
                    {"isbn": "978-4048930611", "title": "エキスパートPythonプログラミング改訂2版", "price": 5256}
                ]

                # insert many
                cur.executemany("insert into book(isbn, title, price) values(:isbn, :title, :price)", books)

例外発生時のロールバック処理も追加。

            except sqlite3.DatabaseError as e:
                # ロールバック
                con.rollback()

                self.fail(f"unexpected error: {e}")
            finally:
                cur.close()

と、こんな感じで。

基本的なことは、これでできるのではないでしょうか。