Merge branch 'dataclass-use'

# Conflicts:
#	pythonicMySQL/__init__.py
#	pythonicMySQL/mysqlclient/response.py

Author: goleos

pythonicMySQL/__init__.py

@@ -0,0 +1 @@
+from pythonicMySQL.mysqlclient import CLIENT

pythonicMySQL/constructor/__init__.py

@@ -0,0 +1,3 @@
+from pythonicMySQL.constructor.table import Table
+from pythonicMySQL.constructor.columnflags import *
+

pythonicMySQL/constructor/column.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+from typing import overload, Union, List, Optional
+from dataclasses import dataclass, is_dataclass, field, fields
+from pythonicMySQL.datatypes.mysqltypes import MySQLType, INT
+
+
+ID_COLUMN = field(default=None, init=True, metadata={"name": "id", "mysql_type": INT(11), "unsigned": True,
+                                                      "primary": True, "auto_increment": True})
+
+
+@dataclass(frozen=True)
+class Column:
+
+    name: str
+    mysql_type: MySQLType
+    default: type = None
+    unsigned: bool = False
+    primary: bool = False
+    null: bool = False
+    unique: bool = False
+    auto_increment: bool = False
+
+    @property
+    def description(self) -> str:
+        query_str = f"`{self.name}` {self.mysql_type.description}"
+        if self.unsigned: query_str += " unsigned"
+        if not self.null: query_str += " NOT NULL"
+        if self.default is not None: query_str += f" Default {self.default}"
+        if self.auto_increment: query_str += " AUTO_INCREMENT"
+        if self.unique:
+            query_str += " UNIQUE"
+        if self.primary:
+            query_str += " PRIMARY KEY"
+        return query_str
+
+
+def column(mysql_type: Union[MySQLType, MySQLType.__class__], *flags: dict, default=None, unsigned: bool = False,
+           null: bool = False, unique: bool = False):
+    if isinstance(mysql_type, type):
+        mysql_type = mysql_type()
+    metadata = {
+        "mysql_type": mysql_type,
+        "default": default,
+        "unsigned": unsigned,
+        "null": null,
+        "unique": unique
+    }
+    for flag in flags:
+        metadata = {**metadata, **flag}
+    if default is not None:
+        return field(default=default, metadata=metadata)
+    else:
+        return field(metadata=metadata)
+
+
+@overload
+def columns(mysql_object: type, attribute: str) -> Optional[Column]: ...
+
+
+@overload
+def columns(mysql_object: type) -> List[Column]: ...
+
+
+def columns(mysql_object: type, attribute: Optional[str] = None) -> Union[List[Column], Column, None]:
+    if not is_dataclass(mysql_object):
+        raise ValueError("MySQL objects need to be a dataclass")
+    columns_ = []
+    if attribute is None:
+        fields_ = fields(mysql_object)
+    else:
+        fields_ = [item for item in fields(mysql_object) if item.name == attribute]
+    for field_ in fields_:
+        if "mysql_type" in dict(field_.metadata).keys():
+            metadata = {**{"name": field_.name}, **field_.metadata}
+            columns_.append(Column(**metadata))
+    if attribute is not None and len(columns_) == 1:
+        return columns_[0]
+    elif attribute is not None and len(columns_) == 0:
+        return None
+    elif attribute is None:
+        return sorted(columns_, key=lambda i: i.primary, reverse=True)
+    else:
+        raise KeyError

pythonicMySQL/constructor/columnflags.py

@@ -0,0 +1,5 @@
+UNIQUE = {"unique": True}
+PRIMARY = {"primary": True}
+NULL = {"null": True}
+UNSIGNED = {"unsigned": True}
+

pythonicMySQL/constructor/liverow.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass, field
+from pythonicMySQL.constructor.table import Table
+
+
+@dataclass
+class LinkedObject:
+
+    id: str = "wew"

pythonicMySQL/constructor/table.py

@@ -0,0 +1,184 @@
+from __future__ import annotations
+from typing import Union, TypeVar, TYPE_CHECKING, overload
+from pythonicMySQL.mysqlclient import CLIENT
+from pythonicMySQL.constructor.column import columns
+from dataclasses import is_dataclass
+
+
+T = TypeVar("T")
+
+
+class Table:
+
+    def __init__(self, database: str, table: str, python_type: T):
+        self._validate_dataclass(python_type)
+        self.python_type = python_type
+        self.database = database
+        self.table = table
+
+    @staticmethod
+    def _validate_dataclass(obj):
+        if not is_dataclass(obj):
+            raise ValueError("Object must be a dataclass")
+
+    def to_mysql_dict(self, obj, remove_id: bool = False) -> dict:
+        self._validate_dataclass(obj)
+        columns_ = columns(obj)
+        mysql_dict = {}
+        for key, value in obj.__dict__.items():
+            matching = [column for column in columns_ if column.name == key]
+            if matching:
+                mysql_dict[key] = matching[0].mysql_type.convert_to_mysql(value)
+        if remove_id:
+            id_column = [column for column in columns_ if column.primary]
+            mysql_dict.pop(id_column[0].name, '')
+        return mysql_dict
+
+    def insert(self, obj: T, update_if_exists=False, commit: bool = True):
+        if update_if_exists and len([column for column in columns(obj) if column.unique]) == 0:
+            raise AssertionError("You cannot use update_if_exists when there are no unique keys")
+        return CLIENT.insert(self.database, self.table, self.to_mysql_dict(obj),
+                             update_if_duplicate_key=update_if_exists, commit=commit)
+
+    def update(self, mysql_id: int, obj: T, commit=True):
+        return CLIENT.update(self.database, self.table, mysql_id=mysql_id, dictionary=self.to_mysql_dict(obj, True),
+                             commit=commit)
+
+    def select(self, ascending: bool = True, limit: int = None, offset: int = 0, where: Union[dict, str] = " "):
+        response = CLIENT.select(self.database, self.table,
+                                 ascending=ascending, limit=limit, offset=offset, where=where)
+        data = response.data
+        result = []
+        for row in data:
+            insertion_object = {}
+            for key, value in row.items():
+                matching = [column for column in columns(self.python_type) if column.name == key]
+                if matching:
+                    insertion_object[key] = matching[0].mysql_type.convert_to_python(value)
+            result.append(self.python_type(**insertion_object))
+        return result
+
+    @overload
+    def get_item(self, _id: int):
+        ...
+
+    @overload
+    def get_item(self, column: str, value: object):
+        ...
+
+    def get_item(self, *args):
+        """Gets first row corresponding to the search. Raises error if there are none."""
+        if isinstance(args[0], int):
+            return self.select(limit=1, where={"id": args[0]})[0]
+        else:
+            return self.select(limit=1, where={args[0]: args[1]})[0]
+
+    def __iter__(self):
+        return iter(self.select())
+
+    def __len__(self):
+        return CLIENT.get_number_of_rows(self.database, self.table)
+
+
+
+#
+
+# @overload
+# def table(_database: str, _table: str) -> object: ...
+#
+# def table(*, cls: Optional = None, mysql_connection: mysql.connector.connection.MySQLConnection = None, database: str = None,
+#           table: str = None, object_type = None,
+#           ascending: bool = True, limit: int = None, offset: int = 0, where: Union[dict, str] = " "):
+#
+#     def wrap(cls):
+#
+#         setattr(cls, "teuuw", 4)
+#         cls.hi = lambda : print("hello")
+#         return cls
+#
+#     return wrap
+#
+#
+# @table
+# class Songs:
+#
+#     def __init__(self):
+#         self.h = 3
+#
+#
+# song = Songs()
+
+#
+#
+# class Table:
+#
+#     object_type = None
+#     client = None
+#
+#     def __init__(self, mysql_connection: mysql.connector.connection.MySQLConnection,
+#                  database: str, table: str, object_type: MySQLObject.__class__, ascending: bool = True, limit: int = None, offset: int = 0,
+#                  where: Union[dict, str] = " "):
+#         self.ascending = ascending
+#         self.limit = limit
+#         self.offset = offset
+#         self.where = where
+#         self.__class__.client = Client(mysql_connection)
+#         self.client = self.__class__.client
+#         self.__class__.object_type = object_type
+#         self.table = table
+#         self.database = database
+#
+#     def validate_columns(self):
+#         mysql_description = self.client.describe_columns(self.database, self.table).fetchall()
+#         assigned_columns = self.__class__.object_type.COLUMNS + Column.id_column()
+#         for desc in mysql_description:
+#             if desc["Field"] not in [column.name for column in assigned_columns]:
+#                 raise ValueError("Columns do not match")
+#
+#     def create(self):
+#         columns = [Column.id_column()] + self.object_type.COLUMNS
+#         descriptions = [column.description for column in columns]
+#         return self.client.create_table(self.database, self.table, descriptions)
+#
+#     def update_columns(self):
+#         columns = [Column.id_column()] + self.object_type.COLUMNS
+#         descriptions = [column.description for column in columns]
+#         self.__class__.client.update_table_columns(self.database, self.table, descriptions)
+#
+#     def get(self) -> List[MySQLObject]:
+#         result = self.client.select(self.database, self.table,
+#                                     ascending=self.ascending, limit=self.limit, offset=self.offset, where=self.where)
+#         response = SQLResponse(result)
+#         data = response.data
+#         result = []
+#         for item in data:
+#             id_ = item.pop(Column.ID_COLUMN_NAME)
+#             obj = self.__class__.object_type(**item)
+#             obj.mysql_row_id = id_
+#             result.append(obj)
+#         return result
+#
+#     def insert(self, obj: MySQLObject, update_if_duplicate_key: bool = True, commit=True) -> OneDMLSQLResponse:
+#         cursor = self.client.insert(self.database, self.table, obj.as_mysql_dict(),
+#                                     update_if_duplicate_key=update_if_duplicate_key, commit=commit)
+#         return OneDMLSQLResponse(cursor)
+#
+#     def update(self, obj: MySQLObject, commit=True) -> OneDMLSQLResponse:
+#         cursor = self.client.update(self.database, self.table, obj.mysql_row_id, obj.as_mysql_dict(), commit=commit)
+#         response = OneDMLSQLResponse(cursor)
+#         return response
+#
+#     def __getitem__(self, item: Union[tuple, int]) -> MySQLObject:
+#         if isinstance(item, tuple) and len(item) == 2:
+#             result = self.client.select(self.database, self.table, where={item[0]: item[1]})
+#         elif isinstance(item, int):
+#             result = self.client.select(self.database, self.table, where={"id": item})
+#         else:
+#             raise KeyError
+#         if len(result) == 1:
+#             return result[0]
+#         else:
+#             raise KeyError
+#
+#     def __iter__(self):
+#         return iter(self.get())

pythonicMySQL/constructor/tablecreation.py

@@ -0,0 +1,13 @@
+from pythonicMySQL.mysqlclient import CLIENT
+from pythonicMySQL.constructor.column import Column
+from typing import List
+
+
+def create_table(database, table, columns: List[Column], create_database=False):
+    descriptions = [column.description for column in columns]
+    CLIENT.create_table(database, table, column_descriptions=descriptions, create_database=create_database)
+
+
+def update_table(database, table, columns: List[Column]):
+    descriptions = [column.description for column in columns]
+    CLIENT.update_table_columns(database, table, column_descriptions=descriptions)

pythonicMySQL/datatypes/__init__.py

@@ -0,0 +1,2 @@
+from pythonicMySQL.datatypes.mysqltypes import *
+from pythonicMySQL.datatypes.customtypes import *

pythonicMySQL/datatypes/customtypes.py

@@ -0,0 +1,28 @@
+from pythonicMySQL.datatypes.mysqltypes import MySQLType
+import datetime
+
+
+class BOOLEAN(MySQLType):
+
+    def __init__(self):
+        super(BOOLEAN, self).__init__("tinyint", length=1, python_type=bool)
+
+    def convert_to_mysql(self, python_value) -> object:
+        return 1 if python_value is True else 0
+
+    def convert_to_python(self, mysql_value) -> object:
+        if not isinstance(mysql_value, int):
+            raise ValueError
+        return bool(mysql_value)
+
+
+class TIMEDELTA(MySQLType):
+
+    def __init__(self):
+        super(TIMEDELTA, self).__init__("decimal", length="10,3", python_type=datetime.timedelta)
+
+    def convert_to_python(self, mysql_value) -> datetime.timedelta:
+        return datetime.timedelta(seconds=float(mysql_value))
+
+    def convert_to_mysql(self, python_value: datetime.timedelta) -> float:
+        return python_value.seconds