Construir cadenas de conexión de forma programática

Muchas aplicaciones necesitan construir cadenas de conexión dinámicamente en lugar de almacenarlas como valores de configuración estáticos. Elige el enfoque que se adapte a tu despliegue:

• Variables de entorno: Mejor para contenedores, CI/CD y aplicaciones de 12 pasos. Sencillo y ampliamente soportado.
• Archivos de configuración JSON/YAML: Ideales para aplicaciones con múltiples entornos (desarrollo, staging, producción) que necesitan configuración estructurada.
• Azure Key Vault: Ideal para despliegues en producción donde los secretos deben ser gestionados y auditados centralmente.
• Clase builder: Ideal para bibliotecas o frameworks que necesitan construir cadenas de conexión a partir de la entrada del usuario con escape automático.

Construcción básica de cuerdas

Usa f-strings

Las cadenas f son un enfoque común para scripts rápidos y prototipos. Evita este patrón cuando los valores provienen de la entrada del usuario, porque un valor malicioso como mydb;Server=evil.com podría alterar el destino de conexión:

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)

Usa «Unirse»

El join enfoque separa los pares clave-valor en una llamada a función tipo diccionario, que es más fácil de leer y mantener que una cadena f larga. También filtra None los valores automáticamente, por lo que puedes pasar parámetros opcionales sin lógica condicional adicional:

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)

Clase constructora de cuerdas de conexión

Una clase builder proporciona una API fluida con escape automático. Este enfoque es útil en bibliotecas o aplicaciones multitenant donde los parámetros de conexión provienen de diferentes fuentes:

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

Configuración basada en el entorno

A partir de variables de entorno

Leer los parámetros de conexión desde las variables de entorno evita incluir las credenciales en el código fuente y funciona tanto en entornos de desarrollo local como en contenedores y pipelines de CI/CD. La función comprueba qué método de autenticación usar en función de las variables que se han establecido:

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

Con python-dotenv

El paquete python-dotenv carga pares clave-valor de un archivo .env en variables de entorno, de modo que tu código lea las credenciales de la misma forma tanto en el desarrollo local como en producción. El .env archivo permanece fuera del control de versiones (añádelo a .gitignore), mientras que los entornos desplegados inyectan las mismas variables a través de su almacén secreto de plataforma.

Instale con pip install python-dotenv.

Crea un .env archivo en la raíz de tu proyecto con tus parámetros de conexión:

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

Luego carga y usa esos valores en tu 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)

Configuración basada en archivos

A partir de la configuración JSON

Un archivo de configuración JSON te permite definir configuraciones de conexión para múltiples entornos (desarrollo, staging, producción) en un solo lugar. La función lee el archivo, selecciona el entorno destino y construye la cadena de conexión a partir de los ajustes estructurados:

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

Desde la configuración de YAML

Los archivos de configuración YAML son una alternativa legible a JSON. Se usan comúnmente en proyectos de Python y despliegues de Kubernetes. Este enfoque lee la configuración de conexión de un archivo YAML estructurado y construye la cadena de conexión basándose en el tipo de autenticación definido en la configuración.

Instala el paquete ejecutando pip install pyyaml.

Crea un database.yml archivo en tu proyecto:

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

Luego carga y usa esos ajustes en tu 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")

Integración de Azure Key Vault

Para despliegues en producción, almacena las credenciales de conexión en Azure Key Vault en lugar de en archivos de configuración o variables de entorno. Key Vault proporciona gestión centralizada de secretos, auditoría de acceso y rotación automática. Instala los paquetes necesarios ejecutando pip install azure-keyvault-secrets azure-identity. Para una guía completa, consulta Quickstart: Azure Key Vault biblioteca secreta cliente para 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)

Manejo de caracteres especiales

Puntos y coma de escape y férulas

Necesitas escapar de los valores de cadena de conexión que contienen caracteres especiales. Envuelva el valor entre llaves {} y duplique cualquier llave de cierre interna }:

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

Constructor con escape automático

Esta clase constructora envuelve automáticamente todos los valores, así que los llamadores no necesitan recordar las reglas de escape. Úsalo cuando los parámetros de conexión provienen de entradas externas, como formularios de usuario, APIs de configuración o almacenes secretos donde los valores pueden contener puntos y coma o llaves:

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

Antes de usar una cadena de conexión dinámicamente construida en tu aplicación, verifica que realmente se conecte. Esta función auxiliar intenta una consulta ligera y devuelve un resultado booleano:

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

Contenido relacionado

• Cadenas de conexión
• Autenticación de Microsoft Entra
• Administración de conexiones