Skip to content
Snippets Groups Projects
Verified Commit 2df4367c authored by Gaëtan Blond's avatar Gaëtan Blond
Browse files

Added database

parent 60bd55ee
No related branches found
No related tags found
1 merge request!1Refactor and clean old code
from .database import EntriesDatabase
from .entry import Entry from .entry import Entry
import sqlite3
from typing import Any, Iterable, Optional
from .entry import Entry
CREATE_ENTRIES_TABLE_SQL = """\
CREATE TABLE IF NOT EXISTS {table_name} (
feed_id integer,
tag_id text,
name text,
version text,
url text
)
"""
SEARCH_FOR_ENTRY_SQL = """\
SELECT
tag_id
FROM {table_name}
WHERE feed_id=?
AND tag_id=?
"""
INSERT_ENTRY_SQL = """\
INSERT INTO {table_name}
(feed_id, tag_id, name, version, url)
VALUES (?, ?, ?, ?, ?)
"""
class EntriesDatabase:
def __init__(self, db_path: str, table_name: str = "entries"):
self.table_name = table_name
self._conn = sqlite3.connect(db_path)
self._cursor: Optional[sqlite3.Cursor] = None
self._execute(CREATE_ENTRIES_TABLE_SQL)
self._conn.commit()
def _execute(
self, query: str, params: Iterable[Any] = None, no_cursor=False
) -> sqlite3.Cursor:
if params is None:
params = []
if no_cursor or self._cursor is None:
execute_func = self._conn.execute
else:
execute_func = self._cursor.execute
return execute_func(query.format(table_name=self.table_name), params)
def contains(self, entry: Entry) -> bool:
if not isinstance(entry, Entry):
raise TypeError("Element must be of type Entry")
cursor = self._execute(
SEARCH_FOR_ENTRY_SQL, (entry.feed_id, entry.tag_id), True
)
return cursor.fetchone() is not None
def insert(self, entry: Entry) -> None:
self._execute(
INSERT_ENTRY_SQL,
(entry.feed_id, entry.tag_id, entry.name, entry.version, entry.url),
)
def __contains__(self, entry: Entry) -> bool:
return self.contains(entry)
def __enter__(self) -> None:
if self._cursor is not None:
raise RuntimeError("Global cursor already activated.")
self._cursor = self._conn.cursor()
def __exit__(self, *_) -> None:
# TODO decide wether all transactions are committed when an exception is raised
if self._cursor is None:
raise RuntimeError("Global cursor not activated")
self._conn.commit()
self._cursor = None
def __iadd__(self, entry: Entry) -> None:
self.insert(entry)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment