TOOLBOX/SQLite.py at master · YUDONGLING/TOOLBOX · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def CreateDB(Path: str) -> dict:
    '''
    Create a SQLite Database.
    '''
    import os
    import sqlite3

    if not __package__:
          from  Init import DotAccessDict; from  Log import MakeErrorMessage
    else: from .Init import DotAccessDict; from .Log import MakeErrorMessage

    Response = DotAccessDict({
        'Ec': 0, 'Em': '', 'Result': None
    })

    try:
        if os.path.dirname(Path):
            os.makedirs(os.path.dirname(Path), exist_ok = True)
    except Exception as Error:
        Response['Ec'] = 50001; Response['Em'] = MakeErrorMessage(Error); return Response

    try:
        Conn = sqlite3.connect(Path)
        Conn.close()
        Response['Result'] = Path
    except Exception as Error:
        Response['Ec'] = 50002; Response['Em'] = MakeErrorMessage(Error); return Response

    return Response


def ExecuteDB(Path: str, Query: str, Param: tuple = None) -> dict:
    '''
    Execute a SQLite Query.
    '''
    import sqlite3

    if not __package__:
          from  Init import DotAccessDict; from  Log import MakeErrorMessage
    else: from .Init import DotAccessDict; from .Log import MakeErrorMessage

    Response = DotAccessDict({
        'Ec': 0, 'Em': '', 'Result': None
    })

    try:
        Conn   = sqlite3.connect(Path)
        Cursor = Conn.cursor()
        Cursor.execute(Query, Param or ())
    except Exception as Error:
        if Conn:
            try: Conn.close()
            except: pass
        Response['Ec'] = 50001; Response['Em'] = MakeErrorMessage(Error); return Response

    try:
        Response['Result'] = Cursor.fetchall()
    except Exception as Error:
        if Conn:
            try: Conn.close()
            except: pass
        Response['Ec'] = 50002; Response['Em'] = MakeErrorMessage(Error); return Response

    try:
        if any(_ in Query.upper() for _ in ('INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP', 'ALTER')):
            Conn.commit()
        Conn.close()
    except Exception as Error:
        Response['Ec'] = 50003; Response['Em'] = MakeErrorMessage(Error); return Response

    return Response