sqlalchemy-json

import os from datetime import datetime from sqlalchemy import create_engine, Column, Text, ForeignKey, Integer, DateTime from sqlalchemy.orm import declarative_base, sessionmaker, Mapped, mapped_column from sqlalchemy_json import MutableJson, NestedMutableJson # Setup database (in-memory SQLite for example) DATABASE_URL = os.environ.get('DATABASE_URL', 'sqlite:///:memory:') engine = create_engine(DATABASE_URL, echo=True) Session = sessionmaker(bind=engine) Base = declarative_base() # Define models class Author(Base): __tablename__ = "authors" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(Text) # MutableJson for top-level changes handles: Mapped[dict] = mapped_column(MutableJson) class Article(Base): __tablename__ = "articles" id: Mapped[int] = mapped_column(Integer, primary_key=True) author_id: Mapped[int] = mapped_column(ForeignKey('authors.id')) content: Mapped[str] = mapped_column(Text) # NestedMutableJson for deep changes references: Mapped[dict] = mapped_column(NestedMutableJson) # Create tables Base.metadata.create_all(engine) session = Session() # Example for MutableJson author = Author(name='John Doe', handles={'twitter': '@JohnDoe', 'facebook': 'JohnDoe'}) session.add(author) session.commit() # Retrieve and modify top-level JSON retrieved_author = session.query(Author).first() print(f"Original handles: {retrieved_author.handles}") retrieved_author.handles['twitter'] = '@JDoe' session.commit() # Change is detected print(f"Updated handles (MutableJson): {retrieved_author.handles}") # Example for NestedMutableJson article = Article( author_id=retrieved_author.id, content='Some article content', references={'github.com': {'repo1': 4, 'repo2': 7}, 'example.com': {'link1': 2}} ) session.add(article) session.commit() # Retrieve and modify nested JSON retrieved_article = session.query(Article).first() print(f"Original references: {retrieved_article.references}") retrieved_article.references['github.com']['repo1'] += 10 # Nested change session.commit() # Change is detected print(f"Updated references (NestedMutableJson): {retrieved_article.references}") session.close()