import hashlib
from datetime import datetime, timezone
from sqlalchemy import BigInteger, Column, DateTime, String, UniqueConstraint
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class Event(Base):
__tablename__ = "events"
__table_args__ = (UniqueConstraint("dedupe_key", name="uq_events_dedupe"),)
id = Column(BigInteger, primary_key=True)
source = Column(String(64), nullable=False)
name = Column(String(128), nullable=False)
payload = Column(String, nullable=False, default="")
dedupe_key = Column(String(64), nullable=False)
created_at = Column(DateTime(timezone=True), nullable=False)
@classmethod
def to_row(cls, raw):
source = raw["source"]
name = raw["name"]
payload = raw.get("payload", "")
digest = hashlib.sha256(f"{source}:{name}:{payload}".encode()).hexdigest()
return {
"source": source,
"name": name,
"payload": payload,
"dedupe_key": digest[:64],
"created_at": raw.get("created_at") or datetime.now(timezone.utc),
}
from itertools import islice
def chunked(iterable, size):
if size < 1:
raise ValueError("size must be >= 1")
iterator = iter(iterable)
while True:
chunk = list(islice(iterator, size))
if not chunk:
return
yield chunk
from sqlalchemy.dialects.postgresql import insert
from batching import chunked
from models import Event
def insert_events(session, raw_events, batch_size=2000):
inserted = 0
rows = (Event.to_row(raw) for raw in raw_events)
for chunk in chunked(rows, batch_size):
stmt = (
insert(Event)
.values(chunk)
.on_conflict_do_nothing(index_elements=["dedupe_key"])
)
result = session.execute(stmt)
session.commit()
inserted += result.rowcount or 0
return inserted
def load_from_stream(session_factory, raw_events, batch_size=2000):
session = session_factory()
try:
return insert_events(session, raw_events, batch_size=batch_size)
except Exception:
session.rollback()
raise
finally:
session.close()
Inserting tens of thousands of rows one session.add at a time is slow because each row generates its own INSERT round-trip and the ORM tracks every object in the identity map. This snippet shows how to push large batches through SQLAlchemy Core while keeping the ergonomics of a model, and how to chunk arbitrary iterables so memory stays flat.
The models.py tab defines a plain Event mapped class plus a to_row classmethod that normalizes a dict into exactly the columns the table expects, filling defaults like created_at and a computed dedupe_key. Keeping this mapping in one place means the loader never hand-builds partial dicts that drift from the schema, which is the usual source of NULL constraint surprises during bulk loads.
The chunked helper in batching.py is deliberately generic: it consumes any iterable lazily via iter and islice, yielding lists of at most size. Because it never materializes the whole input, it works on a streaming cursor or a multi-gigabyte file just as well as a list. The while True loop stops when islice returns an empty chunk, which is the clean way to detect exhaustion without peeking ahead.
The bulk_loader.py tab ties them together. insert_events maps raw dicts through Event.to_row, then feeds chunked so each database round-trip carries batch_size rows. It uses insert(Event).values(rows) — a Core multi-values INSERT — rather than the ORM unit-of-work, avoiding per-object flush overhead. on_conflict_do_nothing from the Postgres dialect makes the load idempotent against the dedupe_key unique index, so re-running a partially applied batch is safe. Each chunk commits inside its own transaction so a failure late in a large load doesn't roll back everything already written.
The trade-off is that bypassing the ORM means no automatic relationship cascades or Python-side default execution, so to_row must supply everything the row needs. For pure inserts of independent rows this is exactly the right tool; when relationship graphs or per-row events matter, the regular session flow is still preferable. Tuning batch_size balances round-trip count against statement size and lock duration — a few thousand rows per statement is a common sweet spot.
Related snips
import os
import stat
for root, _dirs, files in os.walk('/etc'):
for name in files:
path = os.path.join(root, name)Python security audit script for exposed risky filesystem state
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
Rails.application.configure do
config.after_initialize do
Bullet.enable = true
Bullet.alert = false
Bullet.bullet_logger = true
Bullet.console = trueN+1 query detection with Bullet gem
from django.urls import path
from . import views
app_name = 'blog'
urlpatterns = [Django URL namespacing and reverse lookups
Share this code
Here's the card — post it anywhere.
