将 python-polars 连接到 SQL Server（目前不支持）

Question

如何直接将 MS SQL Server 连接到 Polars？

该文档没有列出任何支持的连接，但建议使用 pandas。

更新：

SQL Server 身份验证按答案工作，但 Windows 域身份验证不起作用。参见问题

Answer 1

在这里，您可以使用 Polar 连接到 MS SQL Server（底层的connectorx）。只需使用连接字符串：

import polars as pl

# usually don't store sensitive info in plain text
username = 'my_username'
password = '1234'
server = 'SERVER1'
database = 'db1'
trusted_conn = 'no' # or yes

conn = f'mssql://{username}:{password}@{server}/{database}?driver=SQL+Server&trusted_connection={trusted_conn}'

query = "SELECT * FROM table1"

df = pl.read_sql(query, conn)

Answer 2

啊，实际上，MsSQL支持直接加载到 Polars 中（通过完成这项工作的底层库，即connectorx）；该文档只是稍微过时了 - 我会看一下并相应地刷新它。

Answer 3

下面的类可以轻松连接到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.")