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) 비고
INTEGER PRIMARY KEY AUTOINCREMENT int IDENTITY(1,1) PRIMARY KEY Microsoft SQL에서는 자동 증가에 IDENTITY를 사용합니다.
TEXT nvarchar(255) 또는 nvarchar(max) 항상 길이를 명시하세요. 유니코드에는 nvarchar를 사용하세요.
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-파이썬:

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

매개변수 순서가 반대입니다. Transact-SQL은 OFFSET를 FETCH NEXT 앞에 둡니다.

업서트(삽입 또는 업데이트):

SQLite:

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

MSSQL-파이썬:

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-파이썬:

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-파이썬:

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"] 구문을 사용하는 dict와 같은 행을 반환합니다. mssql-python 드라이버는 동일한 문자열 키 접근과 속성 및 인덱스 접근을 지원하는 객체를 반환합니다 Row :

SQLite(row_factory 포함):

row["name"]

MSSQL-파이썬:

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-파이썬:

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(4000) 대신 nvarchar(100) 사용) 또는 SQLite가 강제하지 않은 제약 조건을 추가하세요.
  • SQLite 테이블이 TEXT를 사용해 날짜를 저장하는 경우, 값을 삽입하기 전에 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()

기능 차이점

마이그레이션 후, 애플리케이션은 SQLite가 지원하지 않는 Microsoft SQL 기능에 접근할 수 있게 됩니다:

특징 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 (실제 한계는 더 낮음) 524PB
연결 풀링 (Connection Pooling) 해당 없음 (진행 중) mssql-python에 기본 제공

관련 콘텐츠

  • Last updated on