libs.functions.tools.vacuum

libs/functions/tools/vacuum.py

View Source

 1"""
 2libs/functions/tools/vacuum.py
 3"""
 4
 5import logging
 6import os
 7from contextlib import closing
 8from typing import TYPE_CHECKING
 9
10import libs.global_value as g
11from libs.domain import modify
12from libs.utils import dbutil
13
14if TYPE_CHECKING:
15    from sqlite3 import Connection
16
17
18def main() -> None:
19    """vacuum実行"""
20    g.cfg.initialization()
21
22    modify.db_backup()
23    before_size = os.path.getsize(g.cfg.setting.database_file)
24
25    with closing(dbutil.connection(g.cfg.setting.database_file)) as cur:
26        before_page = db_info(cur, "page_count")
27        before_freelist = db_info(cur, "freelist_count")
28        cur.execute("vacuum;")
29        after_page = db_info(cur, "page_count")
30        after_freelist = db_info(cur, "freelist_count")
31
32    after_size = os.path.getsize(g.cfg.setting.database_file)
33
34    logging.info("file size: %s -> %s", before_size, after_size)
35    logging.info("page_count: %s -> %s", before_page, after_page)
36    logging.info("freelist_count: %s -> %s", before_freelist, after_freelist)
37
38
39def db_info(cur: "Connection", kind: str) -> int:
40    """
41    page_countを取得
42
43    Args:
44        cur (Connection): Connectionオブジェクト
45        kind (str): 取得する内容
46
47    Returns:
48        int: page_count / freelist_count
49
50    """
51    match kind:
52        case "page_count":
53            count = int(cur.execute("pragma page_count;").fetchone()[0])
54        case "freelist_count":
55            count = int(cur.execute("pragma freelist_count;").fetchone()[0])
56        case _:
57            count = 0
58
59    return count

def main() -> None: View Source

19def main() -> None:
20    """vacuum実行"""
21    g.cfg.initialization()
22
23    modify.db_backup()
24    before_size = os.path.getsize(g.cfg.setting.database_file)
25
26    with closing(dbutil.connection(g.cfg.setting.database_file)) as cur:
27        before_page = db_info(cur, "page_count")
28        before_freelist = db_info(cur, "freelist_count")
29        cur.execute("vacuum;")
30        after_page = db_info(cur, "page_count")
31        after_freelist = db_info(cur, "freelist_count")
32
33    after_size = os.path.getsize(g.cfg.setting.database_file)
34
35    logging.info("file size: %s -> %s", before_size, after_size)
36    logging.info("page_count: %s -> %s", before_page, after_page)
37    logging.info("freelist_count: %s -> %s", before_freelist, after_freelist)

vacuum実行

def db_info(cur: sqlite3.Connection, kind: str) -> int: View Source

40def db_info(cur: "Connection", kind: str) -> int:
41    """
42    page_countを取得
43
44    Args:
45        cur (Connection): Connectionオブジェクト
46        kind (str): 取得する内容
47
48    Returns:
49        int: page_count / freelist_count
50
51    """
52    match kind:
53        case "page_count":
54            count = int(cur.execute("pragma page_count;").fetchone()[0])
55        case "freelist_count":
56            count = int(cur.execute("pragma freelist_count;").fetchone()[0])
57        case _:
58            count = 0
59
60    return count

page_countを取得

Arguments:

cur (Connection): Connectionオブジェクト
kind (str): 取得する内容

Returns:

int: page_count / freelist_count