Bağlantı dizelerini programatik olarak oluştur

Birçok uygulama, bağlantı dizileri statik yapılandırma değerleri olarak depolamak yerine dinamik olarak oluşturmak zorundadır. Dağıtımınıza uygun yaklaşımı seçin:

  • Ortam değişkenleri: Konteynerler, CI/CD ve 12 faktörlü uygulamalar için en iyisi. Basit ve geniş çapta destekleniyor.
  • JSON/YAML yapılandırma dosyaları: Yapılandırılmış yapılandırma gerektiren birden fazla ortama (geliştirme, sahneleme, yapım) sahip uygulamalar için en iyisidir.
  • Azure Key Vault: Sırların merkezi olarak yönetilmesi ve denetlenmesi gereken üretim dağıtımları için en iyisidir.
  • Builder sınıfı: Kullanıcı girdisinden otomatik kaçış ile bağlantı dizileri oluşturması gereken kütüphaneler veya framework'ler için en iyisidir.

Temel tel yapısı

f-stringleri kullanın

f-dizileri, hızlı betikler ve prototipler için yaygın bir yaklaşımdır. Kullanıcı girdisinden gelen değerler olduğunda bu desenden kaçının, çünkü kötü niyetli bir değer bağlantı mydb;Server=evil.com hedefini değiştirebilir:

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)

Join kullanın

join yaklaşımı, anahtar-değer çiftlerini, uzun bir f-dizgesine kıyasla okunması ve bakımı daha kolay olan sözlük biçimindeki bir işlev çağrısına ayırır. Ayrıca, None değerlerini otomatik olarak filtreler; böylece ek koşullu mantık olmadan isteğe bağlı parametreleri geçebilirsiniz:

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)

Bağlantı dizisi oluşturucu sınıfı

Bir builder sınıfı, otomatik kaçış ile akıcı bir API sağlar. Bu yaklaşım, bağlantı parametrelerinin farklı kaynaklardan geldiği kütüphanelerde veya çoklu kiracı uygulamalarında faydalıdır:

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

Çevre tabanlı yapılandırma

Çevre değişkenlerinden

Bağlantı parametrelerini ortam değişkenlerinden okumak, kimlik bilgilerini kaynak kodundan uzak tutar ve yerel geliştirme, konteynerler ve CI/CD boru hatları arasında çalışır. Fonksiyon, ayarlanan değişkenlere göre hangi kimlik doğrulama yönteminin kullanılacağını kontrol eder:

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 ile

Paket, python-dotenv bir .env dosyadan anahtar-değer çiftlerini ortam değişkenlerine yükler, böylece kodunuz, yerel geliştirme ve üretimde kimlik bilgilerini aynı şekilde okur. .env dosyası sürüm kontrolünün dışında tutulur (.gitignore’e ekleyin); dağıtıma alınan ortamlar ise aynı değişkenleri platformun gizli bilgi deposu aracılığıyla sağlar.

pip install python-dotenv ile yükleyin.

Proje kökünüzde bağlantı parametrelerinizle bir .env dosya oluşturun:

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

Sonra bu değerleri betikte yükleyip kullanın:

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() Ortamda zaten ayarlanmış değişkenlerin üzerine yazmaz. Üretimde, platformunuzda aynı değişken isimlerini ayarlayın (örneğin, App Service uygulama ayarları veya konteyner ortamı değişkenleri) ve dosyayı .env tamamen atlayın.

Dosya tabanlı yapılandırma

JSON yapılandırmasından

JSON yapılandırma dosyası, birden fazla ortam (geliştirme, aşamalama, üretim) için bağlantı ayarlarını tek bir yerde tanımlamanızı sağlar. Fonksiyon dosyayı okur, hedef ortamı seçer ve yapılandırılmış ayarlardan bağlantı dizesi'i oluşturur:

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 config'den

YAML yapılandırma dosyaları, JSON'a okunabilir bir alternatiftir. Python projelerinde ve Kubernetes dağıtımlarında yaygın olarak kullanılırlar. Bu yaklaşım, yapılandırılmış bir YAML dosyasından bağlantı ayarlarını okur ve yapılandırmada tanımlanan kimlik doğrulama türüne göre bağlantı dizesi'i oluşturur.

Paketi çalıştırarak pip install pyyamlkurun.

Projenizde bir database.yml dosya oluşturun:

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

Sonra betikte bu ayarları yükleyip kullanın:

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 entegrasyonu

Üretim dağıtımları için, bağlantı kimlik bilgilerini yapılandırma dosyaları veya ortam değişkenleri yerine Azure Key Vault'ta saklayabilirsiniz. Key Vault, merkezi gizli yönetim, erişim denetimi ve otomatik rotasyon sağlar. Gerekli paketleri çalıştırarak pip install azure-keyvault-secrets azure-identitykurun. Tam bir rehber için Quickstart: Azure Key Vault secret client library for Python sayfasına bakınız.

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)

Özel karakterleri işleme

Noktalı virgülleri ve süslü parantezleri kaçışla

Özel karakterler içeren bağlantı dizesi değerlerinden kaçmanız gerekir. Değeri braketlerle {} sarın ve iç kapanış }desteklerini ikiye katlayın:

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

Otomatik kaçışa sahip yapımcı

Bu oluşturucu sınıf, her değeri otomatik olarak sarmalar; böylece çağıran kodun kaçış kurallarını hatırlaması gerekmez. Bağlantı parametreleri kullanıcı formları, yapılandırma API'leri veya değerlerin noktalı virgül veya braket içerebileceği gizli depolar gibi harici girdilerden geldiğinde kullanın:

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

Uygulamanızda dinamik olarak oluşturulmuş bir bağlantı dizesi kullanmadan önce, gerçekten bağlanıp bağlanmadığını kontrol edin. Bu yardımcı fonksiyon hafif bir sorgu dener ve bir boolean sonucu döndürür:

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

İlgili içerik

  • Last updated on