Postgres JSONB Parameters

[jriggins]

Hi. I'm brand new to Mayim and was wondering how one might insert data into PostgreSQL JSONB columns.

class Blob(pydantic.BaseModel):
    id: int
    key: str
    data: dict = None


@mayim.register
class BlobExecutor(mayim.PostgresExecutor):
    @mayim.query(
        """
        CREATE TABLE IF NOT EXISTS blobs (id SERIAL, key VARCHAR, data JSONB)
        """
    )
    def create(self):
        ...

    @mayim.query(
        """
        DROP TABLE IF EXISTS blobs
        """
    )
    def drop(self):
        ...

    @mayim.query(
        """
        INSERT INTO blobs (key, data)
        VALUES ($key, $data)
        """
    )
    def insert(self, key: str, data: dict):
        ...

    @mayim.query(
        """
        SELECT id, key, data FROM blobs
        WHERE key = $key
        """
    )
    def select(self, key: int) -> Blob:
        ...


class TestJsonb:
    @pytest.fixture
    async def blob_executor(self):
        mayim.Mayim(dsn=os.environ["DATABASE_URL"])
        executor = BlobExecutor()
        await executor.create()
        try:
            yield executor
        finally:
            await executor.drop()

    async def test_works(self, blob_executor: BlobExecutor):
        # works
        await blob_executor.insert(key="none", data=None)
        none_result = await blob_executor.select(key="none")
        assert none_result

        # fails
        await blob_executor.insert(key="with_data", data={"sample": 123})
        result = await blob_executor.select(key="with_data")
        assert result

>           raise ex.with_traceback(None)
E           psycopg.ProgrammingError: cannot adapt type 'dict' using placeholder '%s' (format: AUTO)

binary     = False
cur        = <psycopg.AsyncCursor [no result] [IDLE] (host=rockhopper-datastore user=db_user database=devdb) at 0xffffa5998670>
params     = {'data': {'sample': 123}, 'key': 'with_data'}
prepare    = None
query      = 'INSERT INTO blobs (key, data)\nVALUES (%(key)s, %(data)s)'
self       = <psycopg.AsyncConnection [IDLE] (host=rockhopper-datastore user=db_user database=devdb) at 0xffffa5819ed0>

apiserver/venv/lib/python3.11/site-packages/psycopg/connection_async.py:288: ProgrammingError

[jriggins]

I actually came up with this that seems to work. Is this the best way?

    @mayim.query(
        """
        INSERT INTO blobs (key, data)
        VALUES ($key, $data)
        """
    )
    async def insert(self, key: str, data: dict):
        import psycopg.types.json as js

        query = self.get_query("insert")
        await self.execute(query, params={"key": key, "data": js.Jsonb(data)})