Table Repository Added

.gitignore

@@ -0,0 +1,5 @@
+
+mylib.db
+databaseClasses/__pycache__
+tables/__pycache__
+services/__pycache__

.vscode/launch.json

@@ -0,0 +1,16 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python: Current File",
+            "type": "python",
+            "request": "launch",
+            "program": "app.py",
+            "console": "integratedTerminal",
+            "justMyCode": true
+        }
+    ]
+}

app.py

@@ -0,0 +1,18 @@
+from services.customerService import CustomerService
+from tables.customer import Customer
+
+customerService = CustomerService()
+customer = Customer()
+customer.customerName = "Test"
+customer.contactLastName = "Test"
+customer.contactFirstName = "Test"
+customer.phone = "Test"
+customer.addressLine1 = "Test"
+customer.addressLine2 = "Test"
+customer.city = "Test"
+customer.state = "Test"
+customer.postalCode = "Test"
+customer.country = "Test"
+customer.creditLimit = 1
+
+customerService.addCustomer(customer,"Murphy")

databaseClasses/column.py

@@ -0,0 +1,40 @@
+from enum import Enum
+
+class ColumnDataType(Enum):
+    id = 1,
+    idIncrement = 2,
+    idFloat = 3,
+    idString = 4,
+    string = 5,
+    dateTime = 6,
+    bool = 7,
+    int = 8,
+    float = 9,
+    blob = 10,
+
+
+class Column:
+    name = str()
+    dataType = ColumnDataType
+    nullable = bool()
+
+    def __init__(self, name : str, dataType : ColumnDataType, nullable : bool = False):
+        self.name = name
+        self.dataType = dataType
+        self.nullable = nullable
+
+    def isPrimaryKey(self):
+        return self.dataType == ColumnDataType.id or self.dataType == ColumnDataType.idIncrement or self.dataType == ColumnDataType.idFloat or self.dataType == ColumnDataType.idString
+
+    def isIncrement(self):
+        return self.dataType == ColumnDataType.idIncrement 
+
+    def getSqlValue(self,item):
+        temp = ""
+        keys = vars(item).keys()
+        if(self.name in keys):
+            temp = vars(item)[self.name]
+        if(self.dataType in [ColumnDataType.string,ColumnDataType.idString]):
+            temp = f"'{temp}'"
+        return temp
+

databaseClasses/database.py

@@ -0,0 +1,56 @@
+import mysql.connector
+from databaseClasses.table import Table
+from databaseClasses.query import Query
+from tables.customer import Customer
+from tables.employee import Employee
+from tables.office import Office
+from tables.orderdetail import OrderDetail
+from tables.order import Order
+from tables.payment import Payment
+from tables.productline import ProductLine
+from tables.product import Product
+
+
+def dict_factory(cursor, row):
+    d = {}
+    for idx, col in enumerate(cursor.description):
+        d[col[0]] = row[idx]
+    return d
+
+class Database :
+    __con : str = None
+    tables : list[Table] = [
+        Customer(),
+        Employee(),
+        Office(),
+        OrderDetail(),
+        Order(),
+        Payment(),
+        ProductLine(),
+        Product(),
+    ]
+    DataBaseName = "classicmodels"
+    host="localhost"
+    user="root"
+    password="123456"
+    __cur : str = None
+
+
+
+    def __init__(self):
+        if(Database.__con == None):
+            Database.__con = mysql.connector.connect(host=self.host,user=self.user,password=self.password)
+            Database.__cur = Database.__con.cursor()
+            Database.__con.row_factory = dict_factory
+            Database.__cur.execute(f"use {Database.DataBaseName}")
+
+
+
+    def executeQuery(self,query : str = "",commit : bool = False):
+        Database.__cur.execute(query)
+        if(commit):
+            Database.__con.commit()
+        res = [dict_factory(Database.__cur, row) for row in Database.__cur.fetchall()]
+        return res
+
+

databaseClasses/query.py

@@ -0,0 +1,79 @@
+from databaseClasses.table import Table
+
+class Query:
+
+    def listQuery(table : Table,where : str = "",order: str = ""):
+        query = f"SELECT * FROM {table.tableName}"
+        if(len(where) > 0):
+            query += f" WHERE {where}"
+        if(len(order) > 0):
+            query += f" ORDER BY {order}"
+        return query
+
+    def pageQuery(table : Table,where : str = "",order: str = "",skip : int = 0,take : int = 10):
+        query = f"SELECT * FROM {table.tableName}"
+        if(len(where) > 0):
+            query += f" WHERE {where}"
+        if(len(order) > 0):
+            query += f" ORDER BY {order}"
+        query += f" LIMIT {skip},{take}"
+        return query
+
+    def insertQuery(table : Table):
+        query = f"INSERT INTO {table.tableName}("
+        for column in table.columns:
+            if(column.isIncrement() == False):
+                query += f"{column.name},"
+        query = query[:-1]
+        query += ") VALUES("
+        for column in table.columns:
+            if(column.isIncrement() == False):
+                query += f"{column.getSqlValue(table)},"
+        query = query[:-1]
+        query += ")"
+        return query
+
+    def updateQuery(table : Table):
+        query = f"UPDATE {table.tableName} SET "
+        for column in table.columns:
+            if(column.isPrimaryKey() == False):
+                query += f"{column.name} = {column.getSqlValue(table)},"
+        query = query[:-1]
+        primaryColumn = next(filter(lambda c: c.isPrimaryKey(),table.columns))
+        query += f" WHERE {primaryColumn.name} = '{primaryColumn.getSqlValue(table)}'"
+        return query
+
+    def insertUpdateQuery(table : Table):
+        query = f"INSERT OR REPLACE INTO {table.tableName}("
+        for column in table.columns:
+            if(column.isIncrement() == False):
+                query += f"{column.name},"
+        query = query[:-1]
+        query += ") VALUES("
+        for column in table.columns:
+            if(column.isIncrement() == False):
+                query += f"{column.getSqlValue(table)},"
+        query = query[:-1]
+        query += ")"
+        return query
+
+    def deleteQuery(table : Table,id):
+        query = f"DELETE FROM {table.tableName}"
+        primaryColumn = next(filter(lambda c: c.isPrimaryKey(),table.columns))
+        query += f" WHERE {primaryColumn.name} = '{id}'"
+        return query
+
+    def deleteAllQuery(table : Table):
+        query = f"DELETE FROM {table.tableName}"
+        return query
+
+    def deleteWithConditionQuery(table : Table,where : str):
+        query = f"DELETE FROM {table.tableName}"
+        query += f" WHERE {where}"
+        return query
+
+    def countQuery(table : Table,where : str = ""):
+        query = f"SELECT COUNT(*) AS count FROM {table.tableName}"
+        if(len(where) > 0):
+            query += f" WHERE {where}"
+        return query

databaseClasses/table.py

@@ -0,0 +1,14 @@
+from databaseClasses.column import Column
+
+class Table:
+    tableName : str()
+    columns : list[Column]
+    def __init__(self,tableName : str = ""):
+        self.tableName = tableName if len(tableName ) > 0  else type(self).__name__
+        self.__getColumns()
+
+    def __getColumns(self):
+        self.columns = list()
+        for c in type(self).__annotations__.keys():
+            temp = Column(vars(type(self).__annotations__[c])['name'],vars(type(self).__annotations__[c])['dataType'],vars(type(self).__annotations__[c])['nullable']) 
+            self.columns.append(temp)

databaseClasses/tableRepository.py

@@ -0,0 +1,103 @@
+from databaseClasses.table import Table
+from databaseClasses.database import Database
+from databaseClasses.query import Query
+
+class TableRepository:
+    table : Table
+    def __init__(self, table:Table):
+        self.table = table
+
+    def __convertData(self,data):
+        list : list[type(self)] = []
+        for item in data:
+            temp = type(self)()
+            for column in temp.table.columns:
+                setattr(temp,column.name,item[column.name])
+            list.append(temp)
+        return list
+
+    def getListDictionary(self,where : str = "",order: str = ""):
+        database = Database()
+        query = Query.listQuery(self.table,where,order)
+        result = database.executeQuery(query)
+        return result
+
+    def getList(self,where : str = "",order: str = ""):
+        return self.__convertData(self.getListDictionary(where,order))
+
+    def getPageDictionary(self,where : str = "",order: str = "",skip : int = 0,take : int = 10):
+        database = Database()
+        query = Query.pageQuery(self.table,where,order,skip,take)
+        result = database.executeQuery(query)
+        return result
+
+    def getPage(self,where : str = "",order: str = "",skip : int = 0,take : int = 10):
+        return self.__convertData(self.getPageDictionary(where,order,skip,take))
+
+    def getById(self,id):
+        primaryColumn = next(filter(lambda c: c.isPrimaryKey(),self.table.columns))
+        list = self.getList(f"'{primaryColumn.name}' = '{id}'")
+        if(len(list) > 0):
+            return list[0]
+
+    def getFirst(self,where : str = "",order: str = ""):
+        list = self.getPage(where,order,0,1)
+        if(len(list) > 0):
+            return list[0]
+
+    def getCount(self,where : str = ""):
+        database = Database()
+        query = Query.countQuery(self.table,where)
+        result = database.executeQuery(query)
+        return result[0]["count"]
+
+    def add(self,table : Table):
+        database = Database()
+        query = Query.insertQuery(table)
+        database.executeQuery(query,True)
+
+    def update(self,table : Table):
+        database = Database()
+        query = Query.updateQuery(table)
+        database.executeQuery(query,True)
+
+    def addOrUpdate(self,table : Table):
+        database = Database()
+        query = Query.insertUpdateQuery(table)
+        database.executeQuery(query,True)
+
+    def addRange(self,list : list[Table]):
+        for item in list:
+            self.add(item)
+
+    def updateRange(self,list : list[Table]):
+        for item in list:
+            self.update(item)
+
+    def addOrUpdateRange(self,list : list[Table]):
+        for item in list:
+            self.addOrUpdate(item)
+
+    def delete(self,id):
+        database = Database()
+        query = Query.deleteQuery(self.table,id)
+        database.executeQuery(query,True)
+
+    def deleteRange(self,list : list[str]):
+        for item in list:
+            self.delete(item)
+
+    def deleteWithCondition(self,where : str):
+        database = Database()
+        query = Query.deleteWithConditionQuery(self.table,where)
+        database.executeQuery(query,True)
+
+    def deleteAll(self):
+        database = Database()
+        query = Query.deleteAllQuery(self.table)
+        database.executeQuery(query,True)
+
+    def rawQuery(self,query:str,commit:bool):
+        database = Database()
+        result = database.executeQuery(query,commit)
+        return result

services/customerService.py

@@ -0,0 +1,28 @@
+from databaseClasses.tableRepository import TableRepository
+from tables.customer import Customer
+from tables.employee import Employee
+from services.employeeService import EmployeeService
+import random
+import string
+
+
+class CustomerService(TableRepository):
+    employeeService : EmployeeService 
+    def  __init__(self):
+        CustomerService.employeeService = EmployeeService()
+        super().__init__(Customer())
+
+    def __getNewCustomerNumber(self):
+        temp = self.getFirst(order="customerNumber desc")
+        if(temp == None):
+            return 1
+        else:
+            return temp.customerNumber + 1
+
+    def addCustomer(self,data:Customer,employeeName:str):
+        data.customerNumber = self.__getNewCustomerNumber()
+        employee  = CustomerService.employeeService.searchByName(employeeName)
+        if(employee == None):
+            raise Exception("Employee not found")
+        data.salesRepEmployeeNumber = employee.employeeNumber
+        self.add(data)

services/employeeService.py

@@ -0,0 +1,16 @@
+from databaseClasses.tableRepository import TableRepository
+from tables.employee import Employee
+import random
+import string
+
+
+class EmployeeService(TableRepository):
+    def  __init__(self):
+        super().__init__(Employee())
+
+    def searchByName(self,name:str):
+        temp = self.getFirst(where=f"concat(firstName, ' ', lastName) like '%{name}%'")
+        return temp
+
+
+