Initial commit

Bot/sqlitewrapper/__init__.py

@@ -0,0 +1 @@
+from .model import *  # noqa

Bot/sqlitewrapper/model.py

@@ -0,0 +1,309 @@
+# mypy: disable-error-code=attr-defined
+from __future__ import annotations
+from pathlib import Path
+from sqlite3 import (
+    Connection,
+    connect,
+)
+from typing import (
+    Generator,
+    Optional,
+    Tuple,
+    List,
+    Dict,
+    Any,
+)
+
+
+__all__ = (
+    'Row',
+    'Model',
+    'Datatype',
+)
+
+
+class Row:
+    def __init__(self, **attrs: Any) -> None:
+        for name, value in attrs.items():
+            self.__dict__[name] = value
+
+    def values(self) -> Tuple[Any, ...]:
+        return tuple(self.__dict__.values())
+
+    def keys(self) -> Tuple[Any, ...]:
+        return tuple(self.__dict__.keys())
+
+    def dict(self) -> Dict[Any, Any]:
+        return self.__dict__
+
+    def items(self) -> Any:
+        return self.__dict__.items()
+
+    def __str__(self) -> str:
+        return f"<Row{self.__dict__}>"
+
+    def __repr__(self) -> str:
+        return str(self)
+
+    def __iter__(self) -> Row:
+        self.__n = 0
+        return self
+
+    def __next__(self) -> Any:
+        keys = tuple(self.__dict__.keys())[:-1]  # Remove the `self.__n`
+        if self.__n == len(keys):
+            raise StopIteration
+        data = keys[self.__n]
+        self.__n += 1
+        return data
+
+    def __getitem__(self, __k: Any) -> Any:
+        return self.__dict__[__k]
+
+
+class Datatype:
+    ID = 'INTEGER PRIMARY KEY'
+    NULL = None
+    INT = 'INTEGER'
+    REAL = 'REAL'
+    STR = 'TEXT'
+    BLOB = 'BLOB'
+
+
+class ConnectionManager:
+    def __init__(self, db: str) -> None:
+        self.db = db
+        self._connection = connect(self.db)
+
+    def __enter__(self) -> Connection:
+        return self._connection
+
+    def __exit__(self, *args: Any) -> None:
+        self._connection.commit()
+        self._connection.close()
+
+
+class Model:
+    def __init__(self, db_name: str, save_path: Path, **table: Any) -> None:
+        self.name = db_name
+        self.path = str(Path(f"{save_path}/.{db_name}.sqlite"))
+        self.table = table
+        self.table['id'] = Datatype.ID
+
+        self.table_values = ' '.join(
+            f"{name} {datatype}," for (name, datatype) in table.items()
+        )[:-1]
+
+    def __str__(self) -> str:
+        data = list(self.fetch_all())
+        return f"{self.__class__.__name__}{data}"
+
+    def __repr__(self) -> str:
+        return str(self)
+
+    def __hash__(self) -> int:
+        return hash(self.path)
+
+    def _get_conditions(self, **where: Any) -> str:
+        keys = tuple(where.keys())
+
+        condition = ""
+        for index, key in enumerate(keys):
+            condition += f"{key} = ?"
+            if index != len(keys) - 1:
+                condition += " AND "
+
+        return condition
+
+    def execute(self, query: str, values: Optional[Tuple[Row, ...]] = None) -> Any:
+        """Execute a query
+
+        :param query: An SQL Query
+        :type query: str
+        :param values: vales to be added, if any, defaults to None
+        :type values: Optional[Tuple[Row, ...]], optional
+        :raises Exception: If tha database has not been initialized
+                            before trying to execute any queries
+        :return: Whatever the query would return
+        :rtype: Any
+        """
+        with ConnectionManager(self.path) as cur:
+            if values is None:
+                data = cur.execute(query)
+            else:
+                data = cur.execute(query, values)
+            return data.fetchall()
+
+    def init(self) -> None:
+        """Create a table based on the `self.table` (**table) kwargs
+        provided upon initialization
+        """
+        query = f"""
+        CREATE TABLE IF NOT EXISTS {self.name} (
+            {self.table_values}
+        )
+        """
+        self.execute(query)
+
+    def save(self, row: Row) -> None:
+        """Save a row into the db. Example:
+        ```
+            >>> row = Row(name='Pantelis', age=13)
+            >>> self.save(row)
+        ```
+
+        :param row: A row object
+        :type row: Row
+        :raises ValueError: If the Row values does not match the db schema
+        """
+        fields = self.table
+        #              - 1 for the id field
+        if len(fields) - 1 != len(row.keys()):
+            raise ValueError(f"Row fields {row.keys()} do not much db schema\
+ {tuple(self.table.keys())[:-1]}. Consider adding 'Datatype.NULL' for the missing fields")
+
+        marks = []
+        for _ in row.values():
+            marks.append('?')
+
+        query = f"""
+        INSERT INTO {self.name} {row.keys()}
+        VALUES (
+            {", ".join(marks)}
+        )
+        """
+        self.execute(query, row.values())
+
+    def delete(self, **where: Any) -> None:
+        """Delete a row from the db. Example:
+        ```
+            >>> # Query: `DELETE FROM {self.name} WHERE name = ? AND age = ?`
+            >>> # This will delete every row with name='John' and age=15
+            >>> self.delete(name='John')
+        ```
+        """
+        values = tuple(where.values())
+        condition = self._get_conditions(**where)
+
+        query = f"""
+        DELETE FROM {self.name}
+        WHERE
+            {condition}
+        """
+        self.execute(query, values)
+
+    def edit(self, row: Row) -> None:
+        """After you picked and changed a row, use this instead of `save` in order
+        for the entry to preserver the same `id`. Example:
+        ```
+            >>> row = self.get(name='john')
+            >>> row.name = 'Mary'
+            >>> self.edit(row)
+        ```
+
+        :param row: _description_
+        :type row: Row
+        """
+        id = row.id
+        self.delete(id=id)
+
+        marks = []
+        for _ in row.values():
+            marks.append('?')
+
+        query = f"""
+        INSERT INTO {self.name} {row.keys()}
+        VALUES (
+            {", ".join(marks)}
+        )
+        """
+
+        self.execute(query, row.values())
+
+    def _entries_as_rows(self, data: List[Any]) -> List[Row]:
+        """Take a list of entries and convert it to a list of `Row`s
+
+        :param data: The list of entries
+        :type data: List[Any]
+        :return: A copy of the data as list or `Row`s
+        :rtype: List[Row]
+        """
+        # rows = [
+        #     <Row{'name': 'Pantelis', 'age': 12, 'id': 1}>,
+        #     <Row{'name': 'Pantelis', 'age': 12, 'id': 2}>,
+        # ]
+        table_keys = tuple(self.table.keys())
+        rows = []
+
+        for row in data:
+            struct = {}
+            for index, col in enumerate(row):
+                struct[table_keys[index]] = col
+            rows.append(Row(**struct))
+            struct.clear()
+
+        return rows
+
+    def fetch_all(self) -> Generator[Row, None, None]:
+        query = f"SELECT * FROM {self.name}"
+        data = self.execute(query)
+
+        rows = self._entries_as_rows(data)
+        yield from rows
+
+    def filter(self, **where: Any) -> Generator[Row, None, None]:
+        """Filter out data from the db based on the `where` conditions. Example
+        ```
+            >>> data = self.filter(name='Pantelis', age=13)
+            >>> # Query created
+            >>> # SELECT * FROM test WHERE name = Pantelis AND age = 13
+            >>> for i in data:
+            ...     i
+            <Row{...}>
+        ```
+
+        :yield: Row
+        :rtype: Generator[Row, None, None]
+        """
+        # cursor.execute("SELECT * FROM my_table WHERE name = ? AND age = ?", (name, age))
+        values = tuple(where.values())
+        condition = self._get_conditions(**where)
+
+        query = f"""
+        SELECT * FROM {self.name}
+        WHERE
+            {condition}
+        """
+
+        data = self.execute(query, values)
+        rows = self._entries_as_rows(data)
+        yield from rows
+
+    def get(self, **where: Any) -> Row:
+        """Find the first occurance matching the `where` condition(s) Example:
+        ```
+            >>> self.get(name="Pantelis", age=12)
+            <Row{...}>
+        ```
+
+        :return: A `Row` with the values of the matching row
+        :rtype: Row
+        """
+        values = tuple(where.values())
+
+        condition = self._get_conditions(**where)
+
+        query = f"""
+        SELECT * FROM {self.name}
+        WHERE
+            {condition}
+        """
+        data = self.execute(query, values)[0]
+        row = {}
+        for value, name in zip(data, tuple(self.table.keys())):
+            row[name] = value
+        return Row(**row)
+
+
+if __name__ == '__main__':
+    pass

Bot/sqlitewrapper/tests.py

@@ -0,0 +1,83 @@
+# mypy: disable-error-code=attr-defined
+import unittest
+import os
+from pathlib import Path
+from .model import (
+    Row,
+    Model,
+    Datatype
+)
+
+
+class TestRow(unittest.TestCase):
+    def setUp(self) -> None:
+        return super().setUp()
+
+    def tearDown(self) -> None:
+        return super().tearDown()
+
+    def test_init(self) -> None:
+        name, age = 'Mary', 14
+        row = Row(name=name, age=age)
+        result = row.values()
+        self.assertEqual(result, (name, age))
+
+
+class TestModel(unittest.TestCase):
+    def setUp(self) -> None:
+        self.base_dir = Path(__file__).parent
+        self.name = 'testdb'
+        self.db = Model(
+            self.name,
+            self.base_dir,
+            name=Datatype.STR,
+            age=Datatype.INT,
+        )
+        self.db.init()
+        return super().setUp()
+
+    def tearDown(self) -> None:
+        os.remove(self.db.path)
+        return super().tearDown()
+
+    def test_init(self) -> None:
+        self.assertTrue(os.path.exists(self.db.path))
+
+    def test_save(self) -> None:
+        name, age = 'John', 14
+        self.db.save(Row(name=name, age=age))
+        self.db.get(name=name, age=age)  # This must not raise an Exception
+
+        with self.assertRaises(ValueError):
+            self.db.save(Row(name='test'))
+
+    def test_delete(self) -> None:
+        name, age = 'John', 14
+        self.db.save(Row(name=name, age=age))
+        self.db.delete(name=name, age=age)
+        self.assertEqual(len(tuple(self.db.fetch_all())), 0)
+
+    def test_edit(self) -> None:
+        name, age = 'John', 14
+        self.db.save(Row(name=name, age=age))
+        r = self.db.get(name=name, age=age)
+        r.name = 'Mary'
+        self.db.edit(r)
+        self.assertEqual(len(tuple(self.db.fetch_all())), 1)
+        self.db.get(name='Mary', age=age)  # This should not raise an exception
+
+    def test_filter(self) -> None:
+        data = (
+            ('John', 14),
+            ('Mary', 14),
+            ('Mary', 15),
+        )
+
+        for i in data:
+            self.db.save(Row(name=i[0], age=i[1]))
+
+        age = 14
+        filtered = self.db.filter(age=age)
+        data = list(filtered)  # type: ignore
+        self.assertTrue(all(i.age == age for i in data))  # type: ignore
+        self.assertEqual(len(data), 2)