"Can't generate DDL for NullType()" when using pydantic.SecretStr

[maxmouchet]

First Check

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

Commit to Help

• [X] I commit to help with one of those options 👆

Example Code

from pydantic import SecretStr
from sqlmodel import Field, Session, SQLModel, create_engine

class A(SQLModel, table=True):
    id: int = Field(primary_key=True)
    secret: SecretStr 

engine = create_engine("sqlite://")
SQLModel.metadata.create_all(engine)

# sqlalchemy.exc.CompileError: (in table 'a', column 'secret'): Can't generate DDL for NullType(); did you forget to specify a type on this Column?

Description

• Create a class with a pydantic.SecretStr field
• Create the table
• Execution fails with Can't generate DDL for NullType() exception

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.4

Python Version

3.9.6

Additional Context

We can specify the column type manually, but it fails with another error:

from pydantic import SecretStr
from sqlmodel import Column, Field, Session, String, SQLModel, create_engine

class A(SQLModel, table=True):
    id: int = Field(primary_key=True)
    secret: SecretStr = Field(sa_column=Column(String))

engine = create_engine("sqlite://")
SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    a = A(id=1, secret="secret")
    session.add(a)
    session.commit()

# sqlalchemy.exc.InterfaceError: (sqlite3.InterfaceError) Error binding parameter 0 - probably unsupported type.
# [SQL: INSERT INTO a (secret, id) VALUES (?, ?)]
# [parameters: (SecretStr('**********'), 1)]

[artemkalugin01]

If I am not mistaken column secret should be declared as: secret: str = Field(sa_column=Column(String)) So specifying type using Type Hinting we should only use types from standard python library.

[redb0]

I recently came across a similar task and came across this question.

I understand that it has been a long time, but it may be useful to someone.

You can use TypeDecorator from sqlalchemy (https://docs.sqlalchemy.org/en/20/core/custom_types.html#sqlalchemy.types.TypeDecorator)

It will look something like this

import sqlalchemy as sa

class SecretStrType(sa.types.TypeDecorator):
    impl = sa.types.TEXT

    def process_bind_param(self, value: SecretStr, dialect):
        return value.get_secret_value()

    def process_result_value(self, value: str, dialect):
        return SecretStr(value)

and full example

from typing import Optional

import sqlalchemy as sa
from pydantic import SecretStr
from sqlmodel import Field, Session, SQLModel, create_engine, select


class SecretStrType(sa.types.TypeDecorator):
    impl = sa.types.TEXT

    def process_bind_param(self, value: SecretStr, dialect):
        return value.get_secret_value()

    def process_result_value(self, value: str, dialect):
        return SecretStr(value)


class A(SQLModel, table=True):
    id: Optional[int] = Field(  # type: ignore[call-overload]
        title='Идентификатор',
        default=None,
        primary_key=True,
        description='Идентификатор',
    )
    access_token: SecretStr = Field(  # type: ignore[call-overload]
        title='Логин',
        min_length=5,
        max_length=255,
        sa_type=SecretStrType(),
        nullable=False,
        description='Логин',
    )


engine = create_engine(url='postgresql://postgres:postgres@localhost:5432/postgres')


SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    a = A.model_validate({"id": 1, "access_token": "secret"})
    session.add(a)
    session.commit()

with Session(engine) as session:
    res = session.exec(select(A))
    for i in res:
        print(f'{i = }')