ورود امن CSV با پایتون؛ خطای هر ردیف را قبل از ذخیره پیدا کنید

import csv
import io

text = "id,amount\nA1,1200\nA1,500\nA2,-1\n"
reader = csv.DictReader(io.StringIO(text, newline=""))
if reader.fieldnames != ["id", "amount"]:
    raise ValueError("unexpected headers")

accepted, errors, seen = [], [], set()
for number, row in enumerate(reader, start=2):
    try:
        if None in row or any(value is None for value in row.values()):
            raise ValueError("incorrect field count")
        key = row["id"].strip()
        if not key:
            raise ValueError("empty id")
        amount = int(row["amount"])
        if amount < 0:
            raise ValueError("negative amount")
        if key in seen:
            raise ValueError("duplicate id")
        seen.add(key)
        accepted.append((key, amount))
    except ValueError as error:
        errors.append(f"record {number}: {error}")

print(errors)
print("Ready to commit:", not errors)

خروجی باید خطای شناسه تکراری رکورد 3، مبلغ منفی رکورد 4 و Ready to commit: False باشد. نمونه عمداً در دیتابیس نمی‌نویسد. شناسه A1 در این قرارداد به حروف بزرگ و کوچک حساس است.

این تابع را در csv_import.py ذخیره کنید. فایل یا جریان متن از بیرون دریافت می‌شود تا هم از دیسک بخوانیم و هم با StringIO آزمایش کنیم. نتیجه ساختاریافته است و هیچ ردیفی را خودکار در دیتابیس ثبت نمی‌کند.

# csv_import.py
import csv

def validate_orders(stream):
    reader = csv.DictReader(stream)
    if reader.fieldnames != ["id", "amount"]:
        raise ValueError("expected headers: id,amount")
    accepted, errors, seen = [], [], set()
    for number, row in enumerate(reader, start=2):
        try:
            if None in row or any(v is None for v in row.values()):
                raise ValueError("incorrect field count")
            key = row["id"].strip()
            if not key:
                raise ValueError("empty id")
            amount = int(row["amount"])
            if amount < 0:
                raise ValueError("negative amount")
            if key in seen:
                raise ValueError("duplicate id")
            seen.add(key)
            accepted.append((key, amount))
        except ValueError as error:
            errors.append({"record": number, "message": str(error)})
    return accepted, errors

خروجی accepted فقط برای ادامه مرحله پردازش است؛ اگر سیاست «همه یا هیچ» دارید، وجود errors باید مانع commit شود. برای فایل بزرگ، تعداد خطاها، اندازه فایل و حافظه را محدود کنید؛ این نسخه برای فایل کنترل‌شده و کوچک طراحی شده است.

فایل دوم را کنار ماژول اعتبارسنجی بگذارید. قبل از ساختن اتصال، کل فایل اعتبارسنجی می‌شود. سپس همه INSERTها در یک تراکنش قرار می‌گیرند و کلید اصلی دیتابیس، تکرار میان فایل‌های مختلف را هم می‌گیرد.

# import_orders.py; run only against a disposable demo database.
import csv
import sqlite3
import sys
from csv_import import validate_orders

csv.field_size_limit(1_000_000)
with open(sys.argv[1], encoding="utf-8-sig", newline="") as source:
    rows, errors = validate_orders(source)
if errors:
    for error in errors:
        print(f"record {error['record']}: {error['message']}")
    raise SystemExit(2)

db = sqlite3.connect(sys.argv[2])
try:
    db.execute("CREATE TABLE IF NOT EXISTS orders "
               "(id TEXT PRIMARY KEY, amount INTEGER NOT NULL CHECK(amount>=0))")
    with db:
        db.executemany("INSERT INTO orders(id,amount) VALUES(?,?)", rows)
    print("Imported:", len(rows))
except sqlite3.IntegrityError as error:
    print("Nothing from this transaction was committed:", error)
    raise SystemExit(3)
finally:
    db.close()

با python import_orders.py orders.csv demo-orders.sqlite اجرا کنید. اجرای دوباره همان فایل باید با exit 3 شکست بخورد و هیچ بخش جدیدی از همان تراکنش ثبت نشود. این قرارداد، ورود تکراری را رد می‌کند؛ اگر محصول باید retry را موفق بشمارد، شناسه batch و سیاست idempotency را جدا طراحی کنید.

این تست‌ها را با python -m pytest -q test_csv_import.py اجرا کنید. آزمون آخر عمداً ردیف دوم را به کلید تکراری می‌رساند؛ بنابراین هم وجود ردیف قبلی و هم نبود ردیف تازه پس از rollback بررسی می‌شود.

# test_csv_import.py
import io
import sqlite3
from csv_import import validate_orders

def test_clean_file():
    rows, errors = validate_orders(io.StringIO("id,amount\nA1,1200\nA2,0\n"))
    assert rows == [("A1", 1200), ("A2", 0)] and not errors

def test_reports_both_errors():
    rows, errors = validate_orders(io.StringIO("id,amount\nA1,1\nA1,2\nA2,-1\n"))
    assert [e["record"] for e in errors] == [3, 4]

def test_rejects_extra_column():
    _, errors = validate_orders(io.StringIO("id,amount\nA1,1,extra\n"))
    assert errors[0]["message"] == "incorrect field count"

def test_database_rolls_back_batch():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE orders(id TEXT PRIMARY KEY,amount INTEGER)")
    with db:
        db.execute("INSERT INTO orders VALUES('existing',1)")
    try:
        with db:
            db.executemany("INSERT INTO orders VALUES(?,?)",
                           [("new", 2), ("existing", 3)])
    except sqlite3.IntegrityError:
        pass
    assert db.execute("SELECT id FROM orders").fetchall() == [("existing",)]
    db.close()

انتظار 4 passed داریم. این بررسی، تفاوت خطای اعتبارسنجی فایل و خطای محدودیت دیتابیس را قابل مشاهده می‌کند. برای انتشار گزارش ردشدن فایل، شماره رکورد و پیام کافی است؛ کپی کل ردیف خصوصی در لاگ ضرورتی ندارد.