Create a T-SQL code block to facilitate using the "OPENJSON trick"

openjson_block.py

# Copyright 2025 Gordon D. Thompson, gord@gordthompson.com
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# version 1.0 - 2025-12-01

"""
Create a T-SQL code block to facilitate using the "OPENJSON trick" documented at

https://github.com/mkleehammer/pyodbc/wiki/Tips-and-Tricks-by-Database-Platform#passing-row-oriented-parameter-data-as-a-json-string

The column names and types of the user-defined Table type are obtained by reflection.
"""

class OpenjsonBlock:
    def __init__(self, tt_schema, tt_name, tbl_var_name, cnx):
        self.tt_schema = tt_schema
        self.tt_name = tt_name
        self.tbl_var_name = tbl_var_name
        self.cnx = cnx

        _sql = """\
select 
     c.name as column_name
    ,case 
         when t.name in ('nchar', 'nvarchar') 
             then
                 case
                     when c.max_length = -1
                     then t.name + '(max)' 
                     else t.name + '(' + cast(c.max_length / 2 as nvarchar(4)) + ')'
                 end
         when t.name in ('char', 'varchar') 
             then
                 case
                     when c.max_length = -1
                     then t.name + '(max)' 
                     else t.name + '(' + cast(c.max_length as nvarchar(4)) + ')'
                 end
         when t.name in ('decimal', 'numeric') 
             then t.name + '(' + cast(c.precision as nvarchar(4)) + ', ' + cast(c.[scale] as nvarchar(4)) + ')'
        else t.name
     end
     as column_type
from 
    sys.table_types tt 
    inner join sys.schemas s on tt.schema_id = s.schema_id 
    inner join sys.columns c on tt.type_table_object_id = c.object_id 
    inner join sys.types t on c.user_type_id = t.user_type_id 
where s.name = ? and tt.name = ?
order by c.column_id 
        """
        self.col_info = cnx.execute(_sql, (tt_schema, tt_name)).fetchall()

    @property
    def col_name_list(self):
        return [col.column_name for col in self.col_info]

    @property
    def sql(self):
        col_name_str = ", ".join(["[" + col.column_name + "]" for col in self.col_info])
        s = f"DECLARE {self.tbl_var_name} [{self.tt_schema}].[{self.tt_name}];\n"
        s += f"INSERT INTO {self.tbl_var_name} ({col_name_str})\n"
        s += f"SELECT {col_name_str} FROM OPENJSON(?)\n"
        s += "WITH (\n"
        first = True
        for col in self.col_info:
            if not first:
                s += ",\n"
            first = False
            s += f"    [{col.column_name}] {col.column_type} '$.\"{col.column_name}\"'"
        s += "\n);"
        return s


if __name__ == "__main__":
    from decimal import Decimal
    import json
    import mssql_python

    cnxn = mssql_python.connect(
        "SERVER=192.168.0.199;"
        "DATABASE=test;"
        "UID=scott;PWD=tiger^5HHH;"
        "TrustServerCertificate=yes;"
    )

    ob = OpenjsonBlock("dbo", "my_table_type", "@tvp", cnxn)

    # inspect properties
    print(ob.col_name_list)
    """
['id', 'name', 'description', 'unit price']
    """
    print(ob.sql)
    """
DECLARE @tvp [dbo].[my_table_type];
INSERT INTO @tvp ([id], [name], [description], [unit price])
SELECT [id], [name], [description], [unit price] FROM OPENJSON(?)
WITH (
    [id] int '$."id"',
    [name] nvarchar(255) '$."name"',
    [description] nvarchar(max) '$."description"',
    [unit price] decimal(18, 4) '$."unit price"'
);
    """

    # usage example
    tvp_data = [(1, "unit", "single", Decimal("1.00")), (2, "pi", "raspberry", Decimal("3.14"))]
    tvp_json = [dict(zip(ob.col_name_list, row)) for row in tvp_data]
    print(tvp_json)
    """
[
 {'id': 1, 'name': 'unit', 'description': 'single', 'unit price': Decimal('1.00')}, 
 {'id': 2, 'name': 'pi', 'description': 'raspberry', 'unit price': Decimal('3.14')}
]
    """
    sql = f"""\
SET NOCOUNT ON;
{ob.sql}
EXEC dbo.my_tt_sp @tvp
    """
    print(sql)
    """
SET NOCOUNT ON;
DECLARE @tvp [dbo].[my_table_type];
INSERT INTO @tvp ([id], [name], [description], [unit price])
SELECT [id], [name], [description], [unit price] FROM OPENJSON(?)
WITH (
    [id] int '$."id"',
    [name] nvarchar(255) '$."name"',
    [description] nvarchar(max) '$."description"',
    [unit price] decimal(18, 4) '$."unit price"'
);
EXEC dbo.my_tt_sp @tvp
    """
    rows = cnxn.execute(sql, json.dumps(tvp_json, default=str)).fetchall()
    print(rows)
    """
[(1, 'unit', 'single', Decimal('1.0000')), (2, 'pi', 'raspberry', Decimal('3.1400'))]
    """