Bouw verbindingsreeksen programmatisch

Veel applicaties moeten verbindingsstrings dynamisch bouwen in plaats van ze op te slaan als statische configuratiewaarden. Kies de aanpak die past bij jouw inzet:

• Omgevingsvariabelen: Het beste voor containers, CI/CD en 12-factor apps. Eenvoudig en breed ondersteund.
• JSON/YAML configuratiebestanden: Het beste voor applicaties met meerdere omgevingen (dev, staging, productie) die gestructureerde configuratie nodig hebben.
• Azure Key Vault: Het beste voor productie-implementaties waarbij geheimen centraal beheerd en geauditeerd moeten worden.
• Builder-klasse: Het beste voor bibliotheken of frameworks die verbindingsstrings moeten construeren op basis van gebruikersinvoer met automatische escape.

Basisconstructie van snaars

Gebruik f-snaren

F-strings zijn een veelgebruikte aanpak voor snelle scripts en prototypes. Vermijd dit patroon wanneer waarden afkomstig zijn van gebruikersinvoer, omdat een kwaadaardige waarde zoals mydb;Server=evil.com het verbindingsdoel kan veranderen:

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)

Gebruik join

De join aanpak scheidt sleutel-waardeparen in een woordenboekachtige functieaanroep, die gemakkelijker te lezen en te onderhouden is dan een lange f-string. Het filtert ook automatisch waarden eruit None , zodat je optionele parameters kunt doorgeven zonder extra voorwaardelijke logica:

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)

Verbindingsstringbuilder-klasse

Een builder-klasse biedt een vloeiende API met automatische ontsnapping. Deze aanpak is nuttig in bibliotheken of multitenant-toepassingen waar verbindingsparameters uit verschillende bronnen komen:

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

Omgevingsgebaseerde configuratie

Uit omgevingsvariabelen

Het lezen van verbindingsparameters uit omgevingsvariabelen houdt inloggegevens buiten de broncode en werkt over lokale ontwikkeling, containers en CI/CD-pijplijnen. De functie controleert welke authenticatiemethode gebruikt moet worden op basis van de variabelen die zijn ingesteld:

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

Met python-dotenv

Het python-dotenv-pakket laadt sleutel-waardeparen uit een .env-bestand in omgevingsvariabelen, zodat je code inloggegevens op dezelfde manier leest tijdens lokale ontwikkeling en in productie. Het .env bestand blijft buiten de broncode (voeg het toe aan .gitignore), terwijl gedeployte omgevingen dezelfde variabelen injecteren via hun platformgeheime opslag.

Installeren met pip install python-dotenv.

Maak een .env bestand aan in je projectwortel met je verbindingsparameters:

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

Laad en gebruik die waarden vervolgens in je script:

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)

Bestandsgebaseerde configuratie

Uit JSON-config

Met een JSON-configuratiebestand kun je verbindingsinstellingen definiëren voor meerdere omgevingen (ontwikkeling, staging, productie) op één plek. De functie leest het bestand, selecteert de doelomgeving en bouwt de verbindingsreeks op basis van de gestructureerde instellingen:

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

Uit YAML-configuratie

YAML-configuratiebestanden zijn een leesbaar alternatief voor JSON. Ze worden vaak gebruikt in Python-projecten en Kubernetes-implementaties. Deze aanpak leest verbindingsinstellingen uit een gestructureerd YAML-bestand en bouwt de verbindingsreeks op basis van het authenticatietype dat in de configuratie is gedefinieerd.

Installeer het pakket door pip install pyyaml uit te voeren.

Maak een database.yml bestand aan in je project:

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

Laad en gebruik dan die instellingen in je script:

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-integratie

Voor productie-implementaties sla je verbindingsgegevens op in Azure Key Vault in plaats van in configuratiebestanden of omgevingsvariabelen. Key Vault biedt gecentraliseerd geheimbeheer, toegangsaudits en automatische rotatie. Installeer de benodigde pakketten door te draaien pip install azure-keyvault-secrets azure-identity. Voor een volledige walkthrough, zie Quickstart: Azure Key Vault secret client library voor 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)

Speciale tekens verwerken

Ontsnappingspuntkomma's en beugels

Je moet ontsnappen aan verbindingsreeks-waarden die speciale tekens bevatten. Plaats de waarde tussen accolades {} en verdubbel eventuele interne sluitaccolades }:

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

Bouwer met automatische ontsnapping

Deze builder-klasse wikkelt automatisch elke waarde, zodat callers de ontsnappingsregels niet hoeven te onthouden. Gebruik het wanneer verbindingsparameters afkomstig zijn van externe invoer zoals gebruikersformulieren, configuratie-API's of geheime opslagen waar waarden puntkomma's of haakjes kunnen bevatten:

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

Controleer voordat je een dynamisch opgebouwde verbindingsreeks in je applicatie gebruikt of deze daadwerkelijk verbinding maakt. Deze helperfunctie probeert een lichtgewicht query en geeft een booleaans resultaat terug:

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

Verwante onderwerpen

• Verbindingsreeksen
• Microsoft Entra-authenticatie
• Verbindingsbeheer