sql_client.py

"""
MSSQLTips.com General database connection configuration class
"""
import os
from attrs import define, field, validators
from sqlalchemy.engine import Engine, create_engine, URL
import mssql_python
from mssql_python import Connection
import pyodbc
from dotenv import load_dotenv


load_dotenv()


@define
class RDSConfigPyODBC:
    """Pyodbc with SQLAlchemy wrapper"""
    DB_SERVER: str = field(
        factory=lambda: os.getenv("DB_SERVER", ".\\MSSQLSERVER02"),
        validator=validators.instance_of(str)
    )
    DB_DATABASE: str = field(
        factory=lambda: os.getenv("DB_DATABASE", "AdventureWorks2022"),
        validator=validators.instance_of(str)
    )
    DB_USERNAME: str = field(
        factory=lambda: os.getenv("DB_USERNAME", ""),
        validator=validators.instance_of(str)
    )
    DB_PASSWORD: str = field(
        factory=lambda: os.getenv("DB_PASSWORD", ""),
        validator=validators.instance_of(str)
    )
    DB_PORT: int = field(
        factory=lambda: int(os.getenv("DB_PORT", 0)),
        converter=int,
        validator=validators.instance_of(int)
    )
    DB_DRIVER: str = field(
        factory=lambda: os.getenv("DB_DRIVER", "mssql+pyodbc"),
        validator=validators.instance_of(str)
    )

    @property
    def url(self) -> URL:
        if not self.DB_USERNAME or not self.DB_PASSWORD:
            # Windows Authentication
            return URL.create(
                drivername=self.DB_DRIVER,
                host=self.DB_SERVER,
                database=self.DB_DATABASE,
                username=self.DB_USERNAME,
                password=self.DB_PASSWORD,
                # port=self.DB_PORT,
                query=dict(driver='ODBC Driver 18 for SQL Server',
                           Trusted_Connection='True',
                           TrustServerCertificate='yes')
            )
        else:
            # SQL Server Authentication
            return URL.create(
                drivername=self.DB_DRIVER,
                host=self.DB_SERVER,
                database=self.DB_DATABASE,
                username=self.DB_USERNAME,
                password=self.DB_PASSWORD,
                # port=self.DB_PORT,
                query=dict(driver='ODBC Driver 18 for SQL Server',
                           TrustServerCertificate='yes')
            )

    def get_engine(self) -> Engine:
        return create_engine(self.url)


@define
class RDSConfigPyODBCRaw:
    """Raw pyodbc without SQLAlchemy wrapper"""
    DB_SERVER: str = field(
        factory=lambda: os.getenv("DB_SERVER", ".\\MSSQLSERVER02"),
        validator=validators.instance_of(str)
    )
    DB_DATABASE: str = field(
        factory=lambda: os.getenv("DB_DATABASE", "AdventureWorks2022"),
        validator=validators.instance_of(str)
    )
    DB_USERNAME: str = field(
        factory=lambda: os.getenv("DB_USERNAME", ""),
        validator=validators.instance_of(str)
    )
    DB_PASSWORD: str = field(
        factory=lambda: os.getenv("DB_PASSWORD", ""),
        validator=validators.instance_of(str)
    )
    DB_DRIVER: str = field(
        factory=lambda: os.getenv("DB_ODBC_DRIVER", "ODBC Driver 18 for SQL Server"),
        validator=validators.instance_of(str)
    )

    @property
    def connection_string(self) -> str:
        if not self.DB_USERNAME or not self.DB_PASSWORD:
            # Windows Authentication
            return (f'DRIVER={{{self.DB_DRIVER}}};\
                    SERVER={self.DB_SERVER};\
                    DATABASE={self.DB_DATABASE};\
                    Trusted_Connection=yes;\
                    TrustServerCertificate=yes;')
        else:
            # SQL Server Authentication
            return (f'DRIVER={{{self.DB_DRIVER}}};\
                    SERVER={self.DB_SERVER};\
                    DATABASE={self.DB_DATABASE};\
                    UID={self.DB_USERNAME};\
                    PWD={self.DB_PASSWORD};\
                    TrustServerCertificate=yes;')

    def get_connection(self) -> pyodbc.Connection:
        return pyodbc.connect(self.connection_string)


@define
class RDSConfigMSSQLPython:
    DB_SERVER: str = field(
        factory=lambda: os.getenv("DB_SERVER", ".\\MSSQLSERVER02"),
        validator=validators.instance_of(str)
    )
    DB_DATABASE: str = field(
        factory=lambda: os.getenv("DB_DATABASE", "AdventureWorks2022"),
        validator=validators.instance_of(str)
    )
    DB_USERNAME: str = field(
        factory=lambda: os.getenv("DB_USERNAME", ""),
        validator=validators.instance_of(str)
    )
    DB_PASSWORD: str = field(
        factory=lambda: os.getenv("DB_PASSWORD", ""),
        validator=validators.instance_of(str)
    )

    def get_connection(self) -> Connection:
        connection_str = ""
        if not self.DB_USERNAME or not self.DB_PASSWORD:
            # Windows Authentication
            connection_str = f"Server={self.DB_SERVER};\
                Database={self.DB_DATABASE};\
                UID={self.DB_USERNAME};\
                PWD={self.DB_PASSWORD};\
                Trusted_Connection=yes;\
                TrustServerCertificate=yes;"
        else:
            # SQL Server Authentication
            connection_str = f"Server={self.DB_SERVER};\
                Database={self.DB_DATABASE};\
                UID={self.DB_USERNAME};\
                PWD={self.DB_PASSWORD};\
                TrustServerCertificate=yes;"

        return mssql_python.connect(connection_str)


if __name__ == "__main__":
    db_cfg_pyodbc_sqlal = RDSConfigPyODBC()
    db_cfg_pyodbc_raw = RDSConfigPyODBCRaw()
    db_cfg_mssqlpython = RDSConfigMSSQLPython()

    engine1 = db_cfg_pyodbc_sqlal.get_engine()
    engine2 = db_cfg_pyodbc_raw.get_connection()
    engine3 = db_cfg_mssqlpython.get_connection()
    print(engine1, engine2, engine3, sep='\r\n')