QuotaCost = NewType('QuotaCost', int)
YoutubeId = NewType('YoutubeId', str)
PathStr = NewType('PathStr', str)
-QueryId = NewType('QueryId', int)
+QueryId = NewType('QueryId', str)
QueryText = NewType('QueryText', str)
ProseText = NewType('ProseText', str)
SqlText = NewType('SqlText', str)
# tables to create database with
SCRIPT_INIT_DB = '''
CREATE TABLE yt_queries (
- id INTEGER PRIMARY KEY,
+ id TEXT PRIMARY KEY,
text TEXT NOT NULL,
retrieved_at TEXT NOT NULL
);
definition TEXT NOT NULL
);
CREATE TABLE yt_query_results (
- query_id INTEGER NOT NULL,
+ query_id TEXT NOT NULL,
video_id TEXT NOT NULL,
PRIMARY KEY (query_id, video_id),
FOREIGN KEY (query_id) REFERENCES yt_queries(id),
text: QueryText,
retrieved_at: DatetimeStr
) -> None:
- self.id_ = id_
+ self.id_ = id_ if id_ else QueryId(str(uuid4()))
self.text = QueryText(text)
self.retrieved_at = retrieved_at
- def save(self, conn: DatabaseConnection) -> Cursor:
- """DbData.save but assign .id_ from Cursor's lastrowid if yet unset."""
- cursor = super().save(conn)
- if self.id_ is None:
- assert cursor.lastrowid is not None
- self.id_ = QueryId(cursor.lastrowid)
- return cursor
-
@classmethod
def get_all_for_video(cls,
conn: DatabaseConnection,
query_data.save(conn)
for result in collect_results(query_txt):
result.save(conn)
- assert query_data.id_ is not None
result.save_to_query(conn, query_data.id_)
conn.commit_close()
self._send_http(headers=[('Location', f'/query/{query_data.id_}')],
elif 'video_about' == page_name:
self._send_video_about(YoutubeId(toks_url[2]))
elif 'query' == page_name:
- self._send_query_page(QueryId(int(toks_url[2])))
+ self._send_query_page(QueryId(toks_url[2]))
elif 'queries' == page_name:
self._send_queries_index_and_search()
elif '_last_playlist_update.json' == page_name: