Постройте строки соединения программным способом

Многим приложениям необходимо динамически строить строки соединений, а не хранить их в виде статических конфигурационных значений. Выберите подход, который соответствует вашему развертыванию:

  • Переменные среды: лучше всего подходят для контейнеров, CI/CD и 12-факторных приложений. Просто и широко поддерживается.
  • Конфигурационные файлы JSON/YAML: Лучше всего подходит для приложений с несколькими средами (разработка, staging, prod), которым требуется структурированная конфигурация.
  • Azure Key Vault: Лучше всего подходит для производственных развертываний, где секреты должны централизованно управляться и проверяться.
  • Класс конструктора: Лучше всего подходит для библиотек или фреймворков, которым нужно строить строки соединения на основе пользовательского ввода с автоматическим выходом.

Основная конструкция струн

Используйте f-строки

f-строки — распространённый подход для быстрых скриптов и прототипов. Избегайте этого шаблона, если значения поступают из пользовательского ввода, поскольку вредоносное значение, например mydb;Server=evil.com, может изменить целевой адрес подключения:

import mssql_python

server = "<server>.database.windows.net"
database = "<database>"

connection_string = f"Server={server};Database={database};Authentication=ActiveDirectoryDefault;Encrypt=yes;"

conn = mssql_python.connect(connection_string)

Используйте join

Этот подход разделяет пары join ключ-значение в словарный вызов функции, который проще читать и поддерживать, чем длинная f-строка. Он также автоматически фильтрует None значения, так что вы можете передавать дополнительные параметры без дополнительной условной логики:

def build_connection_string(**kwargs) -> str:
    """Build connection string from keyword arguments."""
    return ";".join(f"{key}={value}" for key, value in kwargs.items() if value is not None)

conn_str = build_connection_string(
    Server="<server>.database.windows.net",
    Database="<database>",
    Authentication="ActiveDirectoryDefault",
    Encrypt="yes"
)

conn = mssql_python.connect(conn_str)

Класс сборщика строк соединения

Класс конструктора предоставляет плавно управляемый API с автоматическим выходом. Этот подход полезен в библиотеках или многоарендных приложениях, где параметры соединения поступают из разных источников:

import mssql_python

class ConnectionStringBuilder:
    """Builder for SQL Server connection strings."""

    def __init__(self):
        self._params = {}

    def server(self, value: str) -> "ConnectionStringBuilder":
        self._params["Server"] = value
        return self

    def database(self, value: str) -> "ConnectionStringBuilder":
        self._params["Database"] = value
        return self

    def trusted_connection(self) -> "ConnectionStringBuilder":
        self._params["Trusted_Connection"] = "yes"
        return self

    def sql_auth(self, username: str, password: str) -> "ConnectionStringBuilder":
        self._params["UID"] = username
        self._params["PWD"] = password
        return self

    def entra_default(self) -> "ConnectionStringBuilder":
        self._params["Authentication"] = "ActiveDirectoryDefault"
        return self

    def entra_msi(self, client_id: str = None) -> "ConnectionStringBuilder":
        self._params["Authentication"] = "ActiveDirectoryMSI"
        if client_id:
            self._params["UID"] = client_id
        return self

    def encrypt(self, value: bool = True) -> "ConnectionStringBuilder":
        self._params["Encrypt"] = "yes" if value else "no"
        return self

    def trust_server_certificate(self, value: bool = True) -> "ConnectionStringBuilder":
        self._params["TrustServerCertificate"] = "yes" if value else "no"
        return self

    def connect_timeout(self, seconds: int) -> "ConnectionStringBuilder":
        self._timeout = seconds
        return self

    def build(self) -> str:
        """Build the connection string."""
        return ";".join(f"{k}={v}" for k, v in self._params.items())

    def connect(self) -> mssql_python.Connection:
        """Build and connect."""
        return mssql_python.connect(self.build(), timeout=getattr(self, '_timeout', 0))


# Usage examples
# Microsoft Entra authentication (recommended)
conn = (ConnectionStringBuilder()
    .server("<server>.database.windows.net")
    .database("<database>")
    .entra_default()
    .encrypt()
    .connect())

# Azure with managed identity
conn = (ConnectionStringBuilder()
    .server("<server>.database.windows.net")
    .database("<database>")
    .entra_msi()
    .encrypt()
    .connect())

Конфигурация, основанная на среде

Из переменных окружающей среды

Чтение параметров подключения из переменных среды позволяет не хранить учетные данные в исходном коде и одинаково хорошо подходит для локальной разработки, контейнеров и CI/CD-конвейеров. Функция проверяет, какой метод аутентификации использовать на основе установленных переменных:

import os
import mssql_python

def get_connection_from_env() -> mssql_python.Connection:
    """Build connection from environment variables."""
    server = os.environ.get("SQL_SERVER")
    database = os.environ.get("SQL_DATABASE")

    if not server or not database:
        raise ValueError("SQL_SERVER and SQL_DATABASE environment variables required")

    # Check for authentication method
    if os.environ.get("SQL_USE_MSI", "").lower() == "true":
        # Azure Managed Identity
        conn_str = f"Server={server};Database={database};Authentication=ActiveDirectoryMSI;Encrypt=yes;"
    elif os.environ.get("SQL_TRUSTED_CONNECTION", "").lower() == "true":
        # Windows authentication
        conn_str = f"Server={server};Database={database};Trusted_Connection=yes;Encrypt=yes;"
    else:
        # SQL authentication
        username = os.environ.get("SQL_USERNAME")
        password = os.environ.get("SQL_PASSWORD")
        if not username or not password:
            raise ValueError("SQL_USERNAME and SQL_PASSWORD required for SQL authentication")
        conn_str = f"Server={server};Database={database};UID={username};PWD={password};Encrypt=yes;"

    return mssql_python.connect(conn_str)

# Usage
conn = get_connection_from_env()

С помощью python-dotenv

Пакет python-dotenv загружает пары «ключ — значение» из файла .env в переменные среды, чтобы ваш код получал учетные данные одинаковым образом как при локальной разработке, так и в рабочей среде. Файл .env не попадает в систему контроля версий (для этого его добавляют в .gitignore), а в развернутых средах те же переменные передаются через хранилище секретов платформы.

Установите с помощью pip install python-dotenv.

Создайте .env файл в корень проекта с параметрами соединения:

# .env - add this file to .gitignore
SQL_SERVER=<server>.database.windows.net
SQL_DATABASE=<database>
SQL_USE_MSI=true

Затем загрузите и используйте эти значения в своём скрипте:

from dotenv import load_dotenv
import os
import mssql_python

# Load .env file into os.environ (no-op if the file doesn't exist)
load_dotenv()

server = os.getenv("SQL_SERVER")
database = os.getenv("SQL_DATABASE")

if not server or not database:
    raise ValueError("SQL_SERVER and SQL_DATABASE must be set in .env or as environment variables")

use_msi = os.getenv("SQL_USE_MSI", "false").lower() == "true"

if use_msi:
    conn_str = f"Server={server};Database={database};Authentication=ActiveDirectoryMSI;Encrypt=yes;"
else:
    conn_str = f"Server={server};Database={database};Authentication=ActiveDirectoryDefault;Encrypt=yes;"

conn = mssql_python.connect(conn_str)

Tip

load_dotenv() не перезаписывает переменные, которые уже заданные в среде. В продакшене задайте одни и те же имена переменных через вашу платформу (например, настройки приложения App Service или переменные среды контейнера) и полностью пропустите файл .env .

Конфигурация на основе файлов

Из конфигурации JSON

Конфигурационный файл JSON позволяет задавать настройки соединения для нескольких сред (разработка, стадирование, производство) в одном месте. Функция читает файл, выбирает целевую среду и строит строка подключения из структурированных настроек:

import json
import io
import mssql_python

def load_connection_from_json(config_file, environment: str = "development") -> str:
    """Load connection settings from a JSON config file or file-like object."""
    config = json.load(config_file)

    env_config = config.get(environment, {})
    db_config = env_config.get("database", {})

    params = {
        "Server": db_config.get("server"),
        "Database": db_config.get("database"),
        "Encrypt": "yes" if db_config.get("encrypt", True) else "no",
    }

    auth_type = db_config.get("authentication", "sql")
    if auth_type == "msi":
        params["Authentication"] = "ActiveDirectoryMSI"
    elif auth_type == "default":
        params["Authentication"] = "ActiveDirectoryDefault"
    elif auth_type == "windows":
        params["Trusted_Connection"] = "yes"
    else:
        params["UID"] = db_config.get("username")
        params["PWD"] = db_config.get("password")

    return ";".join(f"{k}={v}" for k, v in params.items() if v)

# Example: load from an inline JSON config (in production, use open("config.json"))
sample_config = json.dumps({
    "development": {
        "database": {
            "server": "localhost",
            "database": "devdb",
            "authentication": "windows",
            "encrypt": False
        }
    },
    "production": {
        "database": {
            "server": "prod.database.windows.net",
            "database": "proddb",
            "authentication": "msi",
            "encrypt": True
        }
    }
})

conn_str = load_connection_from_json(io.StringIO(sample_config), "production")
print(f"Connection string: {conn_str}")

Из конфигурации YAML

Конфигурационные файлы YAML являются читаемой альтернативой JSON. Они часто используются в проектах Python и развертываниях Kubernetes. Этот подход считывает настройки соединения из структурированного YAML-файла и строит строка подключения на основе типа аутентификации, определённого в конфигурации.

Установите пакет, запустив pip install pyyaml.

Создайте database.yml файл в вашем проекте:

database:
  server: <server>.database.windows.net
  name: <database>
  authentication: msi
  encrypt: true

Затем загрузите и используйте эти настройки в скрипте:

import yaml
import mssql_python

def load_from_yaml(config_path: str) -> mssql_python.Connection:
    """Load connection from YAML config."""
    with open(config_path) as f:
        config = yaml.safe_load(f)

    db = config["database"]

    parts = [
        f"Server={db['server']}",
        f"Database={db['name']}",
    ]

    if db.get("trusted_connection"):
        parts.append("Trusted_Connection=yes")
    elif db.get("authentication") == "msi":
        parts.append("Authentication=ActiveDirectoryMSI")
    else:
        parts.append(f"UID={db['username']}")
        parts.append(f"PWD={db['password']}")

    if db.get("encrypt", True):
        parts.append("Encrypt=yes")
    if db.get("trust_server_certificate"):
        parts.append("TrustServerCertificate=yes")

    return mssql_python.connect(";".join(parts))

conn = load_from_yaml("database.yml")

Интеграция с Azure Key Vault

Для производственных развертываний храните учетные данные соединения в Azure Key Vault, а не в конфигурационных файлах или переменных среды. Key Vault обеспечивает централизованное управление секретами, аудит доступа и автоматическую ротацию. Установите необходимые пакеты, запустив pip install azure-keyvault-secrets azure-identity. Полный обзор см. Quickstart: Azure Key Vault secret client library for Python.

import os
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import mssql_python

def get_connection_from_keyvault(vault_url: str) -> mssql_python.Connection:
    """Build connection using secrets from Azure Key Vault."""
    credential = DefaultAzureCredential()
    client = SecretClient(vault_url=vault_url, credential=credential)

    server = client.get_secret("sql-server").value
    database = client.get_secret("sql-database").value
    username = client.get_secret("sql-username").value
    password = client.get_secret("sql-password").value

    conn_str = f"Server={server};Database={database};UID={username};PWD={password};Encrypt=yes;"
    return mssql_python.connect(conn_str)

vault_url = os.environ.get("AZURE_KEY_VAULT_URL")
if vault_url:
    conn = get_connection_from_keyvault(vault_url)

Обрабатывать специальные символы

Точки с запятой и брекеты при выходе

Необходимо экранировать значения в строке подключения, содержащие специальные символы. Оберните значение в скобки {} и удвойте любые внутренние закрывающие брекеты }:

def escape_value(value: str) -> str:
    """Escape special characters in connection string values."""
    if ";" in value or "{" in value or "}" in value:
        # Wrap in braces and escape internal braces
        value = value.replace("}", "}}")
        return "{" + value + "}"
    return value

# Password with semicolon
password = "my;complex;password"
escaped_password = escape_value(password)  # {my;complex;password}

conn_str = f"Server=<server>;Database=<database>;UID=<login>;PWD={escaped_password};"

Конструктор с автоматическим выходом

Этот класс конструктора автоматически обворачивает все значения, поэтому звонящим не нужно запоминать правила побега. Используйте его, когда параметры соединения поступают из внешних входов, таких как пользовательские формы, API конфигурации или секретные хранилища, где значения могут содержать точки с запятой или скобки:

class SafeConnectionStringBuilder:
    """Connection string builder with automatic escaping."""

    SPECIAL_CHARS = {";", "{", "}"}

    def __init__(self):
        self._params = {}

    def _escape(self, value: str) -> str:
        if any(c in value for c in self.SPECIAL_CHARS):
            value = value.replace("}", "}}")
            return "{" + value + "}"
        return value

    def set(self, key: str, value: str) -> "SafeConnectionStringBuilder":
        self._params[key] = self._escape(value)
        return self

    def build(self) -> str:
        return ";".join(f"{k}={v}" for k, v in self._params.items())

# Safely handles special characters
builder = SafeConnectionStringBuilder()
builder.set("Server", "<server>.database.windows.net")
builder.set("Database", "<database>")
builder.set("PWD", "pass;word{with}special")  # Automatically escaped

conn_str = builder.build()

Validation

Прежде чем использовать динамически построенную строка подключения в вашем приложении, убедитесь, что она действительно подключается. Эта вспомогательная функция пытается выполнить лёгкий запрос и возвращает булевой результат:

import mssql_python

def validate_connection_string(conn_str: str) -> bool:
    """Validate a connection string by attempting to connect."""
    try:
        conn = mssql_python.connect(conn_str)
        cursor = conn.cursor()
        cursor.execute("SELECT 1")
        cursor.fetchone()
        conn.close()
        return True
    except mssql_python.Error as e:
        print(f"Connection failed: {e}")
        return False

# Test before using
conn_str = "Server=<server>.database.windows.net;Database=<database>;Authentication=ActiveDirectoryDefault;Encrypt=yes;"
if validate_connection_string(conn_str):
    print("Connection string is valid")

Связанные материалы

  • Last updated on