Migrace z SQLite na Microsoft SQL pomocí mssql-python

SQLite je výchozí databáze pro mnoho Python projektů, FastAPI a aplikace Flask. Umožňuje vám rychle vytvořit aplikaci tím, že rozhodnutí o datové platformě odložíte na později. Když přijde čas převést aplikaci do produkce, potřebujete podporu pro současné uživatele, bezpečnost založenou na rolích, vysokou dostupnost, obnovu po havárii a další podnikové funkce. Musíte migrovat na Microsoft SQL pomocí ovladače mssql-python.

Rozdíly v SQL dialektech

Když migrujete ze SQLite na Microsoft SQL, musíte řešit dvě věci: přepsat SQL příkazy pro Transact-SQL (T-SQL) a migrovat data.

Následující tabulka mapuje běžné vzory SQLite na jejich ekvivalenty Microsoft SQL:

SQLite SQL Server (T-SQL) Poznámky
INTEGER PRIMARY KEY AUTOINCREMENT int IDENTITY(1,1) PRIMARY KEY Microsoft SQL používá pro automatické zvyšování hodnot IDENTITY.
TEXT nvarchar(255) nebo nvarchar(max) Vždy uveďte délku. Použití nvarchar pro Unicode.
REAL float nebo decimal(18,2) Použijte decimal pro přesné hodnoty jako peníze.
BLOB varbinary(max) Stejné chování, jiné jméno.
BOOLEAN (uloženo jako INTEGER) bit Žádná z databází nemá nativní booleovský typ. Oba ukládají 0/1.
DATETIME('now') GETDATE() nebo SYSDATETIME() SYSDATETIME() Dává vyšší přesnost.
LIMIT 10 OFFSET 20 OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY Vyžaduje klauzuli ORDER BY .
\|\| (strunová konkace) + nebo CONCAT() CONCAT() zpracovává NULL hodnoty.
IFNULL(a, b) ISNULL(a, b) nebo COALESCE(a, b) COALESCE je standardem ANSI.
GROUP_CONCAT(col) STRING_AGG(col, ',') Dostupné v SQL Server 2017+.
INSERT OR REPLACE INTO Prohlášení MERGE SQLite maže a znovu vkládá; MERGE aktualizuje na místě. Viz následující příklad.
last_insert_rowid() OUTPUT INSERTED.id Použijte OUTPUT v příkazu INSERT. SCOPE_IDENTITY() také funguje, ale vyžaduje samostatné SELECT.
typeof(x) SQL_VARIANT_PROPERTY(x, 'BaseType') Zřídka potřeba při přísném psaní na klávesnici.

CREATE TABLE Příklad

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

Příklady dotazů

Stránkování:

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

Pořadí parametrů je obrácené. Transact-SQL dává OFFSET před FETCH NEXT.

Upsert (vložit nebo aktualizovat):

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

Klauzule USING definuje source.[key] a source.value jako aliasy sloupců. Klauzule WHEN odkazují na tyto aliasy. Stačí jen dva ? markery.

Poslední vložené ID:

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

Aktualizujte kód připojení

Nahraďte sqlite3.connect() za 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;"
    )

Přístup k řadám funguje podobně. Metoda factory v SQLite vrací řádky podobné slovníkům se syntaxí row["column"]. Ovladač mssql-python vrací Row objekty, které podporují stejný přístup ke stringovým klíčům, plus přístup k atributům a indexům:

SQLite (s row_factory):

row["name"]

mssql-python:

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

Styl aktualizace parametrů

Jako parametr používají jak SQLite, tak mssql-python ? , takže většina dotazů funguje bez změn.

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

Jediný rozdíl: SQLite umožňuje pojmenované parametry se syntaxí :name . Ovladač mssql-python místo toho používá %(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})

Migrovat existující data

Důležité aspekty při migraci dat ze SQLite do Microsoft SQL:

  • Microsoft SQL ukládá hodnoty nvarchar(max) a varbinary(max) jako velké objekty (LOB), které se čtou a zapisují pomaleji než data v řádku. Udržujte sloupce řetězců na nvarchar(4000) nebo méně, kdykoli to vaše data dovolí. SQL Server ukládá tyto hodnoty přímo do datového řádku, čímž se vyhýbá zátěži LOB.
  • Mapování typů se řídí pravidly typové afinity SQLite. Prohlédněte si vygenerované tabulky po migraci, abyste zúžili velikost sloupců (například nvarchar(100) místo nvarchar(4000)) nebo přidejte omezení, která SQLite nevynucoval.
  • U SQLite tabulek, které používají TEXT pro ukládání dat, možná budete muset před vložením rozložit hodnoty do Python datetime objektů. Microsoft SQL očekává správné hodnoty datového času, nikoli textové řetězce.

Použijte tento skript k přečtení schématu a dat ze SQLite a vytvoření shodných tabulek v 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()

Rozdíly ve funkcích

Po migraci vaše aplikace získá přístup k funkcím Microsoft SQL, které SQLite nepodporuje:

funkce SQLite SQL Server
Souběžné zápisy Jeden spisovatel po druhém Plná souběžnost se zamykáním na úrovni řádků
Autentizace Pouze oprávnění k souboru ověřování SQL, ověřování Windows, Microsoft Entra ID
Uložené procedury Nepodporováno Plná programovatelnost T-SQL
Encryption Není vestavěná TLS v tranzitu, TDE v klidu
Transakce Body obnovení, základní úrovně izolace Úrovně plné izolace, distribuované transakce
Podpora JSON json_extract() OPENJSON(), JSON_VALUE(), FOR JSON
Fulltextové vyhledávání Rozšíření FTS5 Vestavěné indexování plného textu
Maximální velikost databáze ~281 TB (praktický limit nižší) 524 Petabajtů
Sdílení připojení N/A (v procesu) Integrované v mssql-python

Související obsah

  • Last updated on