#wip(py): SQL Server Driver.

Python/AnP/Drivers/SQLServerDriver.py

@@ -1,17 +1,154 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-from typing import Self, Any, Sequence
+from typing import Self, Any, Sequence, Optional
+from pyodbc import Connection, connect as sql_server_connect, Cursor
+from re import Match as REMatch
 from AnP.Abstracts.DatabaseAbstract import DatabaseAbstract
 from AnP.Interfaces.Application.AnPInterface import AnPInterface
+from AnP.Models.DatabaseResponseModel import DatabaseResponseModel
 from AnP.Utils.Common import Common
+from AnP.Utils.Patterns import RE
+from AnP.Utils.Checks import Check
 
 class SQLServerDriver(DatabaseAbstract):
 
     def __init__(self:Self, anp:AnPInterface, inputs:dict[str, Any|None]|Sequence[Any|None]) -> None:
         self.anp:AnPInterface = anp
-        self.__host:str = Common.get_value("host", inputs)
-        self.__port:int = Common.get_value("port", inputs)
-        self.__user:str = Common.get_value("user", inputs)
+        self.__connection:Connection|None = None
+        self.__host:str = Common.get_value("host", inputs, "localhost")
+        self.__port:int = Common.get_value("port", inputs, 1433)
+        self.__user:str = Common.get_value("user", inputs, "sa")
         self.__password:str = Common.get_value("password", inputs)
-        self.__database:str = Common.get_value("database", inputs)
+        self.__database:str = Common.get_value("database", inputs)
+        self.__driver:str = Common.get_value("driver", inputs, "{ODBC Driver 17 for SQL Server}")
+        self.__string_connection:str = Common.string_variables(Common.get_value("string_connection", inputs, "DRIVER={driver};SERVER={host},{port};UID={user};PWD={password};DATABASE={database}"), {
+            "driver" : self.__driver,
+            "host" : self.__host,
+            "port" : self.__port,
+            "user" : self.__user,
+            "password" : self.__password,
+            "database" : self.__database
+        })
+
+    def connect(self:Self) -> bool:
+        if self.__connection is None:
+            try:
+                self.__connection = sql_server_connect(self.__string_connection, autocommit = True)
+            except Exception as exception:
+                self.anp.exception(exception, "anp_sql_server_driver_connection_exception", {
+                    "driver" : self.__driver,
+                    "host" : self.__host,
+                    "port" : self.__port,
+                    "user" : self.__user,
+                    "password" : self.__password,
+                    "database" : self.__database
+                })
+                return False
+        return True
+    
+    def is_disconnected(self:Self) -> bool:
+        return self.__connection is None
+
+    def disconnect(self:Self) -> bool:
+        if self.__connection is not None:
+            try:
+                self.__connection.close()
+                self.__connection = None
+            except Exception as exception:
+                self.anp.exception(exception, "anp_sql_server_driver_disconnection_exception", {
+                    "driver" : self.__driver,
+                    "host" : self.__host,
+                    "port" : self.__port,
+                    "user" : self.__user,
+                    "password" : self.__password,
+                    "database" : self.__database
+                })
+                return False
+        return True
+    
+    def __process(self:Self, query:str, inputs:Optional[dict[str, Any|None]] = None) -> tuple[str, list[str]]:
+
+        variables:list[str] = []
+
+        def callback(matches:REMatch) -> str:
+            
+            key:str = matches.group(1)
+
+            if key is not None:
+                if key in inputs:
+
+                    value:Any|None = inputs[key]
+
+                    return (
+                        "null" if value is None else 
+                        "'" + value + "'" if Check.is_string(value) else 
+                        ("1" if value is True else "0") if Check.is_boolean(value) else 
+                        # "'" + value.strftime("%Y-%m-%d %H:%M:%S") + "'" if Check.is_date(value) else 
+                        "'" + value.isoformat(sep = " ") + "'" if Check.is_date(value) else 
+                    str(value).replace("'", "''"))
+            else:
+                
+                variable_name:str = matches.group(2)
+
+                if variable_name is not None and variable_name not in variables:
+                    variables.append(variable_name)
+                
+            return matches.group(0)
+
+        query = RE.ODBC_STRING_VARIABLE.sub(callback, query.strip())
+
+        return (
+            "\n".join("declare @" + variable + " varchar(max)" for variable in variables) + "\n\n" + 
+            query + "\n\n"  + 
+            "select " + ", ".join("@" + variable + " as [" + variable + "]" for variable in variables)
+        )
+    
+    def execute(self:Self, query:str, inputs:Optional[dict[str, Any|None]] = None) -> DatabaseResponseModel:
+
+        response:DatabaseResponseModel = DatabaseResponseModel()
+
+        self._ready()
+
+        if self.connect():
+            
+            cursor:Cursor = self.__connection.cursor()
+            variables:list[str]
+            i:int = 0
+
+            query, variables = self.__process(query, inputs)
+
+            try:
+
+                cursor.execute(query)
+
+                while True:
+                    if cursor.description is not None:
+
+                        columns:list[str] = [column[0] for column in cursor.description]
+
+                        response.add(i, [dict(zip(columns, row)) for row in cursor.fetchall()])
+                        i += 1
+
+                    if not cursor.nextset():
+                        break
+
+                self.__connection.commit()
+
+            except Exception as exception:
+                self.anp.exception(exception, "anp_sql_server_driver_execution_exception", {
+                    "driver" : self.__driver,
+                    "host" : self.__host,
+                    "port" : self.__port,
+                    "user" : self.__user,
+                    "password" : self.__password,
+                    "database" : self.__database,
+                    "query" : query
+                })
+                self.__connection.rollback()
+
+            response.set(variables)
+
+        self._unbusy()
+
+        return response

Python/AnP/Drivers/SQLiteDriver.py

@@ -28,7 +28,9 @@ class SQLiteDriver(DatabaseAbstract):
                 self.__connection = sqlite_connect(self.__path, check_same_thread = False)
 
             except Exception as exception:
-                self.anp.exception(exception, "anp_sqlite_driver_connection_exception")
+                self.anp.exception(exception, "anp_sqlite_driver_connection_exception", {
+                    "path" : self.__path
+                })
                 return False
         return True
     
@@ -40,10 +42,12 @@ class SQLiteDriver(DatabaseAbstract):
             try:
                 self.__connection.close()
                 self.__connection = None
-                return True
             except Exception as exception:
-                self.anp.exception(exception, "anp_sqlite_driver_disconnection_exception")
-        return False
+                self.anp.exception(exception, "anp_sqlite_driver_disconnection_exception", {
+                    "path" : self.__path
+                })
+                return False
+        return True
     
     def __set_variables(self:Self, query:str, inputs:Optional[dict[str, Any|None]] = None) -> str:
 
@@ -57,16 +61,16 @@ class SQLiteDriver(DatabaseAbstract):
 
                 return (
                     "null" if value is None else 
-                    "'" + value.replace("'", "''") + "'" if Check.is_string(value) else 
+                    "'" + value + "'" if Check.is_string(value) else 
                     ("1" if value is True else "0") if Check.is_boolean(value) else 
                     # "'" + value.strftime("%Y-%m-%d %H:%M:%S") + "'" if Check.is_date(value) else 
                     "'" + value.isoformat(sep = " ") + "'" if Check.is_date(value) else 
-                str(value))
+                str(value).replace("'", "''"))
             return matches.group(0)
 
         return RE.STRING_VARIABLES.sub(callback, query.strip())
     
-    def __process(self:Self, query:str, inputs:Optional[dict[str, Any|None]] = None) -> Any|None:
+    def __process(self:Self, query:str, inputs:Optional[dict[str, Any|None]] = None) -> tuple[list[str], list[str]]:
 
         sentences:list[str] = []
         variables:list[str] = []
@@ -154,6 +158,7 @@ class SQLiteDriver(DatabaseAbstract):
 
             except Exception as exception:
                 self.anp.exception(exception, "anp_sqlite_driver_execution_exception", {
+                    "path" : self.__path,
                     "i" : i, 
                     "sentence" : sentence
                 })

Python/AnP/Utils/Patterns.py

@@ -19,4 +19,5 @@ class RE:
     DOUBLE_NEW_LINE:REPattern = re_compile(r'(?:\r\n){2}|\r{2}|\n{2}')
     HTTP_HEADER_PARAMETER:REPattern = re_compile(r'([^:]+):\s*(.+)')
     HTTP_REQUEST:REPattern = re_compile(r'^([^\s]+)\s([^\s\?\#]+)(?:\?([^#]+))?(?:\#([^\s]+))?\s([^\/]+)\/([0-9\.]+)$')
-    PYTHON_MODULES_PATH:REPattern = re_compile(r'[\/\\]python[0-9\.]+[\/\\]site-packages[\/\\]?$', RE_IGNORECASE)
+    PYTHON_MODULES_PATH:REPattern = re_compile(r'[\/\\]python[0-9\.]+[\/\\]site-packages[\/\\]?$', RE_IGNORECASE)
+    ODBC_STRING_VARIABLE:REPattern = re_compile(r'\{([a-z_][a-z0-9_]*)\}|@([a-z0-9_]+)', RE_IGNORECASE)

README.md

@@ -8,7 +8,7 @@ Framework.
 #!/bin/bash
 
 apt update && apt -y upgrade && apt -y autoclean && apt -y autoremove
-pyp3 install websockets selenium selenium-wire --break-system-packages
+pyp3 install websockets selenium selenium-wire pyodbc --break-system-packages
 
 ```