Model stored/encoded as a delimited string in the database?

[kennedy]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

ActualData(BaseModel):
    id: int
    KnownName: str
    NickName: str
    Mother: str
    Father: str
    SocialSecurity: int
    Pets: Optional[boolean]
    Vegan: Optional[boolean]

Database Schema:
    id: int
    name: str # KnownName
    details: str # f"KnownName={KnownName};NickName={NickName};Mother={Mother};Father={Father};SocialSecurity={SocialSecurity};Pets={Pets};Vegan={Vegan}"

Description

I have a very strange constraint, I am using SQLModel as my orm. I have a database with two important fields, for example name and details. The important information I need for my model ActualData is built from the details column from the database.

How do I use the ActualData model in my code, but when I submit/save/read from the database, it is encoded as a character delimited string structure?

pydantic/validators was very helpful but it fills one field. Is it possible to fill out the entire model with one validator? How does one encode the model back to a single string within the database?

Operating System

Windows

Operating System Details

• win11 pro, python from windows store, pipenv

SQLModel Version

0.0.8

Python Version

3.10.8

Additional Context

Think of storing the string components of postgresql database connection string, but storing it as a completed connection string

User ID=root;Password=myPassword;Host=localhost;Port=5432;Database=myDataBase;Pooling=true;Min Pool Size=0;Max Pool Size=100;Connection Lifetime=0;

[meirdev]

I recommend you use a JSON column for details. anyway, you have to create a new type:

from sqlalchemy.types import TypeDecorator, String

class ActualDataType(TypeDecorator):
    impl = String

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = ";".join(f"{k}={v}" for k, v in value.dict().items() if v is not None)
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            d = {}
            for value in value.split(";"):
                k, v = value.split("=")
                d[k] = v
            value = ActualData(**d)
        return value

for JSON:

from sqlalchemy.types import TypeDecorato, JSON

class ActualDataType(TypeDecorator):
    impl = JSON

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = value.dict()
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = ActualData(**value)
        return value

And finally, change the column type to be ActualDataType:

from sqlalchemy import Column

class Data(SQLModel, table=True):
    ...
    details: ActualData = Field(sa_column=Column(ActualDataType))

[YuriiMotov]

Working code example in details:

from pydantic import BaseModel
from sqlalchemy import Column
from sqlalchemy.types import JSON, TypeDecorator
from sqlmodel import Field, Session, SQLModel, create_engine


class ActualData(BaseModel):
    id: int
    KnownName: str


class ActualDataType(TypeDecorator):
    impl = JSON

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = value.model_dump()
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = ActualData(**value)
        return value


class Data(SQLModel, table=True):
    id: int = Field(primary_key=True)
    details: ActualData = Field(sa_column=Column(ActualDataType))


engine = create_engine("sqlite:///", echo=True)

SQLModel.metadata.create_all(engine)

def test_():
    # Insert
    with Session(engine) as session:
        session.add(Data(details=ActualData(id=123, KnownName="qwerty")))
        session.commit()

    # Query
    with Session(engine) as session:
        d = session.get(Data, 1)
    assert d.id == 1
    assert d.details.id == 123
    assert d.details.KnownName == "qwerty"

[kennedy]

@meirdev you are amazing, thank you for your suggestion. I will try this out and report back.