SQLAlchemy JSONField

1.0.2 · active · verified Thu Apr 09

`sqlalchemy-jsonfield` provides a `JSONField` implementation for SQLAlchemy, allowing Python dictionaries to be stored in database JSON columns. It includes `JSONMutableDict` for automatic change tracking of mutable dictionary operations. The current stable version is 1.0.2, and it follows an infrequent, maintenance-focused release cadence.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to define a model with a `JSONField` that uses `JSONMutableDict` for automatic change detection. It shows creating, retrieving, and updating JSON data in-place, with changes being correctly persisted to an in-memory SQLite database.

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from sqlalchemy_jsonfield import JSONField, JSONMutableDict

# Define the base for declarative models
Base = declarative_base()

# Define a model with a JSONField
class Data(Base):
    __tablename__ = 'data'
    id = sa.Column(sa.Integer, primary_key=True)
    # Use JSONField with JSONMutableDict to enable automatic change tracking
    json_data = sa.Column(JSONField(JSONMutableDict), default={})

# Setup database engine and session
engine = sa.create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

try:
    # Create a new item
    item = Data(json_data={'initial_key': 'initial_value', 'list_data': [1, 2]})
    session.add(item)
    session.commit()
    print(f"Created item ID: {item.id}, Data: {item.json_data}")

    # Retrieve the item
    retrieved_item = session.query(Data).filter_by(id=item.id).first()
    print(f"Retrieved item ID: {retrieved_item.id}, Data: {retrieved_item.json_data}")

    # Update the JSON data in-place (changes detected by JSONMutableDict)
    retrieved_item.json_data['new_key'] = 'new_value'
    retrieved_item.json_data['list_data'].append(3)
    session.commit()
    print(f"Updated item ID: {retrieved_item.id}, Data: {retrieved_item.json_data}")

    # Verify update by re-retrieving
    verified_item = session.query(Data).filter_by(id=item.id).first()
    print(f"Verified item ID: {verified_item.id}, Data: {verified_item.json_data}")

finally:
    session.close()

view raw JSON →