-
Notifications
You must be signed in to change notification settings - Fork 171
/
sqlite.py
122 lines (106 loc) · 3.42 KB
/
sqlite.py
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
# Copyright (c) 2017-present, Facebook, Inc.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
import datetime
import logging
import sqlite3
from typing import Iterable, List, Optional, Tuple, Union
from monkeytype.db.base import CallTraceStore, CallTraceThunk
from monkeytype.encoding import CallTraceRow, serialize_traces
from monkeytype.tracing import CallTrace
logger = logging.getLogger(__name__)
DEFAULT_TABLE = "monkeytype_call_traces"
def create_call_trace_table(
conn: sqlite3.Connection, table: str = DEFAULT_TABLE
) -> None:
query = """
CREATE TABLE IF NOT EXISTS {table} (
created_at TEXT,
module TEXT,
qualname TEXT,
arg_types TEXT,
return_type TEXT,
yield_type TEXT);
""".format(
table=table
)
with conn:
conn.execute(query)
QueryValue = Union[str, int]
ParameterizedQuery = Tuple[str, List[QueryValue]]
def make_query(
table: str, module: str, qualname: Optional[str], limit: int
) -> ParameterizedQuery:
raw_query = """
SELECT
module, qualname, arg_types, return_type, yield_type
FROM {table}
WHERE
module == ?
""".format(
table=table
)
values: List[QueryValue] = [module]
if qualname is not None:
raw_query += " AND qualname LIKE ? || '%'"
values.append(qualname)
raw_query += """
GROUP BY
module, qualname, arg_types, return_type, yield_type
ORDER BY date(created_at) DESC
LIMIT ?
"""
values.append(limit)
return raw_query, values
class SQLiteStore(CallTraceStore):
def __init__(self, conn: sqlite3.Connection, table: str = DEFAULT_TABLE) -> None:
self.conn = conn
self.table = table
@classmethod
def make_store(cls, connection_string: str) -> "CallTraceStore":
conn = sqlite3.connect(connection_string)
create_call_trace_table(conn)
return cls(conn)
def add(self, traces: Iterable[CallTrace]) -> None:
values = []
for row in serialize_traces(traces):
values.append(
(
datetime.datetime.now(),
row.module,
row.qualname,
row.arg_types,
row.return_type,
row.yield_type,
)
)
with self.conn:
self.conn.executemany(
"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, ?)".format(
table=self.table
),
values,
)
def filter(
self, module: str, qualname_prefix: Optional[str] = None, limit: int = 2000
) -> List[CallTraceThunk]:
sql_query, values = make_query(self.table, module, qualname_prefix, limit)
with self.conn:
cur = self.conn.cursor()
cur.execute(sql_query, values)
return [CallTraceRow(*row) for row in cur.fetchall()]
def list_modules(self) -> List[str]:
with self.conn:
cur = self.conn.cursor()
cur.execute(
"""
SELECT module FROM {table}
GROUP BY module
ORDER BY date(created_at) DESC
""".format(
table=self.table
)
)
return [row[0] for row in cur.fetchall() if row[0]]