how to correctly process result for jsonb type with schema for sub field

[DeoLeung]

I have a jsonb field with pydantic enchaned

class PydanticExtra(types.TypeDecorator):
  """jsonb with pydantic schema"""
  cache_ok = True
  impl = JSONB

  def __init__(self, *args, model: Optional[BaseModel] = None, **kwargs):
    self.model = model
    # TODO(Deo): support exclude* parameters
    super().__init__(*args, **kwargs)

  def process_bind_param(self, value, dialect):
    if isinstance(value, BaseModel):
      return value.model_dump()
    return value

  def process_result_value(self, value, dialect):
    if isinstance(value, dict):
      return self.model(**value)
    return value

  class comparator_factory(JSONB.Comparator):

    def __or__(self, other):
      """it's concat"""
      return self.op('||', return_type=PydanticExtra)(other)

    def __getitem__(self, index):
      return super().__getitem__(index)

    def __eq__(self, other):
      if isinstance(other, BaseModel):
        other = other.model_dump()
      return self.op('=', return_type=Boolean)(other)

    def contains(self, other) -> ColumnOperators:
      if isinstance(other, BaseModel):
        other = other.model_dump()
      return self.op('@>', return_type=Boolean)(other)

assume the field is called extra
using select(table.c.extra) works correctly
when using select(table.c.extra['some_field']), if it returns a dict, it will be parsed as the pydantic model , how to properly instruct sqlalchemy to avoid process the result in this situation?

[CaselIT]

Hi,

I guess you can use a cast there. You could do that either manually or automatically in your custom __getitem__. The cast would be to jsonb directly.

It's likely enough to just tell sqlalchemy what type to use with type_coerce since for the db the type is already jsonb

[DeoLeung]

I updated the __getitem__ and it works

    def __getitem__(self, index):
      (
          adjusted_op,
          adjusted_right_expr,
          _,
      ) = self._setup_getitem(index)
      return self.operate(adjusted_op, adjusted_right_expr, result_type=JSONB)