49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
import os
|
|
from typing import Optional
|
|
|
|
import pyodbc
|
|
from dotenv import load_dotenv
|
|
|
|
# Force reload environment variables from .env, ignoring system vars.
|
|
load_dotenv(override=True)
|
|
|
|
|
|
def _get_required_env(name: str) -> str:
|
|
value = os.getenv(name, "").strip()
|
|
if not value:
|
|
raise RuntimeError(f"Missing {name} environment variable.")
|
|
return value
|
|
|
|
|
|
def _build_conn_str() -> str:
|
|
explicit = os.getenv("MSSQL_CONN_STR", "").strip()
|
|
if explicit:
|
|
return explicit
|
|
|
|
driver = os.getenv("MSSQL_DRIVER", "ODBC Driver 18 for SQL Server").strip()
|
|
server = _get_required_env("MSSQL_SERVER")
|
|
database = _get_required_env("MSSQL_DATABASE")
|
|
user = _get_required_env("MSSQL_USER")
|
|
password = _get_required_env("MSSQL_PASSWORD")
|
|
encrypt = os.getenv("MSSQL_ENCRYPT", "yes").strip() or "yes"
|
|
trust_cert = os.getenv("MSSQL_TRUST_SERVER_CERT", "yes").strip() or "yes"
|
|
|
|
return (
|
|
f"DRIVER={{{driver}}};"
|
|
f"SERVER={server};"
|
|
f"DATABASE={database};"
|
|
f"UID={user};"
|
|
f"PWD={password};"
|
|
f"Encrypt={encrypt};"
|
|
f"TrustServerCertificate={trust_cert};"
|
|
)
|
|
|
|
|
|
def connect_to_mssql() -> Optional[pyodbc.Connection]:
|
|
"""Establish a connection to the Microsoft SQL Server database."""
|
|
try:
|
|
return pyodbc.connect(_build_conn_str())
|
|
except pyodbc.Error as exc:
|
|
print(f"Error connecting to MSSQL database: {exc}")
|
|
return None
|