refactor: remove redundant functionality

src/cli.py

@@ -3,21 +3,15 @@ import argparse
 from constants import EOLAS_DIRECTORY
 from controllers.controller import Controller
 from services.database_service import DatabaseService
-from services.graph_service import GraphService
 from services.parse_file_service import ParseFileService
 from services.table_service import TableService
-from services.tag_service import TagService
 
 database_service = DatabaseService("eolas")
 database_connection = database_service.connect()
 table_service = TableService(database_connection)
 parse_file_service = ParseFileService(EOLAS_DIRECTORY)
-graph_service = GraphService(database_connection)
-tag_service = TagService(database_connection)
 
-controller = Controller(
-    database_service, table_service, parse_file_service, graph_service, tag_service
-)
+controller = Controller(database_service, table_service, parse_file_service)
 
 
 def main():
@@ -26,7 +20,7 @@ def main():
     )
     parser.add_argument(
         "command",
-        choices=["populate-database", "generate-graph", "export-tags"],
+        choices=["populate-database"],
         help="Command to execute",
     )
     args = parser.parse_args()

src/controllers/controller.py

@@ -1,32 +1,21 @@
-from termcolor import colored
-
-
 class Controller:
     def __init__(
         self,
         database_service,
         table_service,
         parse_file_service,
-        graph_service,
-        tag_service,
     ):
         self.database_service = database_service
         self.table_service = table_service
         self.parse_file_service = parse_file_service
-        self.graph_service = graph_service
-        self.tag_service = tag_service
 
     def execute(self, operation):
         try:
             match operation:
                 case "populate":
                     return self.__populate_database()
-                case "graph":
-                    return self.__generate_graph()
-                case "tags":
-                    return self.__export_tags()
         except Exception as e:
-            raise Exception(colored(f"ERROR {e}", "red"))
+            raise Exception(f"ERROR {e}")
         finally:
             self.database_service.disconnect()
 
@@ -34,9 +23,3 @@ class Controller:
         entries = self.parse_file_service.parse_source_directory()
         self.table_service.populate_tables(entries)
         self.database_service.disconnect()
-
-    def __generate_graph(self):
-        self.graph_service.generate_graph()
-
-    def __export_tags(self):
-        self.tag_service.export_tags()

src/models/graph_edge.py

@@ -1,7 +0,0 @@
-from typing import TypedDict
-
-
-class IGraphEdge(TypedDict):
-    source: str
-    target: str
-

src/models/graph_node.py

@@ -1,7 +0,0 @@
-from typing import TypedDict
-
-
-class IGraphNode(TypedDict):
-    id: str
-    type: str
-

src/models/tag.py

@@ -1,5 +0,0 @@
-from typing import List, TypedDict
-
-
-class Tag(TypedDict):
-    name: str

src/services/database_service.py

@@ -2,8 +2,6 @@ import os
 import sqlite3
 from typing import Optional
 
-from termcolor import colored
-
 
 class DatabaseService:
     def __init__(self, db_name):
@@ -18,7 +16,7 @@ class DatabaseService:
         try:
             if not os.path.exists(self.db_path):
                 os.makedirs(self.db_path)
-                print(colored("INFO Created database directory", "blue"))
+                print("TB-INFO Created database directory")
             self.connection = sqlite3.connect(f"{self.db_path}/{self.db_name}.db")
             self.connection.execute("PRAGMA foreign_keys = ON")
             return self.connection

src/services/graph_service.py

@@ -1,53 +0,0 @@
-import json
-
-from constants import GRAPH_OUTPUT_DIRECTORY
-from models.graph_edge import IGraphEdge
-from models.graph_node import IGraphNode
-from services.sqlite_service import SqliteService
-
-
-class GraphService(SqliteService):
-    error_message_stub = "Could not retrieve contents of table:"
-
-    def __init__(self, db_connection):
-        super().__init__(db_connection)
-
-    def __get_nodes(self) -> list[IGraphNode]:
-        tags = self._query(
-            "SELECT * FROM tags",
-            error_message=f"{self.error_message_stub} tags",
-        )
-        tags = [IGraphNode(id=f"#{tag[0]}", type="tag") for tag in tags]
-        entries = self._query(
-            "SELECT title FROM entries",
-            error_message=f"{self.error_message_stub} entries",
-        )
-
-        entries = [IGraphNode(id=entry[0], type="entry") for entry in entries]
-        return tags + entries
-
-    def __get_edges(self):
-        tags = self._query(
-            "SELECT * FROM entries_tags",
-            error_message=f"{self.error_message_stub} entries_tags",
-        )
-
-        tags = [IGraphEdge(source=f"#{tag[1]}", target=tag[0]) for tag in tags]
-
-        backlinks = self._query(
-            "SELECT * FROM backlinks",
-            error_message=f"{self.error_message_stub} backlinks",
-        )
-
-        backlinks = [
-            IGraphEdge(source=f"{backlink[0]}", target=backlink[1])
-            for backlink in backlinks
-        ]
-
-        return tags + backlinks
-
-    def generate_graph(self):
-        graph = {"nodes": self.__get_nodes(), "edges": self.__get_edges()}
-
-        with open(f"{GRAPH_OUTPUT_DIRECTORY}/eolas-graph.json", "w") as f:
-            json.dump(graph, f, indent=4)

src/services/parse_file_service.py

@@ -2,8 +2,6 @@ import os
 from datetime import datetime
 from pathlib import Path
 
-from termcolor import colored
-
 from models.entry import IEntry
 from services.parse_markdown_service import ParseMarkdownService
 
@@ -30,7 +28,7 @@ class ParseFileService:
         }
 
     def parse_source_directory(self) -> list[IEntry]:
-        print(colored("INFO Indexing entries in source directory", "blue"))
+        print("TB-INFO Indexing entries in source directory")
         parsed_entries = []
         with os.scandir(self.source_directory) as dir:
             for file in dir:

src/services/table_service.py

@@ -1,5 +1,3 @@
-from termcolor import colored
-
 from models.entry import IEntry
 from services.sqlite_service import SqliteService
 from sql.create_tables import tables
@@ -15,7 +13,7 @@ class TableService(SqliteService):
                 table["create_statement"],
                 error_message=f"Problem creating table {table['name']}",
             )
-        print(colored("INFO Created tables", "blue"))
+        print("TB-INFO Created tables")
 
     def __drop_tables(self):
         # Reverse the order of `tables` list to avoid foreign key violation when
@@ -25,7 +23,7 @@ class TableService(SqliteService):
                 f"DROP TABLE IF EXISTS {table['name']}",
                 error_message=f"Problem truncating table {table['name']}",
             )
-        print(colored("INFO Cleared tables", "blue"))
+        print("TB-INFO Cleared tables")
 
     def __entry_exists(self, title) -> bool:
         self._execute("SELECT 1 FROM entries WHERE title = :title", {"title": title})
@@ -47,7 +45,7 @@ class TableService(SqliteService):
                         {"tag_name": tag},
                     )
 
-        print(colored("INFO Base tables populated", "blue"))
+        print("TB-INFO Base tables populated")
 
     def __populate_junction_tables(self, entries: list[IEntry]):
         for entry in entries:
@@ -70,7 +68,7 @@ class TableService(SqliteService):
                             },
                         )
 
-        print(colored("INFO Junction tables populated", "blue"))
+        print("TB-INFO Junction tables populated", "blue")
 
     def populate_tables(self, entries: list[IEntry]):
         self.__drop_tables()

src/services/tag_service.py

@@ -1,20 +0,0 @@
-import json
-
-from services.sqlite_service import SqliteService
-
-
-class TagService(SqliteService):
-    def __init__(self, db_connection):
-        super().__init__(db_connection)
-
-    def __retrieve_entries_for_tag(self, tag):
-        entries = self._query("SELECT * FROM entries_tags WHERE tag_name = ?", (tag,))
-        return sorted([entry[0] for entry in entries], key=str.lower)
-
-    def export_tags(self):
-        tags = self._query("SELECT * FROM tags")
-        tags = sorted([tag[0] for tag in tags], key=str.lower)
-        tag_dict = {}
-        for tag in tags:
-            tag_dict[tag] = self.__retrieve_entries_for_tag(tag)
-        print(json.dumps(tag_dict))