sql-serversqlalchemypyarrowpython-polars

Connect python-polars to SQL server (no support currently)


How can I directly connect MS SQL Server to polars?

The documentation does not list any supported connections but recommends the use of pandas.

Update:

SQL Server Authentication works per answer, but Windows domain authentication is not working. see issue


Solution

  • The class below allows for easy connection to connectorx.

    import keyring
    from typing import Optional
    
    
    class KeyringSQL:
        def __init__(self,  username: str, server: str, database: str, server_type:  Optional[str]= 'mssql' ,driver: Optional[str] = 'SQL+Server', trusted_connection: Optional[bool] = False):
            self.server_type = server_type
            self.username = username
            self.server = server
            self.database = database
            self.driver = driver
            self.trusted_connection = trusted_connection
    
        def safe_to_expose_dict(self):
            return {'server_type': self.server_type, 'server': self.server, 'database': self.database, 'driver': self.driver}
        def full_table_path(self, tablename: str):
            if self.server_type == 'mssql':
                return f"[{self.database}].[dbo].[{tablename}]"
            return None
        def get_connection_uri(self) -> str:
            password = self._get_password() if not self.trusted_connection else ''
            append = '&trusted_connection=true' if self.trusted_connection else ''
            return f"{self.server_type}://{self.username}:{password}@{self.server}/{self.database}?driver={self.driver}{append}"
    
        def _get_password(self) -> str:
            password = keyring.get_password(self.server, self.username)
            if password is None:
                password = self._ask_password()
                keyring.set_password(self.server, self.username, password)
            return password
    
        def _ask_password(self) -> str:
            while True:
                password1 = input(f"Please set the password for {self.server}")
                password2 = input(f"Please confirm the password for {self.server}")
                if password1 == password2:
                    return password1
                print("Passwords did not match. Please try again.")