使用 mssql-python 從 SQLite 遷移到 Microsoft SQL

SQLite 是許多 Python 專案、FastAPI 和 Flask 應用程式的預設資料庫。它讓你能透過延遲資料平台的決策，快速建置應用程式。當你的應用程式要投入生產環境時，你需要支援同時使用者、基於角色的安全性、高可用性、災難復原以及其他企業功能。你需要使用 mssql-python 驅動程式遷移到 Microsoft SQL。

SQL 方言差異

當你從 SQLite 遷移到 Microsoft SQL 時，你需要處理兩件事：重寫 Transact-SQL 的 SQL 語句（T-SQL），以及遷移資料。

下表將常見的 SQLite 模式對應至 Microsoft SQL 對應模式：

SQLite	SQL Server （T-SQL）	Notes
`INTEGER PRIMARY KEY AUTOINCREMENT`	`int IDENTITY(1,1) PRIMARY KEY`	Microsoft SQL 會用`IDENTITY`來自動遞增。
`TEXT`	`nvarchar(255)` 或 `nvarchar(max)`	一定要指定長度。使用 `nvarchar` 表示 Unicode。
`REAL`	`float` 或 `decimal(18,2)`	用 `decimal` 來計算像金錢這樣的精確價值。
`BLOB`	`varbinary(max)`	行為一樣，名字不同。
`BOOLEAN` （以整數儲存）	`bit`	這兩個資料庫都沒有原生布林值。兩者都儲存 `0`/`1`。
`DATETIME('now')`	`GETDATE()` 或 `SYSDATETIME()`	`SYSDATETIME()` 提供更高的精度。
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	需要條款 `ORDER BY` 。
`\\|\\|`（字串串接）	`+` 或 `CONCAT()`	`CONCAT()` 處理 `NULL` 數值。
`IFNULL(a, b)`	`ISNULL(a, b)` 或 `COALESCE(a, b)`	`COALESCE` 是ANSI標準。
`GROUP_CONCAT(col)`	`STRING_AGG(col, ',')`	可在 SQL Server 2017+ 中取得。
`INSERT OR REPLACE INTO`	`MERGE` 陳述	SQLite 刪除與重新插入; `MERGE` 更新已到位。請參考以下範例。
`last_insert_rowid()`	`OUTPUT INSERTED.id`	在 `INSERT` 陳述式中使用 `OUTPUT`。 `SCOPE_IDENTITY()` 也可行，但需要另一個 `SELECT`。
`typeof(x)`	`SQL_VARIANT_PROPERTY(x, 'BaseType')`	嚴格輸入時很少需要。

CREATE TABLE 範例

-- SQLite
CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    price REAL DEFAULT 0.0,
    created_at TEXT DEFAULT (datetime('now')),
    is_active BOOLEAN DEFAULT 1
);

-- SQL Server
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'products')
CREATE TABLE products (
    id int IDENTITY(1,1) PRIMARY KEY,
    name nvarchar(100) NOT NULL,
    price decimal(10,2) DEFAULT 0.0,
    created_at datetime2 DEFAULT SYSDATETIME(),
    is_active bit DEFAULT 1
);

查詢範例

頁碼：

SQLite：

cursor.execute("SELECT * FROM products ORDER BY name LIMIT ? OFFSET ?", (10, 20))

MSSQL-python：

cursor.execute(
    "SELECT * FROM Production.Product ORDER BY Name OFFSET ? ROWS FETCH NEXT ? ROWS ONLY",
    (20, 10)
)

參數順序相反。 Transact-SQL 將 OFFSET 放在 FETCH NEXT 前面。

Upsert（插入或更新）：

SQLite：

cursor.execute("""
    INSERT OR REPLACE INTO settings (key, value)
    VALUES (?, ?)
""", (key, value))

MSSQL-python：

cursor.execute("""
    MERGE #Settings AS target
    USING (SELECT ? AS [key], ? AS value) AS source
    ON target.[key] = source.[key]
    WHEN MATCHED THEN UPDATE SET value = source.value
    WHEN NOT MATCHED THEN INSERT ([key], value) VALUES (source.[key], source.value);
""", (key, value))

該 USING 子句定義 source.[key] 和 source.value 作為欄位別名。這些 WHEN 條款就是指那些別名。只需要兩個 ? 標記。

最後輸入的身分證：

SQLite：

cursor.execute("INSERT INTO products (name) VALUES (?)", ("Widget",))
product_id = cursor.lastrowid

MSSQL-python：

cursor.execute(
    "INSERT INTO #Products (Name) OUTPUT INSERTED.ProductID VALUES (?)",
    ("Widget",)
)
product_id = cursor.fetchval()

更新連線代碼

替換 sqlite3.connect() 為 mssql_python.connect()：

SQLite：

import sqlite3

def get_connection():
    conn = sqlite3.connect("myapp.db")
    conn.row_factory = sqlite3.Row
    return conn

MSSQL-python：

import mssql_python

def get_connection():
    return mssql_python.connect(
        "Server=<server>.database.windows.net;"
        "Database=<database>;"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes;"
    )

列存取的運作方式也類似。 SQLite 的 Row 工廠會回傳類似 dict 的資料列，並可使用 row["column"] 語法存取。 mssql-python 驅動程式會傳回支援相同字串鍵存取的 Row 物件，並另外支援屬性與索引存取：

SQLite（含 row_factory）：

row["name"]

MSSQL-python：

row["name"]   # String-key access, like SQLite
row.name      # Attribute access
row[0]        # Index access

更新參數樣式

SQLite 和 mssql-python 都用作 ? 參數標記，因此大多數查詢都能在不做更改的情況下運作。

# Works with both sqlite3 and mssql-python
cursor.execute("SELECT * FROM Production.Product WHERE Name = ?", ("Adjustable Race",))

唯一的差別是：SQLite 允許有語法的 :name 命名參數。 mssql-python 驅動程式則使用 %(name)s 。

SQLite：

cursor.execute("SELECT * FROM products WHERE id = :id", {"id": 42})

MSSQL-python：

cursor.execute("SELECT * FROM Production.Product WHERE ProductID = %(id)s", {"id": 42})

遷移現有資料

從 SQLite 遷移到 Microsoft SQL 時的重要考量：

Microsoft SQL 將 nvarchar（max）和 varbinary（max）值儲存為大型物件（LOB），這些物件的讀寫速度比列內資料慢。只要資料允許，就把字串欄位設在 nvarchar（4000） 或更低。 SQL Server 將這些值直接儲存在資料列中，避免了 LOB 的開銷。
型別映射遵循 SQLite 的型別親和規則。遷移後檢視產生的資料表，以調整欄位大小（例如 nvarchar（100） 取代 nvarchar（4000））或新增 SQLite 未強制執行的限制。
對於使用 TEXT 來儲存日期的 SQLite 表格，你可能需要先將數值解析成 Python datetime 物件再插入。 Microsoft SQL 期望的是正確的 datetime 值，而不是文字字串。

使用此腳本讀取 SQLite 的結構與資料，並在 Microsoft SQL 中建立相符的資料表：

import sqlite3
import mssql_python

# SQLite type affinity -> Microsoft SQL type
# Keywords follow SQLite's type affinity rules
TYPE_MAP = {
    "INT": "bigint",
    "CHAR": "nvarchar(4000)",
    "CLOB": "nvarchar(max)",
    "TEXT": "nvarchar(4000)",
    "BLOB": "varbinary(max)",
    "REAL": "float",
    "FLOA": "float",
    "DOUB": "float",
}


def map_type(sqlite_type: str) -> str:
    """Map a SQLite column type to a Microsoft SQL type."""
    upper = (sqlite_type or "TEXT").upper()
    for prefix, sql_type in TYPE_MAP.items():
        if prefix in upper:
            return sql_type
    return "decimal(18,6)"  # NUMERIC affinity (default)


# Connect to both databases
sqlite_conn = sqlite3.connect("myapp.db")
sql_conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;"
)

# Get list of tables from SQLite
sqlite_cur = sqlite_conn.cursor()
sqlite_cur.execute(
    "SELECT name FROM sqlite_master "
    "WHERE type='table' AND name NOT LIKE 'sqlite_%'"
)
tables = [row[0] for row in sqlite_cur.fetchall()]

sql_cursor = sql_conn.cursor()

for table in tables:
    # Read column info from SQLite
    sqlite_cur.execute(f"PRAGMA table_info([{table}])")
    columns = sqlite_cur.fetchall()
    # columns: (cid, name, type, notnull, default_value, pk)

    # Build CREATE TABLE statement
    col_defs = []
    for col in columns:
        name, col_type, notnull, pk = col[1], col[2], col[3], col[5]
        sql_type = map_type(col_type)
        parts = [f"[{name}] {sql_type}"]
        if notnull:
            parts.append("NOT NULL")
        if pk:
            parts.append("PRIMARY KEY")
        col_defs.append(" ".join(parts))

    create_sql = f"CREATE TABLE [{table}] ({', '.join(col_defs)})"
    sql_cursor.execute(
        f"IF OBJECT_ID('{table}','U') IS NULL " + create_sql
    )
    sql_conn.commit()

    # Read all rows from SQLite
    sqlite_cur.execute(f"SELECT * FROM [{table}]")
    rows = sqlite_cur.fetchall()

    if not rows:
        print(f"  {table}: created (empty)")
        continue

    # Use bulkcopy for fast insert
    result = sql_cursor.bulkcopy(table, rows)
    print(f"  {table}: {result['rows_copied']} rows copied")

sql_conn.commit()
sqlite_conn.close()
sql_conn.close()

功能差異

遷移後，您的應用程式將獲得 Microsoft SQL 功能，而 SQLite 不支援：

Feature	SQLite	SQL Server
同時寫入	一次只寫一位作家	具備資料列層級鎖定的完整並行處理
Authentication	僅限檔案權限	SQL 驗證、Windows 驗證、Microsoft Entra ID
預存程序	不支援	完整的 T-SQL 可程式化性
Encryption	不是內建的	TLS在運送中，TDE靜止
Transactions	儲存點、基本隔離層級	完整隔離層級，分散式交易
JSON 支援	`json_extract()`	`OPENJSON()`、`JSON_VALUE()`、`FOR JSON`
全文搜尋	FTS5 擴充	內建全文索引
資料庫大小上限	~281 TB（實際限制較低）	524 PB
連線池化	不適用（處理中）	內建於 mssql-python

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-22

使用 mssql-python 從 SQLite 遷移到 Microsoft SQL

SQL 方言差異

CREATE TABLE 範例

查詢範例

更新連線代碼

更新參數樣式

遷移現有資料

功能差異

相關內容

意見反應

其他資源