mssql-python을 이용해 PostgreSQL에서 Microsoft SQL로 마이그레이션하기

많은 Python 팀이 먼저 PostgreSQL을 배웁니다. 작업 중에 시간 테이블, 전체 MERGE 의미론, 컬럼스토어 인덱스 같은 기능이 필요하다면, Microsoft SQL로 마이그레이션하세요. 이 가이드는 Python 애플리케이션을 PostgreSQL(psycopg2 또는 psycopg3 사용)에서 드라이버를 mssql-python 사용해 Microsoft SQL로 옮기는 주요 결정과 코드 변경 사항을 다룹니다.

비고

Azure Database for PostgreSQL에서 마이그레이션하는 경우, 두 서비스 모두 Microsoft Entra 인증과 관리 신원을 지원합니다. 이 가이드의 코드 변경은 PostgreSQL 소스가 자체 관리인지 Azure 호스팅인지와 관계없이 적용됩니다.

Microsoft SQL로 전환함으로써 얻는 이점

Microsoft SQL은 운영 워크로드의 보안, 준수 및 운영을 단순화하는 기능을 포함하고 있습니다. 마이그레이션을 시작하기 전에 이 기능들을 이해하여 전환 과정에서 활용할 수 있도록 하세요:

동적 데이터 마스킹 과 행 수준 보안. 완전한 권한이 필요하지 않은 사용자는 열을 가리고, 보안 정책으로 행 가시성을 제한합니다. 이 기능들은 어떤 드라이버에도 적용됩니다.
시간 테이블 (시스템 버전) Microsoft SQL은 행 기록을 자동으로 추적합니다. 트리거도 없고, 감사 테이블도 없고, 애플리케이션 코드도 없어요.
완전한 MERGE 의미론입니다. 단일 구문은 감사 추적을 위한 OUTPUT 절과 함께 INSERT, UPDATE, DELETE를 처리합니다. PostgreSQL의 절은 ON CONFLICT 단일 제약 조건에 대한 삽입 또는 업데이트만 포함합니다.
열 저장소 인덱스. 하이브리드 OLTP/분석 작업을 위해 기존 테이블에 열형 저장소를 추가하세요. 별도의 분석 데이터베이스가 필요하지 않습니다.
Microsoft Entra ID 인증. 관리 신원, 서비스 주체, 또는 인터랙티브 로그인과 연결하세요. Azure Database for PostgreSQL은 Microsoft Entra 인증도 지원하므로 이미 사용 중이라면 전환이 간단합니다.

드라이버 설치

시작하기 전에 Python 3.10 이상과 목표 SQL 데이터베이스를 갖추고 있는지 확인하세요.

SQL 데이터베이스 만들기

다음 플랫폼 중 하나에서 SQL 데이터베이스를 생성하거나 연결하세요:

Azure Portal을 사용하여 몇 분 안에 SQL 데이터베이스 만들기

PostgreSQL 드라이버는 외부 네이티브 라이브러리가 필요합니다.

# psycopg2 requires pg_config, libpq-dev, and platform-specific build tools
sudo apt-get install libpq-dev  # Debian/Ubuntu
pip install psycopg2

mssql-python 드라이버는 자체 네이티브 레이어를 포함합니다. Windows에서는 외부 드라이버 관리자나 시스템 패키지가 필요하지 않습니다.

pip install mssql-python

리눅스와 macOS에서는 설치 항목에 문서화된 소수의 시스템 라이브러리를 설치하세요. pg_config 또는 libpq-dev에 해당하는 것은 없습니다.

연결 코드 업데이트

다음 섹션에서는 연결 문자열, 인증, 컨텍스트 관리자, 풀링에 대한 주요 변경 사항을 다룹니다.

연결 문자열

psycopg2는 DSN 문자열이나 키워드 인수를 사용합니다.

import psycopg2

conn = psycopg2.connect(
    host="<server>",
    dbname="<database>",
    user="<username>",
    password="<password>"
)

mssql-python은 키워드 인수도 지원하여, 비밀번호에 , @, ; 또는 문자가 포함될 {}때 SQLAlchemy 연결 문자열에서 자주 발생하는 URL 인코딩 문제를 피할 수 있습니다.

import mssql_python

conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="<database>",
    authentication="ActiveDirectoryDefault",
    encrypt="yes"
)

또는 연결 문자열을 사용하세요.

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

연결 문자열 키워드의 전체 집합은 Connection strings를 참조하세요.

Authentication

PostgreSQL 인증은 일반적으로 사용자 이름과 비밀번호를 사용하는 규칙을 사용합니다 pg_hba.conf . Azure Database for PostgreSQL은 Microsoft Entra 인증도 지원합니다. Microsoft SQL은 단일 연결 키워드를 통해 여러 인증 모드를 지원합니다:

PostgreSQL 접근법	MSSQL-파이썬 동등
사용자 이름 및 암호	`UID=...;PWD=...;`
SSL/TLS 암호화	`Encrypt=yes;`(기본적으로 Azure SQL에 대해 활성화됨)
Entra auth (Azure PostgreSQL)	`Authentication=ActiveDirectoryDefault;` (비밀번호 없음)
Managed identity (Azure PostgreSQL)	`Authentication=ActiveDirectoryMSI;`
Service principal (Azure PostgreSQL)	`Authentication=ActiveDirectoryServicePrincipal;`

로컬 개발에는 ActiveDirectoryDefault를 사용합니다. Azure CLI, 환경 변수, 관리 신원을 자동으로 연결합니다. 프로덕션 환경에서는 느린 자격 증명 체인 순회를 피하기 위해 ActiveDirectoryMSI(관리형 ID) 또는 ActiveDirectoryServicePrincipal 같은 특정 모드를 사용하세요. Microsoft Entra 인증은 7가지 인증 모드 모두에 대해 참고하세요.

컨텍스트 관리자

두 드라이버 모두 컨텍스트 관리자를 지원하지만, 동작은 다릅니다:

Psycopg2는 with conn: 성공 시 커밋하고 예외 시 롤백하지만, 연결을 종료 하지는 않습니다 :

with psycopg2.connect(...) as conn:
    with conn.cursor() as cur:
        cur.execute("INSERT INTO ...")
    # conn.commit() happens automatically on success
# Connection is still open here
conn.close()  # Must close explicitly

MSSQL-python은 with conn: 종료 시 연결을 닫습니다. 커밋되지 않은 작업은 롤백됩니다:

with mssql_python.connect(...) as conn:
    with conn.cursor() as cursor:
        cursor.execute("INSERT INTO ...")
    conn.commit()
# Connection is closed here

연결 풀링 (Connection Pooling)

PSYCOPG2는 명시적인 연결 풀 설정 및 관리를 요구합니다.

from psycopg2 import pool

connection_pool = pool.ThreadedConnectionPool(1, 10, dsn="...")
conn = connection_pool.getconn()
# ... use conn ...
connection_pool.putconn(conn)

mssql-python 드라이버는 기본적으로 기본 제공 풀링이 활성화되어 있습니다. 설정이 필요 없습니다.

# Pooling is automatic. Each connect() call reuses pooled connections.
conn = mssql_python.connect(...)

기본 설정이 워크로드에 맞지 않으면 풀 크기를 설정하세요.

import mssql_python

mssql_python.pooling(max_size=20, idle_timeout=300)

수영장 크기 설정과 수영장 고갈에 대한 문제 해결에 대한 안내는 연결 풀링을 참조하세요.

SQL 방언 차이점

다음 표는 일반적인 PostgreSQL 패턴을 Transact-SQL(T-SQL) 대응 문자와 매핑합니다:

PostgreSQL	SQL Server (T-SQL)	비고
`SERIAL` / `BIGSERIAL`	`int IDENTITY(1,1)`	Microsoft SQL에서는 자동 증가에 `IDENTITY`를 사용합니다.
`TEXT`	`nvarchar(max)`	유니코드에는 `nvarchar`를 사용하세요. 데이터가 허용하는 경우 `nvarchar(4000)` 또는 그보다 짧은 것을 선호하세요.
`BOOLEAN`	`bit`	PostgreSQL은 `true`/`false`를 허용합니다; Microsoft SQL은 `1`/`0`를 사용합니다.
`BYTEA`	`varbinary(max)`	같은 개념이지만 이름은 다르다.
`JSONB`	`nvarchar(max)` JSON 함수를 포함해	Microsoft SQL은 JSON을 텍스트로 저장하고 .로 `ISJSON()`검증합니다. JSON 데이터를 참조하세요.
`TIMESTAMP WITH TIME ZONE`	`datetimeoffset`	둘 다 오프셋을 저장합니다. 날짜 처리 항목을 참조하세요.
`INTERVAL`	직접 동등한 항목 없음	`DATEADD()` 및 `DATEDIFF()`로 계산합니다.
`ARRAY`	직접 동등한 항목 없음	별도의 테이블, JSON 배열, 또는 `STRING_SPLIT()`.
`UUID`	`uniqueidentifier`	`mssql-python` 드라이버는 `uuid.UUID`를 기본적으로 매핑합니다. 모듈 구성(Module configuration)을 참조하세요.
`NOW()` / `CURRENT_TIMESTAMP`	`GETDATE()` 또는 `SYSDATETIME()`	`SYSDATETIME()` 더 높은 정밀도를 제공합니다.
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	`ORDER BY` 조항이 필요합니다.
`\\|\\|` (문자열 연결)	`+` 또는 `CONCAT()`	`CONCAT()` 값을 처리합니다 `NULL` .
`COALESCE(a, b)`	`COALESCE(a, b)` 또는 `ISNULL(a, b)`	`COALESCE` 두 언어 모두 동일합니다.
`string_agg(col, ',')`	`STRING_AGG(col, ',')`	SQL Server 2017+에서 이용 가능합니다.
`RETURNING id`	`OUTPUT INSERTED.id`	`INSERT`, `UPDATE` 또는 `DELETE` 문에서 `OUTPUT`을(를) 사용하세요.
`ON CONFLICT ... DO UPDATE`	`MERGE` 진술	`MERGE`는 하나의 문에서 INSERT + UPDATE + DELETE를 지원합니다. 쿼리 재작성 패턴을 참조하세요.
`EXPLAIN ANALYZE`	`SET STATISTICS IO ON; SET STATISTICS TIME ON;`	또는 SSMS / Azure Data Studio에서 실행 계획을 사용하세요.
`\d tablename`	`sp_help 'tablename'`	또는 `INFORMATION_SCHEMA.COLUMNS`를 조회합니다.
`pg_dump`	`bcp`, `BACKUP DATABASE`	Python에서 프로그래밍 데이터 로딩에 사용됩니다`bulkcopy()`.

CREATE TABLE 예제

PostgreSQL:

CREATE TABLE IF NOT EXISTS products (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    price NUMERIC(10, 2) DEFAULT 0.0,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    metadata JSONB,
    is_active BOOLEAN DEFAULT TRUE
);

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 datetimeoffset DEFAULT SYSDATETIMEOFFSET(),
    metadata nvarchar(max),
    is_active bit DEFAULT 1
);

쿼리 재작성 패턴

다음 섹션에서는 일반적인 PostgreSQL 쿼리 패턴과 이에 해당하는 T-SQL을 보여줍니다.

페이지 나누기

PostgreSQL:

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

MSSQL-파이썬:

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

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

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

PostgreSQL은 ON CONFLICT 단일 제약 조건으로 삽입 또는 업데이트를 처리합니다:

cursor.execute("""
    INSERT INTO settings (key, value)
    VALUES (%s, %s)
    ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value
""", (key, value))

Microsoft SQL은 MERGE , , INSERT, UPDATE 를 하나의 문장에서 처리DELETE합니다. 매개변수 별칭을 사용하는 USING 절을 사용하세요:

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

대량 업서트의 경우 bulkcopy()를 사용해 행을 임시 테이블에 저장한 다음, 그 테이블에서 MERGE합니다. 자세한 내용은 스테이징 테이블을 사용한 대량 업서트를 참조하세요.

삽입된 ID 가져오기

PostgreSQL:

cursor.execute(
    "INSERT INTO products (name) VALUES (%s) RETURNING id",
    ("Widget",)
)
product_id = cursor.fetchone()[0]

MSSQL-파이썬:

cursor.execute(
    "INSERT INTO #Products (Name) OUTPUT INSERTED.ProductID VALUES (%(name)s)",
    {"name": "Widget"}
)
product_id = cursor.fetchval()

OUTPUT INSERTED은(는) INSERT, UPDATE 및 DELETE 문과 함께 작동합니다. 여러 열을 반환할 수 있습니다.

매개 변수 표식

Psycopg2는 위치 매개변수와 %s 이름 있는 매개변수를 사용합니다%(name)s. 드라이버는 mssql-python 위치 기반에는 ?, 이름 지정에는 %(name)s를 사용합니다:

Psycopg2:

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

MSSQL-파이썬:

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

트랜잭션과 자동커밋의 차이점

PostgreSQL(psycopg2)는 첫 번째 명령 실행 시 트랜잭션을 자동으로 시작하며, 명시적으로 commit()해야 합니다:

conn = psycopg2.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()

드라이버도 mssql-python 기본적으로 같은 방식으로 작동합니다. 자동 커밋은 꺼져 있고, 명시적으로 다음과 같이 호출 commit() 합니다:

conn = mssql_python.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()

자동 커밋을 활성화하려면:

Psycopg2:

conn = psycopg2.connect(...)
conn.autocommit = True

MSSQL-파이썬:

conn = mssql_python.connect(..., autocommit=True)
# or: conn.autocommit = True

격리 수준, 저장 지점, 교착 상태 재시도 패턴에 대해서는 트랜잭션 관리를 참조하세요.

유형 고려사항

다음 섹션에서는 PostgreSQL과 Microsoft SQL 간의 가장 흔한 타입 매핑 차이점을 다룹니다.

JSON

PostgreSQL은 인덱싱과 쿼리 연산자(->, ->>, @>)를 갖춘 네이티브 JSONB를 지원합니다. Microsoft SQL은 JSON을 nvarchar(max)로 저장하며 쿼리 기능을 제공합니다:

PostgreSQL	SQL Server
`data->>'name'`	`JSON_VALUE(data, '$.name')`
`data->'items'`	`JSON_QUERY(data, '$.items')`
`data @> '{"active": true}'`	`JSON_VALUE(data, '$.active') = 'true'`
`jsonb_array_length(data)`	`(SELECT COUNT(*) FROM OPENJSON(data))`

Python에서는 두 방법 모두 직렬화를 사용합니다json.dumps():

import json

cursor.execute(
    "INSERT INTO #Settings ([key], data) VALUES (%(key)s, %(data)s)",
    {"key": "config", "data": json.dumps({"theme": "dark", "lang": "en"})}
)

JSON 저장 및 쿼리 패턴에 대한 전체 안내는 JSON 데이터를 참조하세요.

UUID

PostgreSQL과 mssql-python 모두 네이티브로 매핑 uuid.UUID 됩니다:

import uuid

cursor.execute(
    "INSERT INTO #Events (EventID, Name) VALUES (%(event_id)s, %(name)s)",
    {"event_id": uuid.uuid4(), "name": "signup"}
)

연결 옵션에 대해서는 native_uuid 참조하세요.

날짜, 시간, 시간대

PostgreSQL은 TIMESTAMPTZ 저장소에서 UTC로 변환됩니다. Microsoft SQL은 datetimeoffset 원본 오프셋을 보존합니다:

from datetime import datetime, timezone, timedelta

eastern = timezone(timedelta(hours=-5))
dt = datetime(2025, 6, 15, 14, 30, tzinfo=eastern)

# PostgreSQL stores as UTC: 2025-06-15 19:30:00+00
# SQL Server stores as-is: 2025-06-15 14:30:00-05:00
cursor.execute("INSERT INTO #Events (EventTime) VALUES (%(event_time)s)", {"event_time": dt})

일관된 UTC 저장이 필요하다면, 다음을 삽입하기 전에 Python에서 변환하세요:

dt_utc = dt.astimezone(timezone.utc)
cursor.execute("INSERT INTO #Events (EventTime) VALUES (%(event_time)s)", {"event_time": dt_utc})

전체 타입 매핑에 대해서는 Datetime 처리 를 참조하세요.

배열

PostgreSQL은 네이티브 배열 열(INTEGER[], TEXT[])을 지원합니다. Microsoft SQL에는 배열 타입이 없습니다. 일반적인 대안:

별도의 표 (정규화됨). 쿼리 가능하고 인덱싱된 데이터에 가장 적합합니다.
nvarchar(max)에 저장된 JSON 배열. 불투명한 메타데이터에 적합합니다.
STRING_SPLIT()이(가) 포함된 쉼표로 구분된 문자열. 단순하지만 한계가 있습니다.

# Option 1: Normalized table
cursor.execute("INSERT INTO #ProductTags (ProductID, Tag) VALUES (%(product_id)s, %(tag)s)", {"product_id": 1, "tag": "electronics"})
cursor.execute("INSERT INTO #ProductTags (ProductID, Tag) VALUES (%(product_id)s, %(tag)s)", {"product_id": 1, "tag": "sale"})

# Option 2: JSON array
import json
tags = json.dumps(["electronics", "sale"])
cursor.execute("INSERT INTO #Products (Name, Tags) VALUES (%(name)s, %(tags)s)", {"name": "Widget", "tags": tags})

Unicode

PostgreSQL은 기본적으로 모든 텍스트를 UTF-8로 저장합니다. Microsoft SQL은 varchar(코드 페이지 인코딩)와 nvarchar(UTF-16)를 구분합니다. 드라이버는 mssql-python 기본적으로 Python str 값을 nvarchar로 보내기 때문에 유니코드 텍스트는 추가 설정 없이도 작동합니다. 스키마가 varchar 컬럼을 사용하고 암묵적 변환을 피해야 한다면, 를 사용하여 setinputsizes() 컬럼 타입을 지정하세요. 인코딩 세부 사항은 문자열 및 유니코드 데이터를 참조하세요.

대량 로딩 및 데이터 이동

PostgreSQL은 대량 작업에 사용됩니다 COPY . MSSQL-python은 다음과 같이 제공합니다 bulkcopy():

Psycopg2:

with open("data.csv") as f:
    cursor.copy_expert("COPY products FROM STDIN CSV HEADER", f)

MSSQL-파이썬:

import csv

with open("data.csv", newline="") as f:
    reader = csv.reader(f)
    next(reader)  # Skip header
    rows = [tuple(row) for row in reader]

cursor.bulkcopy("##Products", rows)

큰 파일의 경우, 전체 파일을 메모리에 로드하지 않도록 생성기를 사용하세요:

import csv

def csv_rows(path):
    with open(path, newline="") as f:
        reader = csv.reader(f)
        next(reader)  # Skip header
        for row in reader:
            yield tuple(row)

cursor.bulkcopy("##Products", csv_rows("data.csv"), batch_size=5000)

컬럼 매핑, 아이덴티티 처리, 성능 팁은 대량 복사 작업을 참조하세요.

스키마 및 데이터 마이그레이션

기존 PostgreSQL 데이터베이스를 마이그레이션할 때 이 방법을 사용하세요:

스키마를 내보내세요. DDL을 가져오려면 pg_dump --schema-only을 사용하세요. 옵션 세부사항과 예외 사례(소유권, 권한, 확장, 필터링)는 PostgreSQL pg_dump 참고 자료를 참조하세요. SQL 방언 차이 표를 사용해 DDL을 다시 작성하세요.
Microsoft SQL로 테이블을 만드세요. 다시 작성된 DDL을 목표 데이터베이스에 대비해 실행하세요.
데이터를 내보냅니다. pg_dump --data-only --format=csv를 사용하거나 psycopg2로 각 테이블을 쿼리하세요. 대규모 데이터셋과 호환성 스위치에 대해서는 PostgreSQL pg_dump 문서, 특히 옵션 섹션을 검토하세요.
bulkcopy로 데이터를 로드합니다. 카탈로그에서 목적지 열 순서를 읽어서 테이블마다 열 목록을 하드코딩하지 않게 하고, 각 테이블을 Microsoft SQL로 스트리밍하세요. 다음은 예제 스크립트입니다.

import json
import psycopg2
from psycopg2 import sql
import mssql_python

pg_conn = psycopg2.connect(host="<pgserver>", dbname="<database>", user="<username>", password="<password>")
sql_conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="<database>",
    authentication="ActiveDirectoryDefault",
    encrypt="yes"
)

def table_columns(cursor, table):
    """Return the ordered column names and identity column from the catalog."""
    cursor.execute(
        "SELECT c.name, c.is_identity FROM sys.columns AS c "
        "WHERE c.object_id = OBJECT_ID(?) ORDER BY c.column_id",
        (table,)
    )
    columns, identity = [], None
    for name, is_identity in cursor.fetchall():
        columns.append(name)
        if is_identity:
            identity = name
    return columns, identity

def parse_pg_table_name(qualified_name):
    """Split a PostgreSQL table name into schema and table parts."""
    if "." in qualified_name:
        schema_name, table_name = qualified_name.split(".", 1)
    else:
        schema_name, table_name = "public", qualified_name
    return schema_name, table_name

def parse_sql_table_name(qualified_name):
    """Split a SQL Server table name into schema and table parts."""
    if "." in qualified_name:
        schema_name, table_name = qualified_name.split(".", 1)
    else:
        schema_name, table_name = "dbo", qualified_name
    return schema_name, table_name

def dependency_order(pg_cursor, table_names, schema_name="public"):
    """Topologically sort tables by foreign key dependencies."""
    table_set = set(table_names)
    incoming = {name: 0 for name in table_set}
    edges = {name: set() for name in table_set}

    pg_cursor.execute(
        """
        SELECT
            child.relname AS child_table,
            parent.relname AS parent_table
        FROM pg_constraint c
        JOIN pg_class child ON c.conrelid = child.oid
        JOIN pg_namespace child_ns ON child.relnamespace = child_ns.oid
        JOIN pg_class parent ON c.confrelid = parent.oid
        JOIN pg_namespace parent_ns ON parent.relnamespace = parent_ns.oid
        WHERE c.contype = 'f'
          AND child_ns.nspname = %s
          AND parent_ns.nspname = %s
        """,
        (schema_name, schema_name),
    )

    for child, parent in pg_cursor.fetchall():
        if child in table_set and parent in table_set and child != parent:
            if child not in edges[parent]:
                edges[parent].add(child)
                incoming[child] += 1

    ready = sorted([name for name, degree in incoming.items() if degree == 0])
    ordered = []

    while ready:
        current = ready.pop(0)
        ordered.append(current)
        for neighbor in sorted(edges[current]):
            incoming[neighbor] -= 1
            if incoming[neighbor] == 0:
                ready.append(neighbor)
        ready.sort()

    # If cycles remain, process remaining tables alphabetically.
    if len(ordered) < len(table_set):
        remaining = sorted(table_set - set(ordered))
        ordered.extend(remaining)

    return ordered

def discover_table_pairs(pg_cursor, sql_cursor, pg_schema="public", sql_schema="dbo"):
    """Find tables that exist in both PostgreSQL and SQL Server, in dependency order."""
    pg_cursor.execute(
        """
        SELECT table_name
        FROM information_schema.tables
        WHERE table_schema = %s AND table_type = 'BASE TABLE'
        """,
        (pg_schema,),
    )
    pg_tables = {row[0] for row in pg_cursor.fetchall()}

    sql_cursor.execute(
        """
        SELECT t.name
        FROM sys.tables AS t
        JOIN sys.schemas AS s ON t.schema_id = s.schema_id
        WHERE s.name = ?
        """,
        (sql_schema,),
    )
    sql_tables = {row[0] for row in sql_cursor.fetchall()}

    common_tables = sorted(pg_tables & sql_tables)
    ordered_tables = dependency_order(pg_cursor, common_tables, schema_name=pg_schema)

    return [(f"{pg_schema}.{name}", f"{sql_schema}.{name}") for name in ordered_tables]

def source_columns(pg_cursor, source_table):
    """Return ordered source columns from PostgreSQL information_schema."""
    schema_name, table_name = parse_pg_table_name(source_table)
    pg_cursor.execute(
        """
        SELECT column_name
        FROM information_schema.columns
        WHERE table_schema = %s AND table_name = %s
        ORDER BY ordinal_position
        """,
        (schema_name, table_name),
    )
    return [row[0] for row in pg_cursor.fetchall()]

def migrate_table(pg_cursor, sql_cursor, source_table, dest_table):
    # The destination defines the authoritative column order for positional bulkcopy().
    dest_columns, identity = table_columns(sql_cursor, dest_table)
    if not dest_columns:
        raise RuntimeError(
            f"No destination columns found for {dest_table}. "
            "Make sure the destination table exists before migration."
        )

    src_columns = source_columns(pg_cursor, source_table)
    if not src_columns:
        raise RuntimeError(
            f"No source columns found for {source_table}. "
            "Check the source table name and schema."
        )

    # Load only columns present on both sides and keep destination column order.
    src_column_set = set(src_columns)
    load_columns = [c for c in dest_columns if c in src_column_set]
    if not load_columns:
        raise RuntimeError(
            f"No shared columns between {source_table} and {dest_table}."
        )

    source_schema, source_name = parse_pg_table_name(source_table)
    select_query = sql.SQL("SELECT {cols} FROM {schema}.{table}").format(
        cols=sql.SQL(", ").join(sql.Identifier(c) for c in load_columns),
        schema=sql.Identifier(source_schema),
        table=sql.Identifier(source_name),
    )
    pg_cursor.execute(select_query)

    copied = 0
    while True:
        batch = pg_cursor.fetchmany(10000)
        if not batch:
            break
        # Serialize JSONB or array values (dict/list) for nvarchar(max) columns.
        rows = [
            tuple(json.dumps(v) if isinstance(v, (dict, list)) else v for v in row)
            for row in batch
        ]
        # keep_identity preserves source primary keys so foreign keys still line up.
        result = sql_cursor.bulkcopy(
            dest_table,
            rows,
            batch_size=10000,
            keep_identity=identity in load_columns,
        )
        copied += result["rows_copied"]
    return copied

pg_cursor = pg_conn.cursor()
sql_cursor = sql_conn.cursor()

# Leave TABLE_MAPPINGS as None to migrate every table that exists in both schemas.
# To migrate only selected tables, replace None with explicit mappings.
TABLE_MAPPINGS = None

if TABLE_MAPPINGS is None:
    tables = discover_table_pairs(pg_cursor, sql_cursor, pg_schema="public", sql_schema="dbo")
else:
    tables = TABLE_MAPPINGS

if not tables:
    raise RuntimeError(
        "No shared tables found between source and destination schemas. "
        "Check schema names and table creation on SQL Server."
    )

print(f"Migrating {len(tables)} table(s)...")
for source_table, dest_table in tables:
    count = migrate_table(pg_cursor, sql_cursor, source_table, dest_table)
    print(f"{dest_table}: copied {count} rows")

# bulkcopy() bypasses constraint checks, so foreign keys are left untrusted.
# Re-validate each table to mark them trusted and surface any orphaned rows.
for _, dest_table in tables:
    dest_schema, dest_name = parse_sql_table_name(dest_table)
    sql_cursor.execute(
        f"ALTER TABLE [{dest_schema}].[{dest_name}] WITH CHECK CHECK CONSTRAINT ALL"
    )
sql_conn.commit()

pg_conn.close()
sql_conn.close()

기본적으로 이 스크립트는 (PostgreSQL)와 public (SQL Server) 모두 dbo 에 존재하는 모든 테이블을 외래 키 의존성 순서대로 마이그레이션합니다. 일부만 마이그레이션하고 싶다면 명시적인 리스트로 설정 TABLE_MAPPINGS 하세요.

이는 출발지와 목적지가 동일한 열명을 사용한다는 가정 하에 이루어지며, 이는 DDL을 다시 작성한 후의 일반적인 경우입니다. 헬퍼는 IDENTITY 열을 자동으로 처리합니다. 즉, 대상에 IDENTITY 열이 있는 경우 keep_identity가 소스 기본 키를 보존하므로 외래 키 참조가 그대로 유지됩니다. 대신 SQL Server가 새 키를 할당하도록 하려면 columns에서 ID 열을 제외하고 keep_identity=False을 전달하세요.

외래 키와 제약 조건

bulkcopy()는 TDS 대량 삽입 프로토콜을 사용하며, 적재 중에는 외래 키 제약 조건이나 검사 제약 조건을 적용하지 않습니다. 명시적으로 확인하라는 요청이 없으면 SQL Server는 BULK INSERT에 설명된 대로 대량 가져오기 작업 중에 CHECK 제약 조건과 FOREIGN KEY 제약 조건을 무시하고, 이후 이를 신뢰할 수 없는 상태로 표시합니다. 이러한 행동은 이동에 두 가지 실질적인 결과를 가져옵니다:

로드 순서는 중요하지 않습니다. 외래 키 제약 조건 위반 없이 자식 테이블을 부모 테이블보다 먼저 로드할 수 있습니다. 보조 키처럼 기본 키를 keep_identity=True로 보존하여 부하 후에도 부모와 자식 키 값이 일치하도록 합니다.
제약은 결국 신뢰받지 못하게 됩니다. 대량 로드 후에는 각 외래 키가 신뢰할 수 없음(sys.foreign_keys.is_not_trusted = 1)으로 표시되는데, 이는 SQL Server가 검증하지 않았기 때문입니다. 스크립트의 마지막 단계는 로 로드된 모든 테이블 ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL을 다시 검증합니다. 이 단계는 쿼리 옵티마이저가 사용할 수 있도록 신뢰하는 제약 조건을 표시하며, 나쁜 데이터를 드러냅니다. 자식 행이 누락된 부모 행을 참조하면 제약 조건 이름이 포함된 무결성 제약 조건 위반으로 해당 구문이 실패하므로, 서비스를 시작하기 전에 고아 행을 수정할 수 있습니다.

Limitations

마이그레이션 전에 이 차이점들을 검토하세요:

주제	PostgreSQL	mssql-python / SQL Server
`callproc()`	지원됨	인상.`NotSupportedError` `cursor.execute("EXECUTE ...")`를 대신 사용하세요.
테이블 값 매개변수(TVP)	직접 동등한 항목 없음	현재 드라이버에서는 지원되지 않습니다. 다중 행 매개변수는 임시 테이블이나 JSON을 사용하세요.
토착 `ARRAY` 기둥	지원됨	배열 타입도 없고요. 정규화된 테이블, JSON 배열, 또는 `STRING_SPLIT()`.
`LISTEN`/`NOTIFY`	지원됨	직접 동등한 항목이 없습니다. 서비스 브로커나 애플리케이션 레벨 폴링을 사용하세요.
`COPY` 스트리밍	지원됨	대량 데이터 로드에 `bulkcopy()`를 사용하세요.
수정된 행을 반환하기	`RETURNING` 조항	`OUTPUT INSERTED` / `OUTPUT DELETED` DML 명세서의 조항.
비동기 드라이버	`psycopg3` 기본 비동기 지원	`mssql-python` 비동기 지원은 우회 방식(스레드 풀)에 의존합니다.
전체 텍스트 검색	`tsvector` / `tsquery`	`CONTAINS()` / `FREETEXT()` 전문 인덱스가 있는
ORM (SQLAlchemy)	완전히 지원됨	SQLAlchemy 2.1.0b2+(프리릴리즈)에 내장된 mssql-python 방언을 통해 지원됩니다.

유효성 검사 목록

이 체크리스트를 사용하여 마이그레이션을 확인하세요:

모든 %s 매개변수 표식을 ? 또는 %(name)s 매개변수로 바꾸세요.
모든 %(name)s 매개변수가 정상 작동하는지 확인하세요(두 드라이버 모두 이 형식을 지원합니다).
LIMIT / OFFSET를 OFFSET/FETCH NEXT로 다시 작성하세요.
RETURNING을(를) OUTPUT INSERTED(으)로 다시 작성하세요.
ON CONFLICT을(를) MERGE로 다시 쓰세요.
로 대체하세요SERIAL / BIGSERIAL.IDENTITY
BOOLEAN 기둥은 비트로 대체되었습니다.
배열 열을 정규화된 테이블이나 JSON 버전으로 대체하세요.
연산자를 JSONBJSON_VALUE() / 로 대체한다.JSON_QUERY()
Microsoft SQL 인증을 위한 연결 문자열 업데이트.
AdventureWorks나 타겟 스키마에 대해 애플리케이션을 테스트해 보세요.

인증 및 배포

자가 관리 PostgreSQL 애플리케이션은 일반적으로 비밀번호가 포함된 연결 문자열과 함께 배포되거나 파일과 .pgpass 환경 변수를 사용합니다PGPASSWORD. Azure Database for PostgreSQL은 Microsoft Entra 인증을 지원하므로, 이미 비밀번호 없는 인증을 사용 중이라면 동일한 신원 모델이 Azure SQL에도 적용됩니다.

Azure SQL에 대한 프로덕션 워크로드에는 관리 ID를 사용하세요:

conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="AdventureWorks",
    authentication="ActiveDirectoryMSI",
    encrypt="yes"
)

로컬 개발 및 CI에 대해서는 Docker, devcontainer, CI 파이프라인 설정 패턴에 관한 컨테이너 및 로컬 개발 을 참조하세요.

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-07-22

mssql-python을 이용해 PostgreSQL에서 Microsoft SQL로 마이그레이션하기

Microsoft SQL로 전환함으로써 얻는 이점

드라이버 설치

SQL 데이터베이스 만들기

연결 코드 업데이트

연결 문자열

Authentication

컨텍스트 관리자

연결 풀링 (Connection Pooling)

SQL 방언 차이점

CREATE TABLE 예제

쿼리 재작성 패턴

페이지 나누기

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

삽입된 ID 가져오기

매개 변수 표식

트랜잭션과 자동커밋의 차이점

유형 고려사항

JSON

UUID

날짜, 시간, 시간대

배열

Unicode

대량 로딩 및 데이터 이동

스키마 및 데이터 마이그레이션

외래 키와 제약 조건

Limitations

유효성 검사 목록

인증 및 배포

관련 콘텐츠

피드백

추가 리소스