なぽろぐ

気ままに感じたことを記事にまとめます。

SQLiteを個人的にいじりやすくしてみました

SQLite3を簡単に使いたい

SQLite3を使ってうーんとなった

めちゃめちゃいいライブラリだと思います。簡単にデータベースっぽいことできるのがよい・・・よい

でもなんかSQLiteのコマンドは忘れるわ、接続いちいちしないといけないわで調べるのに時間かかるし(メモとっとけって話だけど)、なにより接続の部分ががががって感じだった

デコレータ使うとどうやらその辺が改善されるとのことでやってみることにした〜ん。プログラムとかオブジェクト指向とか5ヶ月前に知ったばっかりで赤ちゃんだからなんでこんなことしてるの?とか変数の名前わかりにくすぎwwwwってのがあれば切実に教えて欲しいです。

肝心のプログラム

こんな感じです

import sqlite3
import time
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Tuple

import pandas as pd


class Handle:
    def __init__(self, db: str):
        self.dbpath = db

    # 1つのレコードを挿入する命令を作るメソッド
    def __MakeOneInsertOrder(self, table: str, data: List):
        """
        1つのレコードを挿入する
        data: 時間は自動でセットされる
        return: 1レコードのinsert文
        """
        print("MakeOneInsertOrder start")
        insert = "INSERT INTO {} VALUES ".format(table)
        s = self.__TupleMake(data)
        insert += s
        print(insert)
        print("MakeOneInsertOrder finish")
        return insert

    # 複数のレコードを挿入する命令を作るメソッド
    def __MakeSomeInsertOrder(self, table: str, datas: List[List]):
        """
        複数レコードを挿入する
        data: 挿入したいデータを入れ子にすること
              時間は自動でセットされる
        return: 複数レコードのinsert文 
        """
        print("MakeSomeInsertOrder start")
        insert = "INSERT INTO {} VALUES ".format(table)
        s = [self.__TupleMake(data) for data in datas]
        s = ", ".join(s)
        insert += s
        print(insert)
        print("MakeSomeInsertOrder finsish")
        return insert

    # テーブルを作成する命令を作るメソッド
    def __MakeTableOrder(self, table: str, columns: List):
        """
        テーブルを新しく作る命令文を作るメソッド
        columns: 設定したいカラム, dateは自動でつけられる
        return: create文
        """
        print("MaleTableOrder start")
        create = "CREATE TABLE IF NOT EXISTS {} ".format(table)
        record = "(date DATE PRIMARY KEY"
        for col in columns:
            record += ", "+col+" TEXT"
        record += ")"
        create += record
        print(create)
        print("MaleTableOrder finish")
        return create

    def __MakeSelectOrder(self, table: str, hour: int):
        """
        select文を作るメソッド
        return select文
        """
        print("MakeSelectOrder start")
        if hour > 0:
            hour = self.__CreateTimeStamp() - 3600*hour
            select = "SELECT * from {} WHERE date > {}".format(table, hour)
        else:
            select = "SELECT * from {}".format(table)
        return select

    # レコードに挿入するデータ形式に変換するメソッド
    def __TupleMake(self, data: List):
        """
        リスト形式のデータを()の入れ子に変換する
        """
        print("TupleMake start")
        time = self.__CreateTimeStamp()
        s = "({}".format(time)
        for item in data:
            s += ", '{}'".format(item)
        s += ")"
        print("Tuple finish")
        return s

    # タイムスタンプを返す
    def __CreateTimeStamp(self):
        """
        タイムスタンプを作る
        """
        print("CreateTimeStamp start")
        now = time.time()
        JST = timezone(timedelta(hours=+9), 'JST')
        loc = datetime.fromtimestamp(now, JST).timestamp()
        print("CreateTimeStamp finish")
        return loc

    # sqlite3のコネクションを司るデコレータ
    # わかっていない部分もあるので今後バシバシ使っていくこと
    def connectdb(func):
        from functools import wraps

        @wraps(func)
        def _connectdb(self, *args, **kwargs):
            print("connection start")
            con = sqlite3.connect(self.dbpath)
            cur = con.cursor()
            try:
                result = func(self, *args, **kwargs, con=con, cur=cur)
            except sqlite3.Error as e:
                print('sqlite3.Error occurred:', e.args[0])
                result = None
            finally:
                con.commit()
                con.close()
                print("connection finish")
            return result
        return _connectdb

    @connectdb
    def MakeTable(self, table: str, columns: List, con=None, cur=None):
        """
        テーブルを実際に作るメソッド
        """
        print("MakeTable start")
        order = self.__MakeTableOrder(table, columns)
        cur.execute(order)
        print("MakeTable finish")

    @connectdb
    def InsertRecord(self, table: str, data: List, con=None, cur=None):
        print("InsertRecord start")
        if isinstance(data[0], list):
            order = self.__MakeSomeInsertOrder(table, data)
        else:
            order = self.__MakeOneInsertOrder(table, data)
        cur.execute(order)
        print("InsertRecord finish")

    @connectdb
    def GetData(self, table: str, hour=1, con=None, cur=None):
        """
        データを指定した時間分取得するメソッド
        hour=0の時は全てのデータを取ってくる
        return: pandasデータフレーム
        """
        print("GetData start")
        if table not in self.GetTableList():
            print("!!!Error!!! {} is not exist!!!".format(table))
            print("GetData finish")
            return pd.DataFrame()
        else:
            order = self.__MakeSelectOrder(table, hour)
            df = pd.read_sql(order, con)
            print("GetData finish")
            return df

    @connectdb
    def GetTableList(self, con=None, cur=None):
        """
        作られているテーブル一覧を返すメソッド
        """
        print("GetTableList start")
        select = "SELECT * FROM sqlite_master WHERE TYPE = 'table'"
        cur.execute(select)
        print("GetTableList finish")
        return [x[1] for x in cur.fetchall()]

    @connectdb
    def UpdateRecord(self, table: str, record: dict, update_data: dict, con=None, cur=None):
        """
        table: テーブル名
        record: {"カラム名": "指定するレコードを示すデータ"}
        update_data: {"カラム名": 変更する内容} 
        """
        print("UpdateRecord start")
        rec_key, rec_val = list(record.items())[0]
        update_key, update_val = list(update_data.items())[0]
        order = "update {} set {} = '{}' where {} = {}".format(
            table, update_key, update_val, rec_key, rec_val)
        cur.execute(order)
        print("UpdateRecord finish")

    @connectdb
    def DeleteTable(self, table: str, con=None, cur=None):
        """
        table: 削除したいテーブル名
        """
        print("DeleteTable start")
        # テーブルの削除
        cur.execute("DROP TABLE {}".format(table))
        # 空き容量の解放
        cur.execute("VACUUM")
        print("DeleteTable finish")

    @connectdb
    def DeleteRecord(self, table: str):
        pass

デコレータを使ったところ

# sqlite3のコネクションを司るデコレータ
    # わかっていない部分もあるので今後バシバシ使っていくこと
    def connectdb(func):
        from functools import wraps

        @wraps(func)
        def _connectdb(self, *args, **kwargs):
            print("connection start")
            con = sqlite3.connect(self.dbpath)
            cur = con.cursor()
            try:
                result = func(self, *args, **kwargs, con=con, cur=cur)
            except sqlite3.Error as e:
                print('sqlite3.Error occurred:', e.args[0])
                result = None
            finally:
                con.commit()
                con.close()
                print("connection finish")
            return result
        return _connectdb

wrapsを使うとどうなるのかよくわかってない・・・データを保持できるとかなんとか書いてあったけどいまいち現象が掴めないって感じです

_connectidb(self, *args, **kwargs)にself入れ込んだことでself使えるようになってはっぴーって感じ!

デコレータ自体を書ける気がしない・・・・この書き方なんかこんがらがりませんか????

これ以外のメソッドは大したことしていないので見たらすぐわかると思います。

まとめ

いい感じで動いているので嬉しい