Nota
L'accesso a questa pagina richiede l'autorizzazione. È possibile provare ad accedere o modificare le directory.
L'accesso a questa pagina richiede l'autorizzazione. È possibile provare a modificare le directory.
Molte applicazioni devono costruire stringhe di connessione dinamicamente invece di memorizzarle come valori di configurazione statici. Scegli l'approccio che si adatta alla tua implementazione:
- Variabili ambientali: Migliori per container, CI/CD e app a 12 fattori. Semplice e ampiamente supportato.
- File di configurazione JSON/YAML: Ideali per applicazioni con più ambienti (sviluppo, staging, produzione) che necessitano di una configurazione strutturata.
- Azure Key Vault: Ideale per le implementazioni in produzione dove i segreti devono essere gestiti e auditati centralmente.
- Builder class: Ideale per librerie o framework che devono costruire stringhe di connessione a partire dall'input dell'utente con escape automatico.
Costruzione base delle corde
Usa le f-string
Le F-strings sono un approccio comune per script rapidi e prototipi. Evita questo schema quando i valori provengono da input dell'utente, perché un valore malevolo come mydb;Server=evil.com potrebbe alterare il target di connessione:
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 il join
L'approccio join separa le coppie chiave-valore in una chiamata di funzione in stile dizionario, che è più facile da leggere e mantenere rispetto a una lunga stringa f. Esclude anche automaticamente i valori None, in modo da poter passare parametri opzionali senza logica condizionale aggiuntiva:
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 costruttore di stringhe di connessione
Una classe builder fornisce un'API fluente con escape automatico. Questo approccio è utile nelle librerie o applicazioni multitenant dove i parametri di connessione provengono da fonti diverse:
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())
Configurazione basata sull'ambiente
Dalle variabili ambientali
La lettura dei parametri di connessione dalle variabili dell'ambiente tiene le credenziali fuori dal codice sorgente e funziona nello sviluppo locale, nei container e nelle pipeline CI/CD. La funzione verifica quale metodo di autenticazione utilizzare in base alle variabili impostate:
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
Il pacchetto python-dotenv carica coppie chiave-valore da un file .env nelle variabili d'ambiente, così il tuo codice legge le credenziali allo stesso modo sia nello sviluppo locale sia in produzione. Il .env file rimane fuori dal controllo del servizio (aggiungilo a .gitignore), mentre gli ambienti distribuiti iniettano le stesse variabili attraverso il loro archivio segreto della piattaforma.
Installa con pip install python-dotenv.
Crea un .env file nella radice del tuo progetto con i parametri di connessione:
# .env - add this file to .gitignore
SQL_SERVER=<server>.database.windows.net
SQL_DATABASE=<database>
SQL_USE_MSI=true
Poi carica e usa quei valori nel tuo 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() non sovrascrive variabili già impostate nell'ambiente. In produzione, imposta gli stessi nomi delle variabili tramite la tua piattaforma (ad esempio, impostazioni dell'applicazione App Service o variabili dell'ambiente container) e salta completamente il .env file.
Configurazione basata su file
Da configurazione JSON
Un file di configurazione JSON ti permette di definire le impostazioni di connessione per più ambienti (sviluppo, staging, produzione) in un unico posto. La funzione legge il file, seleziona l'ambiente di destinazione e costruisce la stringa di connessione dalle impostazioni strutturate:
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}")
Da configurazione YAML
I file di configurazione YAML sono un'alternativa leggibile al JSON. Sono comunemente utilizzati nei progetti Python e nelle distribuzioni su Kubernetes. Questo approccio legge le impostazioni di connessione da un file YAML strutturato e costruisce la stringa di connessione basandosi sul tipo di autenticazione definito nella configurazione.
Installa il pacchetto eseguendo pip install pyyaml.
Crea un database.yml file nel tuo progetto:
database:
server: <server>.database.windows.net
name: <database>
authentication: msi
encrypt: true
Poi carica e usa queste impostazioni nel tuo 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")
Integrazione di Azure Key Vault
Per le implementazioni in produzione, memorizza le credenziali di connessione in Azure Key Vault invece che in file di configurazione o variabili di ambiente. Key Vault fornisce gestione centralizzata dei segreti, audit degli accessi e rotazione automatica. Installa i pacchetti necessari eseguendo pip install azure-keyvault-secrets azure-identity. Per una guida completa, consulta 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)
Gestire i caratteri speciali
Punti e virgole di fuga e apparecchio
Devi uscire dai valori delle stringa di connessione che contengono caratteri speciali. Avvolgere il valore tra le bracciali {} e raddoppiare eventuali }bricole interne di chiusura:
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};"
Costruttore con fuga automatica
Questa classe builder racchiude automaticamente ogni valore, quindi chi la utilizza non deve ricordare le regole di escape. Usalo quando i parametri di connessione provengono da input esterni come moduli utente, API di configurazione o archivi segreti dove i valori possono contenere punti virgola o parentesi:
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
Prima di usare una stringa di connessione dinamica nella tua applicazione, verifica che si connetta davvero. Questa funzione di supporto tenta una query leggera e restituisce un risultato 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")