【コピペで完了】9割の機能を網羅!PytonからSQLiteを扱うクラスを作ってみました。

当ページのリンクには広告が含まれています。

PythonのSQLiteパッケージを使うと、メモリに乗り切らないデータを1件づつ読み込んで処理するような小回りの利く操作が可能です。

しかし、そもそもメモリに乗り切らない大量データに対しては、SQLで処理することがデータベースの強みですし、メモリに読み込む必要があるとしても、通常は抽出条件で一部のみを抽出するケースが大半です。

だったら、コネクトやクローズなど必ず付いて回る記述は省略するなどで、もっと簡単にSQLiteを使いたいと思いますよね。

ということで、以前C#でPython用の便利クラスを公開しましたが、今回はPython版を作ったので公開したいと思います。

単にSQLiteへのコネクトやクローズだけではなく、テーブルの存在チェック、テーブル作成、1つのトランザクション内での複数SQLの実行、テーブル一覧やカラム一覧など取得機能も用意していますので、是非参考にして頂ければと思います。

目次

クラスの使い方

今回は LocalCache というクラス名にしています。

使う際には、下記の通り sqlite3 のパッケージをインポートしておいて下さい。

import sqlite3

リファレンス(メソッドの名前と仕様)

今回のクラスに実装されているメソッドの一覧と仕様は以下の通りです。

機能メソッド名と引数戻り値補足
コンストラクタLocalCache(filenameインスタンスデータベースファイルを指定してインスタンスを生成
任意のSQLを実行execute(sql)なし任意のSQLを1つだけ実行
複数のSQLをまとめて実行execute_all(sqls)なしリストに格納された複数のSQLをトランザクション付きで実行。
エラーが発生するとロールバックされる。
問い合わせSQLの実行execute_query(sql)リストselect等の問い合わせSQLを実行し、全ての結果をリストで返す。
例: [ (aa , 123 , True) , (bb , 567 , False) ]
1つの値を返すSQLの実行execute_scalor(sql)単一の値1個の値を返す問い合わせSQLを実行し、結果を取得
テーブルの作成create(
tablename,
columns,
primarykey = '',
isdrop=False
)
なしテーブル名、カラム名、主キーを指定してテーブルを作成する。
既にテーブルがあればエラーとなるが、
isdrop=Trueにすることで、既にテーブルが存在していれば削除を実行する。
テーブルの削除drop(tablename)なしテーブルが存在していれば削除する。
テーブルの存在チェックexists(tablename)True/Falseテーブルが存在していれば True、存在しなければ Falseを返す。
テーブル名の変更rename
(
old_tablename,
new_tablename
)
なしold_tablename を new_table_nameに変更する
カラムの追加add_column
(
tablename,
columns
)
なしtablenameで指定したテーブルにcolumnsで指定した列を追加する。
columnsはカラム名をカンマ区切りで羅列する。
カラム名の後に型を指定することも可能。
例:'col1 , col2 , col3 text , col4 real'
Create文の取得get_create_statment
(
tablename
)
strtablenameで指定したテーブルのCreate文を文字列で返す。
テーブル一覧の取得get_table_list
(
table_type=’’
)
リストデータベースに登録されているテーブル及びビューの名前をリストで返す。
例:[ 'table1' , 'table2' , 'table3' ]
カラムの名前と型の一覧を取得get_column_type(
tablename
)
リストtablenameで指定したテーブルにおいて、カラム名と型をタプルに格納し、リストで返す。
例:[ ('col1' , 'text') , ('col2' , 'real') ]
カラムの名前の一覧を取得get_column_list(
tablename
)
リストtablenameで指定したテーブルにおいて、カラム名をリストで返す。
例:[ 'col1' , 'col2' , 'col3' ]
データベースの空き領域を解放vacuum()なしDeleteやdorp で生じた未使用領域を解放し、データベースファイルの容量を小さくする。
現在のデータベース容量×2倍の空き容量が必要。

簡単な使い方のサンプル

test といテーブルを作成し、データの追加、検索、カラム追加など、一通りの処理を実行するサンプルです。

db = LocalCache('d:/data.db')
db.drop('test2')
db.create('test','name text,num int','name,num',isdrop=True)
sqls = []
for n in range(10):
    sqls.append(f"insert into test values('data{n}',{n})")

db.execute_all(sqls)

data = db.execute_query('select * from test')
print(data)

db.add_column('test','add1 text,add2 int,add3')

print(db.exists('test'))
print(db.get_create_statment('test'))
print(db.get_table_list())
print(db.get_column_list('test'))
print(db.get_column_type('test'))
print(db.execute_scalor("select sum(num) as num from test"))
db.vacuum()
db.rename('test','test2')

クラスのソースコード

以下が今回のクラスのソースコードです。

そのままコピー&ペーストで張り付けて利用できます。

あとは、自由に変更してお使いください。

一応、Pythonで推奨される命名規則とコメントの書き方を使っています。

import sqlite3

class LocalCache:
    '''
    SQLiteのヘルパークラス
    
    Parameters
    ----------
    filename : str
        SQLiteのデータベースファイルへのパス
    
    '''
    
    def __init__(self,dbname):
        self.dbname = dbname
    
    def execute(self,sql):
        '''
        SQLを実行し、結果を取得する
        
        Parameters
        ----------
        columns : str
            実行したいSQL
        '''
        conn = sqlite3.connect(self.dbname)
        cur = conn.cursor()
        cur.execute(sql)
        conn.commit()
        cur.close()
        conn.close()
        
    def execute_all(self,sqls):
        '''
        複数のSQLをトランザクション配下で実行する
        
        Parameters
        ----------
        columns : strs
            実行したいSQLのリスト
        '''
        conn = sqlite3.connect(self.dbname)
        cur = conn.cursor()
        try:
            for sql in sqls:
                cur.execute(sql)
            conn.commit()
        except sqlite3.Error as e:
            conn.rollback()
        
        cur.close()
        conn.close()
        
    def execute_query(self,sql):
        '''
        select 系のSQLを実行し、結果を全て取得する
        
        Parameters
        ----------
        columns : str
            実行したいSQL
        
        Returns
        ----------
        data: list
            1行分をタプルとし、複数行をリストとして返す
            <例> [('RX100','Sony',35000),('RX200','Sony',42000)]
        '''
        conn = sqlite3.connect(self.dbname)
        cur = conn.cursor()
        cur.execute(sql)
        res = cur.fetchall()
        cur.close()
        conn.close()
        return res
        
    def execute_scalor(self,sql):
        '''
        結果の値が1つしかないSQLを実行し、結果を取得する
        
        Parameters
        ----------
        columns : str
            実行したいSQL
        
        Returns
        ----------
        res:
            実行結果により返された値
        '''
        conn = sqlite3.connect(self.dbname)
        cur = conn.cursor()
        cur.execute(sql)
        res = cur.fetchone()
        cur.close()
        conn.close()
        return res[0] if res != None else None
        
    def create(self,tablename,columns,primarykey = '',isdrop=False):
        '''
        テーブルを作成する
        
        Parameters
        ----------
        columns : str
            「列名」又は「列名+型」をカンマ区切りで指定
            <例> 'product text,price int,maker,year'
        primarykey: str
            プライマリーキーをカンマ区切りで指定
            <例> 'product,year'
        '''
        if isdrop :
            self.drop(tablename)
        
        pkey = ',primary key({0})'.format(primarykey) if primarykey != '' else ''
        sql = 'create table {0}({1} {2})'.format(tablename,columns,pkey)
        self.execute(sql)
        
    def drop(self,tablename):
        '''
        指定されたテーブルが存在すれば削除、無ければ何もしない
        
        Parameters
        ----------
        tablename : str
            削除したいテーブル名
        '''
        self.execute("drop table if exists {0}".format(tablename))
        
    def exists(self,tablename):
        '''
        指定したテーブル、又はビューの有無を判定する
        
        Parameters
        ----------
        columns : str
            実行したいSQL
        Returns
        ----------
            テーブル又はビューが存在すればTrue 存在しなければ False
        '''
        res = self.execute_scalor("select count(*) from sqlite_master where name='{0}'".format(tablename))
        return True if res > 0 else False
        
    def rename(self,old_tablename,new_tablename):
        '''
        テーブル名を変更する
        
        Parameters
        ----------
        old_tablename : str
            変更前のテーブル名
        new_tablename : str
            変更後のテーブル名
        '''
        self.execute_query("alter table {0} rename to {1}".format(old_tablename,new_tablename))
        
    def add_column(self,tablename,columns):
        '''
        テーブル名を変更する
        
        Parameters
        ----------
        tablename : str
            テーブル名
        columns : str
            「列名」又は「列名+型」をカンマ区切りで指定
            <例> 'product text,price int,maker,year'
        '''
        sqls = []
        for column in columns.split(','):
            sqls.append("alter table {0} add column {1}".format(tablename,column))
        
        self.execute_all(sqls)
        
    def get_create_statment(self,tablename):
        '''
        指定したテーブルのCreate文を取得する
        
        Parameters
        ----------
        tablename : str
            テーブル名
        
        Returns
        ----------
        res:str
            Create文
        '''
        res = self.execute_scalor("select sql from sqlite_master where name='{0}'".format(tablename))
        return res
        
    def get_table_list(self,table_type=''):
        '''
        登録されているテーブルの一覧を取得する
        
        Parameters
        ----------
        table_type : str
            'table' => テーブルのみ、'view' => ビューのみ、'' => テーブルとビューの両方
        
        Returns
        ----------
        res:str
            リスト形式のテーブル名一覧
            <例>  ['talbe1','table2','table3']
        '''
        res = self.execute_query("select tbl_name from sqlite_master where type like '%{0}%'".format(table_type))
        return [name[0] for name in res]
        
    def get_column_type(self,tablename):
        '''
        指定したテーブルのカラムと型を一覧で取得する
        
        Parameters
        ----------
        tablename : str
            テーブル名
        
        Returns
        ----------
        res:str
            リスト形式でカラム名と型のタプルを返す
            <例>  [('column1','int'),('column2','text'),('column3',real)]
        '''
        res = self.execute_query("Pragma table_info('{0}')".format(tablename))
        return [(name[1],name[2]) for name in res]
        
    def get_column_list(self,tablename):
        '''
        指定したテーブルのカラム名を一覧で取得する
        
        Parameters
        ----------
        tablename : str
            テーブル名
        
        Returns
        ----------
        res:str
            リスト形式のカラム名一覧
            <例>  ['column1','column2','column3']
        '''
        res = self.execute_query("Pragma table_info('{0}')".format(tablename))
        return [name[1] for name in res]
        
    def vacuum(self):
        '''
        データベースの空き領域を解放する
        '''
        self.execute('vacuum')

補足説明

この章では、本クラスについての改良ポイント、課題について解説します。

もし再利用されるのであれば、この点についてご留意ください。

DB接続/クローズ処理の共通化

SQLは大きく2つのカテゴリに分けることが出来ます。

  • データベース又はデータの更新に関するもの(更新系)
  • データの参照に関するもの(参照系)

データベースへの接続(connect) 、カーソル(cursor)の取得、カーソルと接続のクローズ処理は全てに共通ですが、更新はコミットやロールバック処理が必要なのに対し、参照はフェッチ(結果の受信)が必要になるなど、微妙に違いがあります。

#更新系の処理
conn = sqlite3.connect(self.dbname)
cur = conn.cursor()
cur.execute(sql)
conn.commit()
cur.close()
conn.close()

#参照系の処理
conn = sqlite3.connect(self.dbname)
cur = conn.cursor()
cur.execute(sql)


data = cur.fetchall()
cur.close()
conn.close()

最初は、接続とクローズの共通化ということで、下記の様なメソッドを考えました。

def __open(self):
    self.__conn = sqlite3.connect(self.dbname)
    self.__cur = conn.cursor()

def __close(self):
    self.__cur.close()
    self.__conn.close()

しかし、プログラムが少々複雑になるため、第三者が修正しにくくなるのではないかと考えて断念しました。

__open や __close の様なものを用意するのであれば、基底クラスとして作成しておいて、これを継承するような書き方が望ましいのかもしれませんが、今回はそれも凝りすぎかと思って止めました。

カラム名の変更

多くのデータベースは、カラム名を変更するために alter table ~ rename column という命令をサポートしています。

しかし、残念ながらSQLiteではサポートされていません。

対策としては、元のテーブルのcreate文からカラム名を変更した新しいテーブルを作成し、元のテーブルの中身をコピーしてから元のテーブルを変えるという荒業を行わなければなりません。

問題となるのは旧テーブルのCreate文に対して、どのように新しいカラム名を反映させるかです。

一番安直な方法は、カラム名をreplaceすることですが、型も変更したいケースもあるでしょうし、他のカラム名と部分一致してしまう可能性も考えられます。

となると、空白やカンマでトークン分解(splitメソッド)を行い、末尾の primary key を対象外にするとともに、primary key 以降のカラム名は変更対象とする・・・という複雑な手順が必要です。

一般的にカラム名の変更はそれほど頻度が多くないので、今回は見送ることにしました。

まとめ

今回は Python でSQLiteを使う上で、あれば便利な機能を一通り網羅したヘルパークラスを作成したので、公開してみました。

テーブル一覧の取得やカラム一覧の取得、create文の取得など、個々に検索して入手しなければならない情報もまとめて掲載しています。

おそらくだいたいのことは出来ると思いますので、後は皆さんの利用方法に合わせて拡張して頂ければと思います。

C#について同じようなクラスを作成していますので、興味のある方はこちらも参照して頂ければと思います。

よかったらシェアしてね!
  • URLをコピーしました!
  • URLをコピーしました!

コメント

コメントする

目次