PostgreSQL'den Microsoft SQL'e mssql-python ile migrasyon

Birçok Python ekibi önce PostgreSQL'i öğrenir. İş yükünüz zamansal tablolar, tam MERGE semantiği veya sütun deposu dizinleri gibi özelliklere ihtiyaç duyduğunda Microsoft SQL'e geçin. Bu rehber, bir Python uygulamasını PostgreSQL'den (psycopg2 veya psycopg3 kullanılarak) Microsoft SQL'e sürücü mssql-python kullanılarak taşımak için verilen ana kararları ve kod değişikliklerini kapsar.

Note

PostgreSQL için Azure Veri Tabanı'den geçiş yapıyorsanız, her iki hizmet de Microsoft Entra kimlik doğrulama ve yönetilen kimliği destekler. Bu kılavuzdaki kod değişiklikleri, PostgreSQL kaynağınızın kendi kendine yönetilen bir kaynak mı yoksa Azure tarafından barındırılan bir kaynak mı olduğuna bakılmaksızın geçerlidir.

Microsoft SQL'e geçerek ne kazanırsınız?

Microsoft SQL, üretim iş yükleri için güvenlik, uyum ve işlemleri basitleştiren yetenekler içerir. Geçiş sırasında faydalanabilmeniz için göç yapmaya başlamadan önce bu özellikleri anlayın:

Dinamik veri maskelemesi ve satır düzeyinde güvenlik. Tam erişime ihtiyacı olmayan kullanıcılar için sütunları maske edin ve güvenlik politikası ile satır görünürlüğünü kısıtlayın. Bu özellikler herhangi bir sürücü için çalışır.
Zamansal tablolar (sistem versiyonlu). Microsoft SQL satır geçmişini otomatik olarak takip eder. Tetikleyici yok, denetim tablosu yok, uygulama kodu yok.
Tam MERGE anlambilim. Tek bir ifade, denetim izleri için bir OUTPUT yan tümcesiyle INSERT, UPDATE ve DELETE öğelerini işler. PostgreSQL'in maddesi ON CONFLICT yalnızca tek bir kısıtlama üzerinde ekle veya güncelle uygulamasını kapsar.
Columnstore indeksleri. Hibrit OLTP/analitik iş yükleri için mevcut tablolara sütunlu depolama ekleyin. Ayrı bir analiz veritabanına gerek yok.
Microsoft Entra Id kimlik doğrulaması. Yönetilen kimlikler, hizmet prensipleri veya etkileşimli giriş ile bağlantı kurun. PostgreSQL için Azure Veri Tabanı de Microsoft Entra doğrulamasını destekliyor, yani zaten kullanıyorsanız geçiş oldukça basit.

Sürücüyü yükleme

Başlamadan önce, Python 3.10 veya daha yeni ve hedef SQL veritabanınız olduğundan emin olun.

SQL veritabanı oluşturma

Aşağıdaki platformlardan birinde bir SQL veritabanı oluşturun veya bağlanın:

Azure portalını kullanarak dakikalar içinde SQL veritabanı oluşturma

PostgreSQL sürücüleri harici yerel kütüphaneler gerektirir.

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

mssql-python sürücüsü kendi yerel katmanını içerir. Windows'ta harici sürücü yöneticisine veya sistem paketlerine ihtiyacınız yok.

pip install mssql-python

Linux ve macOS'ta, Kurulum'da belgelenmiş küçük bir sistem kütüphanesi seti kurun. pg_config veya libpq-dev için bir karşılık yok.

Bağlantı kodunu güncelle

Aşağıdaki bölümler, bağlantı dizeleri, kimlik doğrulama, bağlam yöneticileri ve havuzlama konusundaki temel değişiklikleri kaplar.

Bağlantı stringleri

psycopg2, DSN dizisi veya anahtar kelime argümanları kullanır.

import psycopg2

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

mssql-python ayrıca anahtar kelime argümanlarını destekler; bu, şifreler , @, veya ; karakterler içerdiğinde {}SQLAlchemy bağlantı dizilerinin sıklıkla yaşadığı URL kodlama sorunlarından kaçınır.

import mssql_python

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

Ya da bir bağlantı dizesi kullanın.

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

Bağlantı dizesi anahtar sözcüklerinin tam listesi için Bağlantı Dizeleri bölümüne bakın.

Authentication

PostgreSQL kimlik doğrulaması genellikle kullanıcı adı ve şifreyle kurallar kullanır pg_hba.conf . PostgreSQL için Azure Veri Tabanı ayrıca Microsoft Entra doğrulamasını destekler. Microsoft SQL, tek bir bağlantı anahtar kelimesi aracılığıyla birden fazla kimlik doğrulama modunu destekler:

PostgreSQL yaklaşımı	mssql-python karşılığı
Kullanıcı adı ve parola	`UID=...;PWD=...;`
SSL/TLS şifreleme	`Encrypt=yes;`(Azure SQL için varsayılan olarak etkinleştirilmiştir)
Entra auth (Azure PostgreSQL)	`Authentication=ActiveDirectoryDefault;` (şifresiz)
Managed identity (Azure PostgreSQL)	`Authentication=ActiveDirectoryMSI;`
Service principal (Azure PostgreSQL)	`Authentication=ActiveDirectoryServicePrincipal;`

Yerel geliştirme için kullanın ActiveDirectoryDefault . Azure CLI, ortam değişkenleri ve yönetilen kimlik üzerinden otomatik olarak zincirler. Üretim için, yavaş kimlik bilgisi zinciri taramasını önlemek amacıyla ActiveDirectoryMSI (yönetilen kimlik) veya ActiveDirectoryServicePrincipal gibi belirli bir mod kullanın. Tüm yedi kimlik doğrulama modu için Microsoft Entra kimlik doğrulamasına bakınız.

Bağlam yöneticileri

Her iki sürücü de bağlam yöneticilerini destekler, ancak davranış farklıdır:

psycopg2'nin with conn: başarılı olduğunda commit eder ve bir özel durumda rollback yapar, ancak bağlantıyı kapatmaz:

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: çıkışında bağlantıyı kapatıyor. Taahhüt edilmemiş işler geri alınır:

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

Bağlantı havuzlama

Psycopg2, bağlantı havuzunun açıkça kurulması ve yönetilmesini gerektirir.

from psycopg2 import pool

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

Sürücü mssql-python varsayılan olarak yerleşik havuzlama etkinleştirilmiş durumda. Kuruluma gerek yok.

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

Varsayılan ayarlar iş yükünüze uymuyorsa havuz boyutunu ayarlayın.

import mssql_python

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

Havuz boyutlandırması ve havuz tükenmesinin sorun giderilmesi hakkında rehberlik için Bağlantı havuzu bölümünü inceleyebilirsiniz.

SQL lehçesi farklılıkları

Aşağıdaki tablo, yaygın PostgreSQL kalıplarını Transact-SQL (T-SQL) eşdeğerleriyle eşler:

PostgreSQL	SQL Server (T-SQL)	Notlar
`SERIAL` / `BIGSERIAL`	`int IDENTITY(1,1)`	Microsoft SQL, otomatik artırım için `IDENTITY` kullanır.
`TEXT`	`nvarchar(max)`	Unicode için kullanın `nvarchar` . Veriler elverdiğinde `nvarchar(4000)` veya daha kısa olanı tercih edin.
`BOOLEAN`	`bit`	PostgreSQL `true`/`false` kabul eder; Microsoft SQL `1`/`0` kullanır.
`BYTEA`	`varbinary(max)`	Aynı kavram, farklı isim.
`JSONB`	`nvarchar(max)` JSON fonksiyonlarıyla	Microsoft SQL, JSON verilerini metin olarak depolar ve `ISJSON()` ile doğrular. JSON verilerine bakınız.
`TIMESTAMP WITH TIME ZONE`	`datetimeoffset`	Her ikisi de ofset depolar. Bkz. Tarih ve saat işleme.
`INTERVAL`	Doğrudan eşdeğeri yok	`DATEADD()` ve `DATEDIFF()` ile hesaplayın.
`ARRAY`	Doğrudan eşdeğeri yok	Ayrı bir tablo, bir JSON dizisi veya `STRING_SPLIT()` kullanın.
`UUID`	`uniqueidentifier`	`mssql-python` sürücüsü, `uuid.UUID` öğesini yerel olarak eşler. Modül yapılandırmasına bakınız.
`NOW()` / `CURRENT_TIMESTAMP`	`GETDATE()` veya `SYSDATETIME()`	`SYSDATETIME()` daha yüksek hassasiyet sağlar.
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	Bir `ORDER BY` maddesi gerekli.
`\\|\\|` (dize birleştirme)	`+` veya `CONCAT()`	`CONCAT()` değerleri `NULL` yönetir.
`COALESCE(a, b)`	`COALESCE(a, b)` veya `ISNULL(a, b)`	`COALESCE` her ikisinde de aynıdır.
`string_agg(col, ',')`	`STRING_AGG(col, ',')`	SQL Server 2017+ içinde mevcuttur.
`RETURNING id`	`OUTPUT INSERTED.id`	`INSERT`, `UPDATE` veya `DELETE` deyiminde `OUTPUT` kullanın.
`ON CONFLICT ... DO UPDATE`	`MERGE` açıklama	`MERGE`, tek bir ifadede INSERT + UPDATE + DELETE destekler. Sorgu yeniden yazma kalıpları bölümünü inceleyebilirsiniz.
`EXPLAIN ANALYZE`	`SET STATISTICS IO ON; SET STATISTICS TIME ON;`	Ya da SSMS / Azure Data Studio'da yürütme planlarını kullanabilirsiniz.
`\d tablename`	`sp_help 'tablename'`	Ya da sorgu `INFORMATION_SCHEMA.COLUMNS`.
`pg_dump`	`bcp`, `BACKUP DATABASE`	Python'dan programatik veri yüklemek için `bulkcopy()` kullanın.

CREATE TABLE Örnek

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

Sorgu yeniden yazma kalıpları

Aşağıdaki bölümler, yaygın PostgreSQL sorgu kalıplarını ve bunların T-SQL karşıtlıklarını gösterir.

Pagination

PostgreSQL:

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

MSSQL-Python:

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

Parametre sırası tersine çevrilmiştir. Microsoft SQL, OFFSET öğesini FETCH NEXT öğesinin önüne koyar.

Upsert (ekle veya güncelle)

PostgreSQL, ON CONFLICT tek bir kısıtlama üzerinde ekle veya güncelleme işliyor:

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

Microsoft SQL'in MERGE, INSERT, UPDATE ve DELETE öğelerini tek bir deyimde yönettiği yapı. Parametre takma adları olan bir USING madde kullanın:

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

Toplu upsert işlemleri için önce satırları bulkcopy() kullanarak geçici bir tabloda hazırlayın, ardından bu tablodan MERGE gerçekleştirin. Daha fazla bilgi için, Toplu Yükseltme ve Sahneleme Tablosu sayfasına bakınız.

Eklenen kimliği alın

PostgreSQL:

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

MSSQL-Python:

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

OUTPUT INSERTED, INSERT, ve UPDATE ifadelerle DELETEçalışır. Birden fazla sütun döndürebilir.

Parametre işaretçileri

Psycopg2, konumsal parametreler için %s, adlandırılmış parametreler için ise %(name)s kullanır. mssql-python sürücüsü, konumsal için ? ve adlandırılmış için %(name)s kullanır:

PsycopG2:

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

MSSQL-Python:

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

İşlem ve otomatik taahhüt farkları

PostgreSQL (psycopg2), ilk komutta otomatik olarak bir işlem başlatır ve açık bir commit() gerektirir:

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

Sürücü mssql-python varsayılan olarak aynı şekilde çalışıyor. Otomatik commit kapalıdır ve commit() öğesini açıkça çağırırsınız:

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

Otomatik commit etkinleştirmek için:

PsycopG2:

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

MSSQL-Python:

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

İzolasyon seviyeleri, geri alma noktaları ve kilitlenme yeniden deneme kalıpları için İşlem yönetimi bölümüne bakın.

Türle İlgili Hususlar

Aşağıdaki bölümler, PostgreSQL ile Microsoft SQL arasındaki en yaygın tür eşleme farklarını ele alır.

JSON

PostgreSQL, indeksleme ve sorgu operatörleriyle (->, ->>, @>) yerleşik JSONB desteğine sahiptir. Microsoft SQL JSON'u nvarchar(max) olarak saklar ve sorgulama için fonksiyonlar sağlar:

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'da her iki yaklaşım da serileştirmek için json.dumps() kullanır:

import json

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

JSON depolama ve sorgulama kalıpları hakkında tam rehberlik için JSON verilerine bakınız.

UUID

Hem PostgreSQL hem de mssql-python doğal olarak eşlendirir uuid.UUID :

import uuid

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

Bağlantı seçeneği için Modül yapılandırmasınative_uuid bölümüne bakın.

Tarih saati ve saat dilimi

PostgreSQL'in TIMESTAMPTZ değeri depolanırken UTC'ye dönüştürülür. Microsoft SQL datetimeoffset orijinal ofseti korur:

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

Tutarlı UTC depolama istiyorsanız, eklemeden önce Python'da dönüştürün:

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

Tam tür eşlemesi için Tarih-saat işleme bölümüne bakın.

Diziler

PostgreSQL, yerel dizi sütunlarını (INTEGER[], ). TEXT[]destekler. Microsoft SQL'in bir dizi tipi yok. Yaygın alternatifler:

Ayrı bir tablo (normalleştirilmiş). Sorgulanabilir, indekslenmiş veriler için en iyisi.
JSON dizisinvarchar(max) içinde saklanır. Opak meta veriler için iyi.
Virgülle ayrılmış dizeSTRING_SPLIT(). Basit ama sınırlı.

# 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 tüm metni varsayılan olarak UTF-8 olarak saklar. Microsoft SQL, varchar (kod sayfası kodlama) ile nvarchar (UTF-16) arasında ayrım yapar. Sürücü, mssql-python varsayılan olarak Python str değerlerini nvarchar olarak gönderiyor, yani Unicode metni ekstra yapılandırma olmadan çalışıyor. Şemanız varchar sütunları kullanıyorsa ve örtük dönüşümden kaçınmanız gerekiyorsa, sütun türünü belirtmek için kullanın setinputsizes() . Kodlama detayları için String ve Unicode verilerine bakınız.

Toplu yükleme ve veri taşımacılığı

PostgreSQL toplu işlemler için kullanır COPY . MSSQL-python şunları sağlar bulkcopy():

PsycopG2:

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

MSSQL-Python:

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)

Büyük dosyalar için, tüm dosyanın belleğe yüklenmemesi için bir üreteci kullanın:

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)

Sütun eşlemeleri, kimlik işleme ve performans ipuçları için Toplu kopyalama işlemlerine bakınız.

Şema ve veri taşıması

Mevcut bir PostgreSQL veritabanını taşımak için bu yaklaşımı kullanın:

Şemayı dışa aktar. DDL almak için pg_dump --schema-only kullanın. Seçenek detayları ve uç durumlar (sahiplik, ayrıcalıklar, uzantılar ve filtreleme) için PostgreSQL pg_dump referansına bakınız. DDL'yi SQL diyalekt farkları tablosuyla yeniden yazın.
Microsoft SQL'de tablolar oluşturun. Yeniden yazılmış DDL'yi hedef veritabanınıza uygulayın.
Verileri dışarı aktarın. pg_dump --data-only --format=csv kullanın veya her tabloyu psycopg2 ile sorgulayın. Büyük veri kümeleri ve uyumluluk anahtarları için PostgreSQL pg_dumpbelgelerine, özellikle seçenekler bölümüne göz atın.
Verileri bulkcopy kullanarak yükleyin. Katalogdan hedef sütun sırasını okuyun ki her tablo için bir sütun listesini sabit kodlamayacaksınız, sonra her tabloyu Microsoft SQL'e aktarın. İşte bir örnek komut dosyası:

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

Varsayılan olarak, bu script hem public (PostgreSQL) hem dbo de (SQL Server)'da var olan her tabloyu yabancı anahtar bağımlılıklarına göre sıralanmış olarak taşıyor. TABLE_MAPPINGS Sadece bir alt kümeyi taşımak istiyorsanız açık bir liste ayarlayın.

Bu, kaynak ve hedef adlarının aynı sütun adlarını kullandığını varsayar; DDL yeniden yazıldığında bu genellikle geçerlidir. Yardımcı, kimlik sütununu otomatik olarak yönetir: keep_identity hedef IDENTITY sütunu olduğunda kaynak birincil anahtarları korur, böylece yabancı anahtar referansları sağlam kalır. SQL Server’ın bunun yerine yeni anahtarlar atamasına izin vermek için, kimlik sütununu columns içinden hariç tutun ve keep_identity=False parametresini geçirin.

Yabancı anahtarlar ve kısıtlamalar

bulkcopy() TDS toplu insert protokolünü kullanır; bu protokol yük sırasında yabancı anahtar veya kontrol kısıtlamalarını zorunlu kılmaz. SQL Server, bunların denetlenmesine yönelik açık bir istek olmadan, toplu içe aktarma sırasında CHECK kısıtlamalarını FOREIGN KEY yok sayar ve BULK INSERT içinde açıklandığı gibi daha sonra bunları güvenilmez olarak işaretler. Bu davranışın göç için iki pratik sonucu vardır:

Yükleme sırası önemli değil. Bir alt tabloyu ana sayfasından önce yükleyebilirsiniz, yabancı anahtar ihlallerine ulaşmadan. Birincil anahtarları yardımcının yaptığı gibi keep_identity=True, ile koruyun, böylece ebeveyn ve çocuk anahtar değerleri yüklemeden sonra da eşleşmeye devam eder.
Kısıtlamalar güvenilmez hale gelir. Toplu yüklemeden sonra, her yabancı anahtar güvenilir değil (sys.foreign_keys.is_not_trusted = 1) olarak işaretlenir çünkü SQL Server bunu doğrulamamıştır. Script'teki son adım, yüklü her tabloyu yeniden doğrular.ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL Bu adım, sorgu optimizatorunun kullanabilmesi için güvenilen kısıtlamaları işaretler ve kötü verileri ortaya çıkarır. Bir alt satır eksik bir ebeveyne referans verirse, ifade bütünlük kısıtlaması ihlali ile başarısız olur ve kısıtlama kısıtlaması ile kısıtlama kullanılır, böylece yetim satırları canlı olmadan önce düzeltebilirsiniz.

Limitations

Taşınmadan önce bu farkları gözden geçirin:

Konu	PostgreSQL	mssql-python / SQL Server
`callproc()`	Destekleniyor	Oluşturur `NotSupportedError`. Bunun yerine `cursor.execute("EXECUTE ...")` kullanın.
Tablo değerine sahip parametreler (TVP'ler)	Doğrudan eşdeğeri yok	Mevcut sürücüde desteklenmiyor. Çok satırlı parametreler için geçici tablolar veya JSON kullanın.
Yerli `ARRAY` sütunlar	Destekleniyor	Dizi tipi yok. Normalize tablolar, JSON dizileri veya `STRING_SPLIT()`.
`LISTEN`/`NOTIFY`	Destekleniyor	Doğrudan eşdeğeri yok. Service Broker veya uygulama düzeyinde anket kullanın.
`COPY` Yayınlama	Destekleniyor	Toplu veri yükleme için `bulkcopy()` kullanın.
Değiştirilen satırları döndürme	`RETURNING` maddesi	`OUTPUT INSERTED` / `OUTPUT DELETED` DML ifadelerinde yan tümcesi.
Asenkron sürücü	`psycopg3` doğal asenkron	`mssql-python` Asenkron destek çözüm odaklıdır (iş parçacığı havuzu).
Tam metin arama	`tsvector` / `tsquery`	`CONTAINS()` / `FREETEXT()` tam metin dizinleriyle.
ORM (SQLAlchemy)	Tamamen desteklenir	SQLAlchemy 2.1.0b2+ (pre-release) içindeki yerleşik mssql-python lehçesi aracılığıyla desteklenmektedir.

Doğrulama denetim listesi

Göçünüzü doğrulamak için bu kontrol listesini kullanın:

Tüm %s parametre işaretleyicilerini ? veya %(name)s parametreleriyle değiştirin.
Tüm %(name)s parametrelerin hâlâ çalıştığından emin olun (her iki sürücü de bu formatı destekliyor).
Yeniden yaz LIMIT/OFFSET olarak .OFFSET/FETCH NEXT
Yeniden yaz RETURNING olarak OUTPUT INSERTED.
Yeniden yaz ON CONFLICT olarak MERGE.
SERIAL / BIGSERIAL yerine IDENTITY ile değiştirin.
BOOLEAN sütunlar bit ile değiştirildi.
Dizi sütunlarını normalize tablolar veya JSON ile değiştirin.
JSONB operatörlerini JSON_VALUE() / JSON_QUERY() ile değiştirin.
Microsoft SQL kimlik doğrulaması için bağlantı dizesi'i güncelle.
Uygulamayı AdventureWorks’e veya hedef şemanıza karşı test edin.

Kimlik doğrulama ve dağıtım

Kendi kendine yönetilen PostgreSQL uygulamaları genellikle parola içeren bağlantı dizeleriyle dağıtılır ya da .pgpass dosyalarını ve PGPASSWORD ortam değişkenlerini kullanır. PostgreSQL için Azure Veri Tabanı Microsoft Entra kimlik doğrulamasını destekliyor, yani zaten şifresiz doğrulama kullanıyorsanız, aynı kimlik modeli Azure SQL'e de geçer.

Azure SQL karşısında üretim iş yükleri için managed identity kullanın:

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

Yerel geliştirme ve CI için, Docker, devcontainer ve CI pipeline kurulum desenleri için Container and local development sayfasına bakınız.

Geri Bildirim

Bu sayfayı yararlı buldunuz mu?

Last updated on 2026-07-22

PostgreSQL'den Microsoft SQL'e mssql-python ile migrasyon

Microsoft SQL'e geçerek ne kazanırsınız?

Sürücüyü yükleme

SQL veritabanı oluşturma

Bağlantı kodunu güncelle

Bağlantı stringleri

Authentication

Bağlam yöneticileri

Bağlantı havuzlama

SQL lehçesi farklılıkları

CREATE TABLE Örnek

Sorgu yeniden yazma kalıpları

Pagination

Upsert (ekle veya güncelle)

Eklenen kimliği alın

Parametre işaretçileri

İşlem ve otomatik taahhüt farkları

Türle İlgili Hususlar

JSON

UUID

Tarih saati ve saat dilimi

Diziler

Unicode

Toplu yükleme ve veri taşımacılığı

Şema ve veri taşıması

Yabancı anahtarlar ve kısıtlamalar

Limitations

Doğrulama denetim listesi

Kimlik doğrulama ve dağıtım

İlgili içerik

Geri Bildirim

Ek kaynaklar