Erstellen Sie Verbindungsstrings programmatisch

Viele Anwendungen müssen Verbindungsstrings dynamisch erstellen, anstatt sie als statische Konfigurationswerte zu speichern. Wählen Sie den Ansatz, der zu Ihrem Einsatz passt:

• Umweltvariablen: Am besten für Container, CI/CD und 12-Faktor-Apps. Einfach und breit unterstützt.
• JSON/YAML-Konfigurationsdateien: Am besten für Anwendungen mit mehreren Umgebungen (Entwicklung, Staging, Produktion), die eine strukturierte Konfiguration benötigen.
• Azure Key Vault: Am besten für Produktionsdeployments, bei denen Geheimnisse zentral verwaltet und geprüft werden müssen.
• Builder-Klasse: Am besten für Bibliotheken oder Frameworks, die Verbindungsstrings aus Benutzereingaben mit automatischem Escape erstellen müssen.

Grundlegende Saitenkonstruktion

Verwenden Sie f-Strings

F-Strings sind ein gängiger Ansatz für schnelle Skripte und Prototypen. Vermeiden Sie dieses Muster, wenn Werte von Benutzereingaben stammen, da ein bösartiger Wert wie mydb;Server=evil.com das Verbindungsziel verändern könnte:

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)

Verwenden Sie Join

Der Ansatz join trennt Schlüssel-Wert-Paare in einen wörterbuchähnlichen Funktionsaufruf, der leichter zu lesen und zu warten ist als ein langer f-String. Es filtert auch automatisch NoneWerte heraus, sodass du optionale Parameter ohne zusätzliche bedingte Logik weitergeben kannst:

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)

Klasse des Verbindungsstring-Builders

Eine Builder-Klasse bietet eine flüssige API mit automatischem Escape. Dieser Ansatz ist nützlich in Bibliotheken oder Multitenant-Anwendungen, bei denen die Verbindungsparameter aus verschiedenen Quellen stammen:

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())

Umgebungsbasierte Konfiguration

Aus Umgebungsvariablen

Das Lesen von Verbindungsparametern aus Umgebungsvariablen hält Zugangsdaten aus dem Quellcode heraus und funktioniert über lokale Entwicklung, Container und CI/CD-Pipelines hinweg. Die Funktion prüft, welche Authentifizierungsmethode verwendet werden soll, basierend auf den gesetzten Variablen:

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()

Mit python-dotenv

Das Paket python-dotenv lädt Schlüssel-Wert-Paare aus einer .env-Datei in Umgebungsvariablen, sodass Ihr Code Zugangsdaten bei der lokalen Entwicklung und in der Produktion auf die gleiche Weise ausliest. Die Datei .env bleibt außerhalb der Versionsverwaltung (fügen Sie sie zu .gitignore hinzu), während bereitgestellte Umgebungen dieselben Variablen über den Secret-Store ihrer Plattform bereitstellen.

Installieren mit pip install python-dotenv.

Erstellen Sie eine .env Datei in Ihrer Projektwurzel mit Ihren Verbindungsparametern:

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

Dann lade und nutze diese Werte in deinem Skript:

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)

Dateibasierte Konfiguration

Aus der JSON-Konfiguration

Eine JSON-Konfigurationsdatei ermöglicht es, Verbindungseinstellungen für mehrere Umgebungen (Entwicklung, Staging, Produktion) an einem Ort zu definieren. Die Funktion liest die Datei, wählt die Zielumgebung aus und erstellt die Verbindungszeichenfolge aus den strukturierten Einstellungen:

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}")

Aus der YAML-Konfiguration

YAML-Konfigurationsdateien sind eine lesbare Alternative zu JSON. Sie werden häufig in Python-Projekten und Kubernetes-Deployments verwendet. Dieser Ansatz liest die Verbindungseinstellungen aus einer strukturierten YAML-Datei und erstellt die Verbindungszeichenfolge basierend auf dem in der Konfiguration definierten Authentifizierungstyp.

Installiere das Paket, indem du pip install pyyaml ausführst.

Erstellen Sie eine database.yml Datei in Ihrem Projekt:

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

Dann lade und nutze diese Einstellungen in deinem Skript:

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")

Die Integration von Azure Key Vault

Für Produktionsbereitstellungen speichern Sie die Verbindungsdaten in Azure Key Vault und nicht in Konfigurationsdateien oder Umgebungsvariablen. Key Vault bietet zentrale Verwaltung von Geheimnissen, Zugriffsprüfung und automatische Rotation. Installiere die erforderlichen Pakete, indem du pip install azure-keyvault-secrets azure-identity ausführst. Für eine vollständige Anleitung siehe Quickstart: Azure Key Vault geheime Client-Bibliothek für 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)

Sonderzeichen verarbeiten

Entkommen von Semikolons und Zahnspangen

Du musst Werte in Verbindungszeichenfolgen maskieren, die Sonderzeichen enthalten. Wickeln Sie den Wert in Streben {} und verdoppeln Sie alle internen Schließstreben }:

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};"

Builder mit automatischem Escaping

Diese Builder-Klasse umschließt automatisch jeden Wert, sodass Aufrufer sich Escape-Regeln nicht merken müssen. Verwenden Sie es, wenn Verbindungsparameter von externen Eingaben wie Benutzerformularen, Konfigurations-APIs oder geheimen Speicher stammen, bei denen Werte Semikolon oder Krasseln enthalten könnten:

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

Bevor Sie einen dynamisch aufgebauten Verbindungszeichenfolge in Ihrer Anwendung verwenden, überprüfen Sie, ob er tatsächlich verbunden ist. Diese Hilfsfunktion versucht eine leichte Abfrage und liefert ein boolsches Ergebnis:

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")

Verwandte Inhalte

• Verbindungszeichenfolgen
• Microsoft Entra-Authentifizierung
• Verbindungsverwaltung