Construire les chaînes de connexion de façon programmatique

De nombreuses applications doivent construire des chaînes de connexion dynamiquement plutôt que de les stocker sous forme de valeurs de configuration statiques. Choisissez l’approche qui correspond à votre déploiement :

  • Variables d’environnement : Idéal pour les conteneurs, CI/CD et applications à 12 facteurs. Simple et largement supporté.
  • Fichiers de configuration JSON/YAML : Idéal pour les applications avec plusieurs environnements (développement, staging, production) nécessitant une configuration structurée.
  • Azure Key Vault : Idéal pour les déploiements en production où les secrets doivent être gérés et audités de manière centralisée.
  • Classe Builder : Idéale pour les bibliothèques ou frameworks qui doivent construire des chaînes de connexion à partir des saisies utilisateur avec évasion automatique.

Construction de base des cordes

Utilisez les f-strings

Les F-Strings sont une approche courante pour les scripts rapides et les prototypes. Évitez ce schéma lorsque les valeurs proviennent d’une saisie utilisateur, car une valeur malveillante comme mydb;Server=evil.com pourrait modifier la cible de connexion :

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)

Utiliser la jointure

Cette join approche sépare les paires clé-valeur en un appel de fonction de type dictionnaire, qui est plus facile à lire et à maintenir qu’une longue chaîne f. Il filtre aussi automatiquement les valeurs None, ce qui vous permet de transmettre des paramètres optionnels sans logique conditionnelle supplémentaire :

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)

Classe constructrice de chaînes de connexion

Une classe builder fournit une API fluide avec évasion automatique. Cette approche est utile dans les bibliothèques ou applications multilocataires où les paramètres de connexion proviennent de sources différentes :

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

Configuration basée sur l’environnement

À partir des variables d’environnement

La lecture des paramètres de connexion à partir des variables d’environnement permet d’exclure les identifiants du code source et fonctionne à travers le développement local, les conteneurs et les pipelines CI/CD. La fonction vérifie quelle méthode d’authentification utiliser en fonction des variables définies :

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

Avec python-dotenv

Le paquet python-dotenv charge des paires clé-valeur à partir d’un fichier .env dans des variables d’environnement afin que votre code lise les informations d’identification de la même manière en développement local et en production. Le fichier .env reste exclu du contrôle de version (ajoutez-le à .gitignore), tandis que les environnements déployés injectent les mêmes variables via le gestionnaire de secrets de leur plateforme.

Installer avec pip install python-dotenv.

Créez un .env fichier dans la racine de votre projet avec vos paramètres de connexion :

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

Ensuite, chargez et utilisez ces valeurs dans votre 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)

Tip

load_dotenv() ne remplace pas les variables déjà définies dans l’environnement. En production, définissez les mêmes noms de variables via votre plateforme (par exemple, paramètres d’application ou variables d’environnement conteneur) et zappez complètement le .env fichier.

Configuration basée sur des fichiers

À partir de la configuration JSON

Un fichier de configuration JSON vous permet de définir les paramètres de connexion pour plusieurs environnements (développement, staging, production) en un seul endroit. La fonction lit le fichier, sélectionne l’environnement cible et construit la chaîne de connexion à partir des paramètres structurés :

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

À partir de la configuration YAML

Les fichiers de configuration YAML sont une alternative lisible au JSON. Ils sont couramment utilisés dans les projets Python et les déploiements Kubernetes. Cette approche lit les paramètres de connexion à partir d’un fichier YAML structuré et construit la chaîne de connexion en fonction du type d’authentification défini dans la configuration.

Installez le package en exécutant pip install pyyaml.

Créez un database.yml fichier dans votre projet :

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

Ensuite, chargez et utilisez ces paramètres dans votre 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")

Intégration d’Azure Key Vault

Pour les déploiements en production, stockez les identifiants de connexion dans Azure Key Vault plutôt que dans des fichiers de configuration ou des variables d’environnement. Key Vault assure une gestion centralisée des secrets, un audit d’accès et une rotation automatique. Installez les paquets requis en exécutant pip install azure-keyvault-secrets azure-identity. Pour une procédure complète, consultez Démarrage rapide : bibliothèque cliente de secrets Azure Key Vault pour 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)

Gérer les caractères spéciaux

Échapper aux points-virgules et aux accolades

Vous devez échapper aux valeurs de chaîne de connexion qui contiennent des caractères spéciaux. Enroulez la valeur en entrecoupes {} et doublez toutes les entrecoupes }internes de fermeture :

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

Constructeur avec évasion automatique

Cette classe constructrice enveloppe automatiquement chaque valeur, donc les appelants n’ont pas besoin de se souvenir des règles d’évasion. Utilisez-le lorsque les paramètres de connexion proviennent d’entrées externes telles que les formulaires utilisateur, les API de configuration ou les magasins secrets où les valeurs peuvent contenir des points-virgules ou des accolades :

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

Avant d’utiliser une chaîne de connexion dynamique dans votre application, vérifiez qu’elle se connecte réellement. Cette fonction d’assistance tente une requête légère et retourne un résultat booléen :

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