Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 78 additions & 0 deletions Projects/table_tracker/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
<h1 align="center"> TableTracker</h1>

<p align="center">
<img src="https://github.com/Musa-Sina-Ertugrul/DBScanner/assets/102359522/1ea4b501-898f-4b57-8a7e-90e853d50cdd">
</p>

<h3 align="center"> Our Purpose</h3>

<p align="center">
<img src="https://github.com/Musa-Sina-Ertugrul/TableTracker/assets/102359522/cd2e98f0-083e-44f4-a532-e4e9b9a29c47">
</p>


# TableTracker

> TableTracker is a desktop application developed in Python that facilitates tracking and managing SQLite database tables. This application allows you to execute SQL queries on SQLite databases, visualize the results, and edit your queries.
* Assigned by Asc. Prof. Dr. Bora CANBULA

<h3 align="center">Requirements</h3>
> Python 3.11 or a newer version

<h3 align="center">Setup</h3>

> To create the necessary virtual environment in the project directory, run the following command:

```console
conda create -n TableTracker python=3.11 pip -y
```

> Activate the created virtual environment:

```console
conda activate TableTracker
```

> To install required modules

```console
pip install -r requirements.txt
```

<h3 align="center">Reformatting</h3>

> For reformatting use <b><i>black</i></b>. It reformat for pep8 as same as pylint but better !!!

```console
black .
```

<h3 align="center">Run App</h3>

> To start the application, run the following command:

```console
python table_tracker
```

> .When the application starts, you can create a new SQLite database or connect to an existing one.
> Write your SQL queries in the text box and execute the query by clicking the "Execute" button.
> The results will be displayed in the "Output Window" section.

<h3 align="center">Linting</h3>

> For running <b><i>pylint</i></b>

```console
pylint ./table_tracker/ ./test/
```

<h3 align="center">Testing</h3>

> For running <b><i>unittest</i></b>

```console
python -m unittest discover -v
```
<h3 align="center">Author</h3>
> Musa Sina Ertuğrul, İrem Demir
8 changes: 8 additions & 0 deletions Projects/table_tracker/__main__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
import sys
from packages import App

sys.path.append("./tmp_files/")

if __name__ == "__main__":
app = App()
app.mainloop()
1 change: 1 addition & 0 deletions Projects/table_tracker/packages/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
from .gui import App
1 change: 1 addition & 0 deletions Projects/table_tracker/packages/events/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
from .sql_event_handler import SQLEventHandler
25 changes: 25 additions & 0 deletions Projects/table_tracker/packages/events/event_handler.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
from abc import ABCMeta
from typing import Literal
from types import NotImplementedType
from packages.utils import check_methods


class EventHandler(metaclass=ABCMeta):
"""
Abstract base class for event handling.

EventHandler class defines the interface for event handling. Subclasses must implement the handle method.
"""

__slots__: tuple = ()

@classmethod
def __subclasshook__(cls, subcls) -> NotImplementedType | Literal[True]:
"""
Check if a subclass implements the required methods.

:param subcls: The potential subclass.
:return: NotImplementedType if the method is not implemented, True otherwise.
:rtype: NotImplementedType | Literal[True]
"""
return check_methods(subcls, "handle")
192 changes: 192 additions & 0 deletions Projects/table_tracker/packages/events/sql_event_handler.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,192 @@
import sqlite3
from typing import Self
from functools import cache, reduce
from sql_formatter.core import format_sql
from sys import getsizeof
import math
from psutil import virtual_memory
import customtkinter
from .event_handler import EventHandler
from ..utils import QUERY_ERROR_NONE_OBJECT


class SQLEventHandler(EventHandler):
"""Handler for SQL events."""

MAX_PAGE_COUNT: int = 10
queries: list[str] = []
query_index: int = 0

@cache
def __new__(cls, *args, **kwargs) -> Self:
"""
Create a new instance of SQLEventHandler class.
Implements the Singleton pattern.
"""
return super().__new__(cls)

def __init__(
self, query: str, cursor: sqlite3.Cursor, result_label: customtkinter.CTkLabel
) -> None:
"""
Initialize the SQLEventHandler instance.
If the singleton query has not been created, it initializes the attributes accordingly.

:param query: The SQL query.
:type query: str
:param cursor: The cursor for executing the query.
:type cursor: sqlite3.Cursor
:param result_label: The label to display the result.
:type result_label: customtkinter.CTkLabel
"""
if not hasattr(self, "_query"):
self._query: str = query
self._cursor: sqlite3.Cursor = cursor
self._result_label: customtkinter.CTkLabel = result_label
self.__col_len: int = 0
self.__row_len: int = 0
self.__total_size: int = 0
self.queries.append(format_sql(query, max_len=1000000000))
self.query_index += 1

@property
def get_query(self) -> str:
"""
Get the SQL query.

:return: The SQL query. otherwise None.
:rtype: str| None
"""
if not sqlite3.complete_statement(self._query):
return QUERY_ERROR_NONE_OBJECT
return self._query

@property
def get_query_result_itr(self) -> sqlite3.Cursor | None:
"""

Get the iterator for executing the SQL query.

:return: Iterator for executing the SQL query. Returns None if there is an error.
:rtype: sqlite3.Cursor or None
:raise KeyError: Raised in a specific error condition.
"""
try:
return self._cursor.execute(self.get_query)
except (sqlite3.ProgrammingError, AttributeError) as error:
print(error)
return QUERY_ERROR_NONE_OBJECT

@staticmethod
def _sizeof_row(row: list[tuple]) -> int:
"""
Calculate the size of a row in bytes.

:param row: List of tuples representing a row.
:type row: list[tuple]
:return: Size of the row in bytes.
:rtype: int
"""
return reduce(getsizeof, [str(atr) for atr in row])

@property
def row_len(self) -> int:
"""
Get the length of the rows.

If the row length attribute is not set, it returns the length calculated from the result set.

:return: Length of the rows.
:rtype: int
"""
return self.__row_len or len(self)

@cache
def __len__(self) -> int:
"""
Get the total number of rows in the result set.

This method iterates through the result set obtained from the query execution and counts the rows. It also
calculates the total size of the result set.

:return: Total number of rows in the result set.
:rtype: int
:raise KeyError: Raised in a specific error condition.
"""
try:
row_itr: sqlite3.Cursor = self.get_query_result_itr
row: list[tuple] = next(row_itr)
self.__col_len = len(row)
self.__total_size += self._sizeof_row(row)
for row_count, row in enumerate(row_itr):
self.__total_size += self._sizeof_row(row)
self.__row_len = row_count + 1
return self.__row_len
except (sqlite3.ProgrammingError, StopIteration, TypeError) as error:
return 0

@property
@cache
def col_len(self) -> int:
"""
Get the total number of columns in the result set.

This method returns the number of columns in the result set.

:return: Total number of columns in the result set.
:rtype: int

"""
return self.__col_len

@property
def divaded_itrs(self) -> tuple[sqlite3.Cursor]:
"""
Divide the result set into multiple cursors.

This method divides the result set into multiple cursors based on the available memory and the size of the result set.
It calculates the number of pages and rows per page, then creates cursors accordingly.

:return: Tuple of cursors representing the divided result set.
:rtype: tuple[sqlite3.Cursor]
:raise KeyError: Raised in a specific error condition.
"""
len(self)
avaible_memory: int = int(virtual_memory()[1])

page_count: int = max(
min(
math.ceil(self.__total_size // (avaible_memory / 4.0)),
self.MAX_PAGE_COUNT,
),
1,
)
row_count: int = self.__total_size // page_count

try:
itrs: sqlite3.Cursor = [
self.get_query_result_itr for _ in range(page_count)
]
except TypeError:
return QUERY_ERROR_NONE_OBJECT

try:
for current_page, itr in enumerate(itrs[1:], 1):
for _ in range(current_page * row_count):
next(itr)
except IndexError:
return itrs

return itrs

def handle(self) -> tuple[sqlite3.Cursor]:
"""
Handle the SQL query result.

This method retrieves the divided iterators of the SQL query result using the `divaded_itrs` method and returns them.

:return: Tuple of cursors representing the divided result set.
:rtype: tuple

"""
return self.divaded_itrs
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
from .syntax_error import SytanxErrorHandler
from .text_coloring import TextColoringHandler
from .format_text import FormatTextHandler
Loading