fullmetalcopy

Fast PostgreSQL bulk loads from CSV using the server-side copy path, wired through SQLAlchemy 2 so you keep your normal engines and connection patterns.

PyPI: fullmetalcopy · Source: GitHub

Why use this?

Bulk-inserting rows with plain INSERT is slow for large datasets. This library streams CSV bytes into a table via PostgreSQL’s efficient copy-style ingestion, with thin wrappers over psycopg3, psycopg2, or asyncpg depending on your SQLAlchemy URL.

Requirements

Install at least one driver extra (or bring your own compatible stack):

Install

pip install fullmetalcopy

# Drivers (pick what matches your engine URL)
pip install "fullmetalcopy[psycopg]"
pip install "fullmetalcopy[psycopg2]"
pip install "fullmetalcopy[asyncpg]"

# Examples with dataframe extras
pip install "fullmetalcopy[psycopg,pandas]"
pip install "fullmetalcopy[asyncpg,polars]"

Public API

Import from the package root:

import fullmetalcopy as fc

fc.__version__          # package version string
fc.copy_from_csv       # synchronous
fc.async_copy_from_csv  # asynchronous (alias of the async module’s copy_from_csv)

Both callables accept a binary CSV stream (io.BytesIO, open(..., "rb"), or any typing.BinaryIO).

Driver is inferred from connection.dialect.driver. Non-PostgreSQL dialects raise a clear error.

Usage examples

Synchronous (psycopg or psycopg2)

import io

import sqlalchemy as sa

import fullmetalcopy as fc

engine = sa.create_engine("postgresql+psycopg2://scott:tiger@hostname/dbname")

with io.BytesIO() as buf:
    buf.write(b"id,name\n1,Ada\n")
    buf.seek(0)
    with engine.connect() as conn:
        fc.copy_from_csv(conn, buf, "people")
        conn.commit()

Asynchronous (asyncpg or psycopg)

import io

from sqlalchemy.ext.asyncio import create_async_engine

import fullmetalcopy as fc

async def main() -> None:
    engine = create_async_engine("postgresql+asyncpg://scott:tiger@hostname/dbname")
    try:
        async with engine.connect() as conn:
            with io.BytesIO() as buf:
                buf.write(b"id,name\n1,Ada\n")
                buf.seek(0)
                await fc.async_copy_from_csv(conn, buf, "people")
                await conn.commit()
    finally:
        await engine.dispose()

Non-default schema

fc.copy_from_csv(conn, buf, "invoices", schema="billing")
await fc.async_copy_from_csv(conn, buf, "invoices", schema="billing")

Header row plus explicit column order

When the file has a header but you want to pin the column list (and skip the header line the same way on every backend):

fc.copy_from_csv(
    conn,
    buf,
    "people",
    columns=["id", "name"],
    headers=True,
)

Driver behavior (summary)

For full detail, see the docstrings in fullmetalcopy.synchronous.copycsv and fullmetalcopy.asynchronous.copycsv, and the per-driver modules under fullmetalcopy.synchronous / fullmetalcopy.asynchronous.

CSV and null semantics

• The header line (when headers=True) is decoded as UTF-8 and parsed with csv.reader, so quoted fields in the header are handled like normal CSV.
• On the psycopg3 path, each data row is also read with csv.reader and must have exactly as many fields as the resolved column list, or a ValueError is raised.
• On the psycopg3 path, any cell that equals the null string is stored as SQL NULL. The default null="" maps empty fields to NULL. If you need to keep empty strings as empty strings, set null to a sentinel that does not appear in real data (for example "\N" where appropriate).

PostgreSQL’s CSV rules and Python’s csv module are very close but not identical for exotic cases (embedded newlines, unusual escapes). For maximum fidelity on huge files, consider a raw server-side COPY … FROM STDIN pipeline; this library targets typical spreadsheet-style CSV.

Development

From a clone of the repository:

pip install -e ".[dev]"
ruff format .
ruff check .
mypy fullmetalcopy
pytest

[dev] pulls all supported drivers, pytest, Ruff, Mypy, and testing.postgresql for the test suite.

Releasing

1. Align the version in pyproject.toml and fullmetalcopy/__init__.py.
2. Update CHANGELOG.md.
3. Run ruff format ., ruff check ., mypy fullmetalcopy, and pytest.
4. Build artifacts: python -m build (optionally twine check dist/*).
5. Tag: git tag -s v0.2.0 -m "Release 0.2.0" then git push origin v0.2.0.
6. Upload to PyPI (trusted publishing or twine upload dist/*).

License

MIT — see LICENSE. Release history: CHANGELOG.md.