使用 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的对应格式：

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 工厂函数返回类似字典的行，并可使用 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表，你可能需要在插入前把数值解析成Pythondatetime对象。 Microsoft SQL 要求的是正确的日期时间值，而不是文本字符串。

使用该脚本读取 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()

功能差异

迁移后，您的应用程序将获得 SQLite 不支持的 Microsoft SQL 功能访问权限：

相关内容

• 安装 mssql-python
• 连接字符串
• 批量复制
• 参数化查询
• 分页
• 使用 mssql-python 配合 Flask