Membangun string koneksi secara terprogram

Banyak aplikasi perlu membangun string koneksi secara dinamis daripada menyimpannya sebagai nilai konfigurasi statis. Pilih pendekatan yang sesuai dengan penyebaran Anda:

• Variabel lingkungan: Terbaik untuk kontainer, CI/CD, dan aplikasi 12 faktor. Sederhana dan didukung secara luas.
• File konfigurasi JSON/YAML: Terbaik untuk aplikasi dengan beberapa lingkungan (dev, staging, prod) yang memerlukan konfigurasi terstruktur.
• Azure Key Vault: Terbaik untuk penyebaran produksi di mana rahasia harus dikelola dan diaudit secara terpusat.
• Kelas Builder: Paling cocok untuk pustaka atau kerangka kerja yang perlu menyusun string koneksi dari input pengguna dengan escape otomatis.

Konstruksi string dasar

Gunakan f-string

f-string merupakan pendekatan yang umum untuk skrip dan prototipe sederhana. Hindari pola ini ketika nilai berasal dari input pengguna, karena nilai berbahaya seperti dapat mydb;Server=evil.com mengubah target koneksi:

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)

Gunakan join

Pendekatan join ini memisahkan pasangan kunci-nilai ke dalam sebuah panggilan fungsi bergaya kamus, yang lebih mudah dibaca dan dipelihara daripada f-string yang panjang. Ini juga memfilter None nilai secara otomatis, sehingga Anda dapat meneruskan parameter opsional tanpa logika bersyarat tambahan:

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)

Kelas pembuat string koneksi

Kelas builder menyediakan API yang lancar dengan pelepasan otomatis. Pendekatan ini berguna di pustaka atau aplikasi multipenyewa di mana parameter koneksi berasal dari sumber yang berbeda:

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

Konfigurasi berbasis lingkungan

Dari variabel lingkungan

Membaca parameter koneksi dari variabel lingkungan mencegah kredensial disimpan dalam kode sumber dan berfungsi baik untuk pengembangan lokal, kontainer, dan pipeline CI/CD. Fungsi ini memeriksa metode autentikasi mana yang akan digunakan berdasarkan variabel yang ditetapkan:

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

Dengan python-dotenv

Paket python-dotenv memuat pasangan kunci-nilai dari file .env ke variabel lingkungan sehingga kode Anda membaca kredensial dengan cara yang sama baik di lingkungan pengembangan lokal maupun produksi. File .env tetap berada di luar sistem kontrol versi (tambahkan ke .gitignore), sementara lingkungan deployment memasok variabel yang sama melalui penyimpanan secret platform mereka.

Instal dengan pip install python-dotenv.

Buat .env file di root project Anda dengan parameter koneksi Anda:

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

Kemudian muat dan gunakan nilai-nilai tersebut dalam skrip Anda:

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() tidak menimpa variabel yang sudah ditetapkan di lingkungan sistem. Dalam produksi, atur nama variabel yang sama melalui platform Anda (misalnya, pengaturan aplikasi App Service atau variabel lingkungan kontainer) dan lewati .env file sepenuhnya.

Mengonfigurasi berbasis file

Dari konfigurasi JSON

File konfigurasi JSON memungkinkan Anda menentukan pengaturan koneksi untuk beberapa lingkungan (pengembangan, pementasan, produksi) di satu tempat. Fungsi membaca file, memilih lingkungan target, dan membangun string koneksi dari pengaturan terstruktur:

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

Dari konfigurasi YAML

File konfigurasi YAML adalah alternatif yang dapat dibaca untuk JSON. Mereka biasanya digunakan dalam proyek Python dan penyebaran Kubernetes. Pendekatan ini membaca pengaturan koneksi dari file YAML terstruktur dan membangun string koneksi berdasarkan jenis autentikasi yang ditentukan dalam konfigurasi.

Instal paket dengan menjalankan pip install pyyaml.

Buat database.yml file di project Anda:

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

Kemudian muat dan gunakan pengaturan tersebut dalam skrip Anda:

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

integrasi Azure Key Vault

Untuk penyebaran produksi, simpan kredensial koneksi di Azure Key Vault daripada di file konfigurasi atau variabel lingkungan. Key Vault menyediakan manajemen rahasia terpusat, audit akses, dan rotasi otomatis. Instal paket yang diperlukan dengan menjalankan pip install azure-keyvault-secrets azure-identity. Untuk panduan lengkap, lihat Mulai Cepat: pustaka klien rahasia Azure Key Vault untuk 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)

Menangani karakter khusus

Melarikan diri dengan titik koma dan kurung gigi

Anda perlu melarikan diri dari nilai string koneksi yang berisi karakter khusus. Bungkus nilai dalam kurung kurawal {} dan gandakan setiap kurung kurawal penutup di dalamnya }:

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

Pembuat dengan escape otomatis

Kelas pembangun ini secara otomatis membungkus setiap nilai, sehingga kode yang memanggilnya tidak perlu mengingat aturan escape. Gunakan saat parameter koneksi berasal dari input eksternal seperti formulir pengguna, API konfigurasi, atau penyimpanan rahasia di mana nilai mungkin berisi titik koma atau kurung kurawal:

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

Sebelum menggunakan string koneksi yang dibuat secara dinamis di aplikasi Anda, verifikasi bahwa itu benar-benar terhubung. Fungsi pembantu ini mencoba kueri ringan dan mengembalikan hasil boolean:

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

Konten terkait

• String koneksi
• Autentikasi Microsoft Entra
• Manajemen koneksi