Kapcsolati stringek építése programozott módon

Sok alkalmazásnak dinamikusan kell kapcsoló stringeket építenie, nem pedig statikus konfigurációs értékekként tárolni őket. Válaszd ki azt a megközelítést, amely megfelel a telepítésednek:

• Környezeti változók: A legjobbak konténerekhez, CI/CD-hez és 12 faktoros alkalmazásokhoz. Egyszerű és széles körben támogatott.
• JSON/YAML konfigurációs fájlok: A legjobbak olyan alkalmazásokhoz, ahol több környezet (fejlesztő, staging, prod) vannak, és strukturált konfigurációra van szükség.
• Azure Key Vault: Legjobb olyan termelési telepítésekhez, ahol a titkokat központilag kell kezelni és auditálni.
• Builder osztály: A legjobb olyan könyvtáraknak vagy keretrendszereknek, amelyeknek felhasználói bemenetből kell kapcsolati stringeket építeniük, automatikus escapingel.

Alapvető húrszerkezet

Használj f-stringeket

Az f-stringek gyakori megközelítések gyors szkriptekhez és prototípusokhoz. Kerüld el ezt a mintát, amikor az értékek a felhasználói bemenetből származnak, mert egy rosszindulatú érték, például mydb;Server=evil.com megváltoztathatja a kapcsolati célpontot:

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)

Használd a join funkciót

Ez a join megközelítés a kulcs–érték párokat egy szótárszerű függvényhívásba rendezi, amely könnyebben olvasható és karbantartható, mint egy hosszú f-string. Automatikusan kiszűri None az értékeket, így opcionális paramétereket is átadhatsz extra feltételes logika nélkül:

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)

Kapcsolati karakterlánc létrehozó osztály

Egy builder osztály egy folyékonyan működő API-t biztosít automatikus escapinggel. Ez a megközelítés hasznos könyvtárakban vagy többbérlős alkalmazásokban, ahol a kapcsolati paraméterek különböző forrásokból származnak:

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

Környezet-alapú konfiguráció

Környezeti változókból

A kapcsolati paraméterek olvasása környezeti változókból távol tartja a hitelesítéseket a forráskódtól, és működik a helyi fejlesztésen, konténereken és CI/CD vezetékeken keresztül. A függvény ellenőrzi, melyik hitelesítési módszert használja a beállított változók alapján:

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

Python-dotenv segítségével

A python-dotenv csomag kulcs-érték párokat tölt be egy .env fájlból környezeti változókba, így a kódod ugyanúgy olvassa fel a hitelesítéseket helyi fejlesztésben és gyártásban. A .env fájl nem kerül be a verziókezelőbe (add hozzá a .gitignore-hoz), míg az üzemelő környezetek ugyanezeket a változókat a platform titoktárán keresztül biztosítják.

Telepítsd a pip install python-dotenv-val.

Hozz létre egy .env fájlt a projekt gyökérben a kapcsolati paraméterekkel:

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

Ezután töltsd be és használd ezeket az értékeket a szkriptedben:

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)

Jótanács

load_dotenv() nem írja felül azokat a változókat, amelyek már be vannak állítva a környezetben. Gyártásban állítsd be ugyanazokat a változóneveket a platformon keresztül (például App Service alkalmazásbeállítások vagy konténer környezeti változók), és teljesen kihagyd a .env fájlt.

Fájlalapú konfiguráció

JSON konfigurációból

Egy JSON konfigurációs fájl lehetővé teszi, hogy több környezethez (fejlesztés, stázis, gyártás) kapcsoló beállításokat definiálj egy helyen. A függvény olvassa a fájlt, kiválasztja a célkörnyezetet, és a strukturált beállításokból építi a kapcsolati karakterlánc-et:

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

YAML-konfigurációból

A YAML konfigurációs fájlok olvasható alternatívát jelentenek a JSON-nak. Gyakran használják Python projektekben és Kubernetes telepítésekben. Ez a megközelítés egy strukturált YAML fájlból olvassa fel a kapcsolati beállításokat, és a konfigurációban meghatározott hitelesítési típus alapján építi fel a kapcsolati karakterlánc-et.

Telepítsd a csomagot futtatva pip install pyyaml.

Készíts egy database.yml fájlt a projektedben:

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

Ezután töltsd be és használd ezeket a beállításokat a szkriptedben:

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-integráció

A gyártási telepítésekhez a kapcsolati hitelesítéseket az Azure Key Vault-ban tárolják, ne konfigurációs fájlokban vagy környezeti változókban. A Key Vault központosított titkos kezelést, hozzáférési auditálást és automatikus rotációt biztosít. Telepítsd a szükséges csomagokat futtatva pip install azure-keyvault-secrets azure-identity. A teljes útmutatóért lásd: 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)

Speciális karakterek kezelése

Escape félvesszők és fogszabályzók

Ki kell menned azoktól a kapcsolati karakterlánc értékektől, amelyek speciális karaktereket tartalmaznak. Csomagoljuk be az értéket fogszabályzóba {} , és duplázzuk meg a belső záró }tartókat:

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

Építő automatikus meneküléssel

Ez az építő osztály automatikusan becsomagolja az összes értéket, így a hívóknak nem kell emlékezniük a menekülési szabályokra. Akkor használja, ha a kapcsolódási paraméterek külső forrásból származnak, például felhasználói űrlapokból, konfigurációs API-kból vagy titoktárolókból, ahol az értékek pontosvesszőt vagy kapcsos zárójeleket tartalmazhatnak:

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

Mielőtt dinamikusan felépített kapcsolati karakterláncot használna az alkalmazásában, ellenőrizze, hogy valóban létrejön-e vele a kapcsolat. Ez a segítő függvény könnyűbb lekérdezést próbál meg, és egy boolean eredményt ad vissza:

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

Kapcsolódó tartalom

• Kapcsolati karakterláncok
• Microsoft Entra-hitelesítés
• Kapcsolatkezelés