For years now, JSON and JSONB have had support from major databases, and over the years, their support has only grown. Even though the idea of JSON structures kind of conflicts with the idea of Relational Databases, there is no denying that sometimes a flexible data structure can be quite beneficial.

JSON structures can make the work of a DB admin or data analyst less comfortable at times, since you cannot make any assumptions about the JSONB data when you encounter a column of this data type. To understand the contents, you must get a large enough set of rows that correctly reflects every permutation of what that column might hold.

When working with JSONB using SQLAlchemy, and mapping JSON or JSONB columns to properties on an SQLAlchemy model, we can deal with these data structures in many ways

• We can map JSON values to str which would make us unmarshal/stringify our input and output
• We can map JSON values to dict which, in many cases, might make our input and output match each other
• We can map JSON or JSONB values to a custom type that we define in our codebase

SQLAlchemy supports JSON and JSONB natively, which makes mapping it to a dict type natural and easy.

We will begin by creating our engine

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///:memory:', echo=True)
Session = sessionmaker(bind=engine)

To define our Declarative tables with support for JSON values, we can actually just define our tables without any kind of manipulation or custom types

from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
from sqlalchemy import Integer, String, JSON

class Base(DeclarativeBase):
  pass

class Creature(Base):
  __tablename__ = 'creatures'
  
  id: Mapped[int] = mapped_column(Integer, primary_key=True)
  name: Mapped[str] = mapped_column(String(200))
  habitat: Mapped[str] = mapped_column(String(200))
  # JSON field for abilities and characteristics
  traits: Mapped[dict] = mapped_column(JSON)

Base.metadata.create_all(engine)

Since SQLite supports JSON, when we migrate our tables we get JSON as a datatype for our tables.

Inserting data into JSON columns

Since we are mapping our column to a dict type we can very easily insert data into our table using a dict

with Session() as session:
  creatures = [
      Creature(name='Dragon', habitat='Mountains', traits={'abilities': ['fly', 'breathe fire'], 'size': 'large'}),
      Creature(name='Unicorn', habitat='Forests', traits={'abilities': ['heal', 'teleport'], 'size': 'medium'}),
      Creature(name='Mermaid', habitat='Oceans', traits={'abilities': ['swim', 'sing'], 'size': 'medium'}),
  ]
  session.add_all(creatures)
  session.commit()

From the output we can tell that SQLAlchemy actually dumps our dict to the database, so all of these operations that we would have to do ourselves are being done for us.

Getting back our data

When we pull our JSON column back from the database, SQLAlchemy also loads our json into a dict

with Session() as session:
  stmt = session.query(Creature).filter(Creature.habitat == 'Mountains')
  for creature in stmt:
      print(f"Creature: {creature.name}, Habitat: {creature.habitat}, size: {creature.traits['size']}")
      # Accessing JSON field
      for ability in creature.traits['abilities']:
          print(f" - Ability: {ability}")

Looking at the output, we can see that we are getting back a dict from our objects.

This is great for many cases; if we were using dataclasses or Pydantic, for example, it would be incredibly easy to convert these dictionaries into typed objects, making it much easier to work with in the context of a large application.

Speaking of dataclasses…

If we wanted to use dataclasses, we couldn’t just map our classes to a dataclass and expect it to work, the reason is simply that SQLAlchemy is very reliant on the identity_map when converting SQL datatypes to Pythonic types.

In this case, we can use two concepts we’ve already touched on in a previous article, which are the identity_map and TypeDecorator which would allow us to create our own datatypes and, in this case, map them to a specific dataclass

We begin by creating a dataclass

from dataclasses import dataclass

@dataclass
class ArtifactPower:
  name: str
  description: str
  power_level: int

Our dataclass will be used as part of our following declarative table, but first, we need to create a brand new type decorator, which will accept our dataclass, and allow us to both write and read our column as our new type when using our new declarative table.

Creating our TypeDecorator

Creating a type decorator is straightforward

from sqlalchemy.types import TypeDecorator
from sqlalchemy import JSON
import json
from dataclasses import asdict

class DataclassFromJSON(TypeDecorator):
  impl = JSON
  
  def __init__(self, dataclass, *args, **kwargs):
      super().__init__(*args, **kwargs)
      self.dataclass = dataclass

  def process_bind_param(self, value, dialect):
      if value is not None:
          return json.dumps(asdict(value))
      return value

  def process_result_value(self, value, dialect):
      if value is not None:
          value = json.loads(value)
          value = self.dataclass(**value)
      return value

What this decorator does is:

1. Defines the underlying datatype using the impl property
2. 1. This property is used by SQLAlchemy begin the scenes when constructing queries
2. We create a constructor to be able to pass the dataclass to the class
3. The process_bind_param function is where we override the “binding” process where SQLAlchemy binds values as part of the query construction
4. The process_result_value function overrides the way SQLAlchemy constructs our objects that come back from the database for this specific type

Using the TypeDecorator

Since our new DataclassFromJSON class supports any dataclass we can use it as part of our mapped_column instead of defining it as part of the type_annotation_map that we’ve seen in previous articles

class Artifact(Base):
  __tablename__ = 'artifacts'
  id: Mapped[int] = mapped_column(Integer, primary_key=True)
  name: Mapped[str] = mapped_column(String(200))
  origin: Mapped[str] = mapped_column(String(200))
  # JSON field for magical properties
  powers: Mapped[ArtifactPower] = mapped_column(DataclassFromJSON(ArtifactPower))
  
Base.metadata.create_all(engine)

In this case we are using our dataclass twice

• Once in our Mapped annotation
• Once as a param to the DataclassFromJSON type decorator __init__ function

We can actually tell from the output of the create_all function that SQLAlchemy actually generates our tables using the JSON type, meaning that by using our new type decorator SQLAlchemy finds out which datatype to use when creating our table.

Writing to our new datatype

with Session() as session:
  artifact = Artifact(
      name='Excalibur',
      origin='Avalon',
      powers=ArtifactPower(
          name='Sword of Destiny',
          description='A legendary sword that grants its wielder immense power.',
          power_level=100
      )
  )
  session.add(artifact)
  session.commit()

Since our TypeDecorator implements both the process_bind_param and the process_result_value functions, we can both write and read the powers column using dataclasses.

Reading from our new Table

from sqlalchemy import select

with Session() as session:
  stmt = select(Artifact).filter(Artifact.name == 'Excalibur')
  artifact = session.execute(stmt).scalars().first()
  if artifact:
      print(f"Artifact: {artifact.name}, Origin: {artifact.origin}")
      print(f" - Power Name: {artifact.powers.name}")
      print(f" - Power Description: {artifact.powers.description}")
      print(f" - Power Level: {artifact.powers.power_level}")

By using the Mapped annotation, we can also enjoy type hinting in this case, as well as make sure that our code is type safe.

Note

This pattern is beneficial for more than just these simple datatypes; in reality, it probably makes sense to implement this pattern when your code uses many dataclasses / Pydantic classes or when you have complex data structures that you need in your database.

It can also be useful when you work with arrays or large result sets that you need to be able to load into your program later.

JSONB operations with PostgreSQL

While SQLite provides excellent support for JSON, when we move to more production-focused databases like PostgreSQL, we often encounter its more advanced binary JSON type: JSONB. JSONB offers several advantages over the standard JSON type, including indexing capabilities and often more efficient storage and querying due to its decomposed binary format.

services:
  postgres:
      image: postgres:latest
      environment:
          POSTGRES_USER: postgres
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: postgres
      ports:
          - "5432:5432"
      volumes:
          - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

SQLAlchemy >= 2.0.40
psycopg2-binary >= 2.9.10

The good news is that SQLAlchemy handles JSONB just as easily as JSON. Let's switch gears and set up a PostgreSQL connection to explore some JSONB-specific operations.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

pg_engine = create_engine('postgresql+psycopg2://user:password@host:port/database', echo=True)
PgSession = sessionmaker(bind=pg_engine)

Next, we'll define our Alchemist and a related Potion table. The relationship between Alchemist and Potion will be based on an ID stored within the Alchemist's JSONB column.

from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
from sqlalchemy import Integer, String, ForeignKey, func
from sqlalchemy.dialects.postgresql import JSONB

class Base(DeclarativeBase):
  pass

class Potion(Base):
  __tablename__ = 'potions'
  id: Mapped[int] = mapped_column(Integer, primary_key=True)
  name: Mapped[str] = mapped_column(String(100), unique=True)
  effect: Mapped[str] = mapped_column(String(250))

class Alchemist(Base):
  __tablename__ = 'alchemists'

  id: Mapped[int] = mapped_column(Integer, primary_key=True)
  name: Mapped[str] = mapped_column(String(150))
  lab_notes: Mapped[dict] = mapped_column(JSONB)

  favorite_potion: Mapped["Potion"] = relationship(
      "Potion",
      primaryjoin=lambda: Potion.id == Alchemist.lab_notes['favorite_potion_id'].astext.cast(Integer),
      foreign_keys=[Potion.id], # Explicit foreign_keys helps SQLAlchemy understand the join
      lazy='joined' # Or 'select', depending on your preference
  )

Base.metadata.create_all(pg_engine)

Populating the database

with PgSession() as session:
  healing_potion = Potion(name='Healing Draught', effect='Restores minor wounds')
  strength_elixir = Potion(name='Elixir of Strength', effect='Temporarily boosts strength')
  invisibility_philter = Potion(name='Invisibility Philter', effect='Grants temporary invisibility')

  session.add_all([healing_potion, strength_elixir, invisibility_philter])
  session.commit()

  alchemists_data = [
      Alchemist(name='Elara', lab_notes={'level': 15, 'discoveries': {'transmutation': 5, 'illusions': 3}, 'inventory_slots': 50, 'affiliation': 'Golden Mortar', 'favorite_potion_id': healing_potion.id}),
      Alchemist(name='Roric', lab_notes={'level': 12, 'specialties': {'poisons': 9, 'explosives': 7}, 'inventory_slots': 75, 'affiliation': "Serpent's Vial", 'favorite_potion_id': strength_elixir.id}),
      Alchemist(name='Lyna', lab_notes={'level': 14, 'experiments': {'growth_serum': 'ongoing', 'elemental_infusion': 'stable'}, 'inventory_slots': 60, 'favorite_potion_id': invisibility_philter.id}),
      Alchemist(name='Borin', lab_notes={'level': 10, 'discoveries': {'transmutation': 2}, 'inventory_slots': 40, 'affiliation': 'Golden Mortar'}),
  ]
  session.add_all(alchemists_data)
  session.commit()

Accessing our data

We can query our JSONB data in multiple different ways, one way is to use the .op() function which will allow us to write our SQLAlchemy queries in a very similar way to how we would do it in native PostgreSQL, another is to use functions that SQLAlchemy provides for us.

Let look at an example

from sqlalchemy import select, cast

with PgSession() as session:
  stmt_direct = select(Alchemist).filter(
      Alchemist.lab_notes['discoveries']['transmutation'].cast(Integer) > 3
  )
  elara = session.execute(stmt_direct).scalars().first()
  if elara:
      print(f"Direct Access: {elara.name} has transmutation > 3. Level: {elara.lab_notes.get('level')}")

  stmt_op = select(Alchemist).filter(
      Alchemist.lab_notes.op('->')('discoveries').op('->>')('transmutation').cast(Integer) > 3
  )
  elara_op = session.execute(stmt_op).scalars().first()
  if elara_op:
      print(f"Operator Access (.op): {elara_op.name} has transmutation > 3. Level: {elara_op.lab_notes.get('level')}")

Let break this down

1. In the first example we are accessing nested key in the JSONB using the dictionary bracket notation
2. 1. SQLAlchemy automatically converts this notation to the JSONB extraction operator (→)
   2. We cast the extracted key to Integer to make sure that we use the > operator on it
2. In the second example we are directly using JSONB ops as part of our query constructors
3. 1. The -> operator is our JSONB extractor operator which all extract the right sided key from the JSONB object
   2. The second operator is the ->> operator which extracts the value as text, which we can again use the cast() function on to construct our query

Usage as selectable columns

We can also use the same idea to select specific JSONB columns from our database

with PgSession() as session:
  stmt_affiliation_direct = select(Alchemist.lab_notes['affiliation']).filter(Alchemist.name == 'Roric')
  roric_affiliation_direct = session.execute(stmt_affiliation_direct).scalars().first()
  if roric_affiliation_direct:
      print(f"Roric's affiliation (direct): {roric_affiliation_direct}")

  stmt_affiliation_op = select(Alchemist.lab_notes.op('->>')('affiliation')).filter(Alchemist.name == 'Roric')
  roric_affiliation_op = session.execute(stmt_affiliation_op).scalars().first()
  if roric_affiliation_op:
      print(f"Roric's affiliation (.op): {roric_affiliation_op}")

Checking for key existence

We can also filter by the existence of keys in our JSONB columns, SQLAlchemy provides us with both the option to again use the op() function as well as the has_key function

with PgSession() as session:
  stmt_has_key_direct = select(Alchemist).where(
      Alchemist.lab_notes.has_key('affiliation')
  )
  print("Alchemists with 'affiliation' key (direct):")
  for alchemist in session.execute(stmt_has_key_direct).scalars():
      print(f"- {alchemist.name}, Affiliation: {alchemist.lab_notes['affiliation']}")

  stmt_has_key_op = select(Alchemist).where(
      Alchemist.lab_notes.op('?')('affiliation')
  )
  print("Alchemists with 'affiliation' key (.op):")
  for alchemist in session.execute(stmt_has_key_op).scalars():
      print(f"- {alchemist.name}, Affiliation: {alchemist.lab_notes['affiliation']}")

We can also check for when keys do not exists

from sqlalchemy import not_

with PgSession() as session:
  stmt_no_specialties_direct = select(Alchemist).where(
      ~Alchemist.lab_notes.has_key('affiliation')
  )
  print("Alchemists without 'affiliation' key (direct):")
  for alchemist in session.execute(stmt_no_specialties_direct).scalars():
      print(f"- {alchemist.name}")

  stmt_no_specialties_op = select(Alchemist).where(
      not_(Alchemist.lab_notes.op('?')('affiliation'))
  )
  print("Alchemists without 'affiliation' key (.op):")
  for alchemist in session.execute(stmt_no_specialties_op).scalars():
      print(f"- {alchemist.name}")

in these queries we are using two different ways for creating a query with the NOT keyword

1. Using the not_() function that we’ve already seen in previous articles, which negates our expression
2. Using the ~ binary operator which is actually overridden by SQLAlchemy for us, and we can use it on our query to negate our expressions

Querying relationships based on JSONB

If you take a look at our Alchemist table definition you can see that we’ve actually created a relationship that is based on a field that exists in our JSONB column, when defining relationships we can actually override the default method that SQLAlchemy use use to query the database and populate our relationships with the required collections.

Using this relationship is straightforward

from sqlalchemy.orm import joinedload

with PgSession() as session:
  print("Alchemists and their favorite potions (via relationship):")
  stmt_alchemists_with_potions = select(Alchemist).options(
      joinedload(Alchemist.favorite_potion)
  )

  alchemists = session.execute(stmt_alchemists_with_potions).scalars().unique().all()

  for alchemist in alchemists:
      if alchemist.favorite_potion:
          print(f"- Alchemist: {alchemist.name}, Favorite Potion: {alchemist.favorite_potion.name} (Effect: {alchemist.favorite_potion.effect})")
      elif 'favorite_potion_id' in alchemist.lab_notes :
              print(f"- Alchemist: {alchemist.name}, Favorite Potion ID: {alchemist.lab_notes['favorite_potion_id']} (Potion not found or relation misconfigured for this entry)")
      else:
          print(f"- Alchemist: {alchemist.name} has no favorite potion listed in lab_notes.")

  # Example: Find alchemists whose favorite potion is 'Healing Draught'
  stmt_fav_healing = select(Alchemist).join(Alchemist.favorite_potion).filter(Potion.name == 'Healing Draught')
  print("Alchemists whose favorite potion is 'Healing Draught':")
  for alchemist in session.execute(stmt_fav_healing).scalars():
      print(f"- {alchemist.name}")

Using a basic joinedload we can pull all of our Alchemists along with their favorite potion just as we would any other relationship, behind the scenes SQLAlchemy constructs the query to pull all of this information using the primaryjoin argument we provided to the relationship function.

This kind of relationship mapping provides us with a powerful pattern for creating groups of objects that rely on JSONB data to collect information from the database

This can be especially useful in cases of polymorphic relationships, where your each polymorphic identity might hold different JSONB structures, and creating relationships specific to a polymorphic identity can be based on the difference in JSON data.

Another more advanced use case can be the creation of dynamic relationships based on data you might find in a JSONB column, where you might use the Hybrid Declarative approach to create table definitions and relationships dynamically based on user queries (think of an advanced data analysis system that allows users to create KPIs and data structures based on their specific needs)

Summary

JSON and JSONB in a relational database can prove to be very useful in systems that aggregate a lot of data from different sources, that might relate to our classic relational models well, in these cases where we are able to programmatically predict their structures, these datatypes prove to be highly wieldable.

While JSON and JSONB may seem like a silver bullet when dealing with large result sets, I would still recommend to use them carefully, they are not a datatype to be used instead of a database table as a way of “saving time”, and should not be used in any case that does not justify their unique properties.

That being said, these datatypes are a great way of enjoying some of the benefits of document based databases in a relational database, even if maintaining them adds quite a bit of overhead, and combined with SQLAlchemy, they open up a world of possibilities to developers that understand the SQLAlchemy Declarative / Hybrid Declarative approaches well.