Add sql server support

docs/src/guide/install.md

@@ -60,3 +60,20 @@ Or, as a convenience:
 ```
 pip install mayim[sqlite]
 ```
+
+## SQL Server
+
+Dependencies:
+- [pyodbc](https://github.com/mkleehammer/pyodbc)
+
+Either install it independently:
+
+```
+pip install pyodbc
+```
+
+Or, as a convenience:
+
+```
+pip install mayim[sqlserver]
+```

setup.cfg

@@ -42,5 +42,7 @@ mysql =
     asyncmy
 sqlite =
     aiosqlite
+sqlserver =
+    pyodbc
 [options.packages.find]
 where=src

src/mayim/__init__.py

@@ -10,6 +10,8 @@
 from .sql.postgres.interface import PostgresPool
 from .sql.sqlite.executor import SQLiteExecutor
 from .sql.sqlite.interface import SQLitePool
+from .sql.sqlserver.executor import SQLServerExecutor
+from .sql.sqlserver.interface import SQLServerPool
 
 __version__ = version("mayim")
 
@@ -24,6 +26,8 @@
     "MysqlPool",
     "PostgresExecutor",
     "SQLiteExecutor",
+    "SQLServerExecutor",
     "PostgresPool",
     "SQLitePool",
+    "SQLServerPool",
 )

src/mayim/sql/sqlserver/executor.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, Sequence
+
+from mayim.sql.sqlserver.query import SQLServerQuery
+
+from ..executor import SQLExecutor
+
+try:
+    import pyodbc  # noqa
+
+    SQLSERVER_ENABLED = True
+except ModuleNotFoundError:
+    SQLSERVER_ENABLED = False
+
+
+class SQLServerExecutor(SQLExecutor):
+    """Executor for interfacing with a SQL Server database"""
+
+    ENABLED = SQLSERVER_ENABLED
+    QUERY_CLASS = SQLServerQuery
+    POSITIONAL_SUB = r"?"
+    KEYWORD_SUB = r":\2"
+
+    async def _run_sql(
+        self,
+        query: str,
+        name: str = "",
+        as_list: bool = False,
+        no_result: bool = False,
+        posargs: Optional[Sequence[Any]] = None,
+        params: Optional[Dict[str, Any]] = None,
+    ):
+        method_name = self._get_method(as_list=as_list)
+        async with self.pool.connection() as conn:
+            exec_values = list(posargs) if posargs else params
+            cursor = conn.execute(query, exec_values or [])
+
+            columns = [column[0] for column in cursor.description]
+            if no_result:
+                return None
+
+            raw = getattr(cursor, method_name)()
+
+            if not as_list:
+                return dict(zip(columns, raw))
+
+            results = []
+            for row in raw:
+                results.append(dict(zip(columns, row)))
+
+            return results

src/mayim/sql/sqlserver/interface.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import Optional
+from urllib.parse import parse_qs, urlparse
+
+from mayim.base.interface import BaseInterface
+from mayim.exception import MayimError
+
+try:
+    import pyodbc
+
+    SQLSERVER_ENABLED = True
+except ModuleNotFoundError:
+    SQLSERVER_ENABLED = False
+
+
+class SQLServerPool(BaseInterface):
+    """Interface for connecting to a SQL Server database"""
+
+    scheme = "mssql+pyodbc"
+
+    def __init__(self, db_path: str):
+        self._db_path = db_path
+        super().__init__()
+
+    def _setup_pool(self):
+        if not SQLSERVER_ENABLED:
+            raise MayimError(
+                "SQL Server driver not found. Try reinstalling Mayim: "
+                "pip install mayim[sqlserver]"
+            )
+
+    def _parse_url(self) -> str:
+        url = urlparse(self._db_path)
+        query = parse_qs(url.query)
+        driver = query.get("DRIVER", "")
+        conn_string = (
+            f"DRIVER={driver};SERVER={url.hostname};PORT={url.port};"
+            f"DATABASE={url.path.strip('/')};"
+            f"UID={url.username};PWD={url.password};"
+        )
+        return conn_string
+
+    async def open(self):
+        """Open connections to the pool"""
+        conn_string = self._parse_url()
+        self._db = pyodbc.connect(conn_string)
+
+    async def close(self):
+        """Close connections to the pool"""
+        self._db.close()
+
+    @asynccontextmanager
+    async def connection(self, timeout: Optional[float] = None):
+        """Obtain a connection to the database
+
+        Args:
+            timeout (float, optional): _Not implemented_. Defaults to `None`.
+
+        Returns:
+            AsyncIterator[Connection]: Iterator that will yield a connection
+
+        Yields:
+            Iterator[AsyncIterator[Connection]]: A database connection
+        """
+        if not self._db:
+            await self.open()
+
+        yield self._db

src/mayim/sql/sqlserver/query.py

@@ -0,0 +1,25 @@
+import re
+
+from mayim.exception import MayimError
+from mayim.sql.query import ParamType, SQLQuery
+
+
+class SQLServerQuery(SQLQuery):
+    __slots__ = ("name", "text", "param_type")
+    PATTERN_POSITIONAL_PARAMETER = re.compile(r"\?")
+    PATTERN_KEYWORD_PARAMETER = re.compile(r"\:[a-z_][a-z0-9_]")
+
+    def __init__(self, name: str, text: str) -> None:
+        super().__init__(name, text)
+        positional_argument_exists = bool(
+            self.PATTERN_POSITIONAL_PARAMETER.search(self.text)
+        )
+        keyword_argument_exists = bool(
+            self.PATTERN_KEYWORD_PARAMETER.search(self.text)
+        )
+        if keyword_argument_exists:
+            raise MayimError("Only Positional arguments allowed in pyODBC")
+        if positional_argument_exists:
+            self.param_type = ParamType.POSITIONAL
+        else:
+            self.param_type = ParamType.NONE