"ValueError: <class 'pydantic.networks.IPvAnyAddress'> has no matching SQLAlchemy type" while trying to use AutoString or Column(String) supposed solutions

[Sulray]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question 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

from sqlmodel import SQLModel, Field, Column, AutoString, String
from pydantic import IPvAnyAddress

class DeviceBase(SQLModel, table=true):
    name: str = Field(index=True, sa_column_kwargs={"unique": True})
    management_ip: IPvAnyAddress = Field()
    # management_ip: IPvAnyAddress = Field(sa_type=AutoString)
    # management_ip: IPvAnyAddress = Field(sa_column=Column(String))

Description

My goal was to have data validation for IP Addresses on my project:

• At first I just had management_ip: IPvAnyAddress = Field() to add the field but it led to the error ValueError: <class 'pydantic.networks.IPvAnyAddress'> has no matching SQLAlchemy type

• Then I found the following discussion ValueError: <class 'pydantic.networks.EmailStr'> has no matching SQLAlchemy type #730 which was on a similar topic and tried to apply the two proposed solutions

• The first solution was to use sa_type=AutoString leading to this line of code : management_ip: IPvAnyAddress = Field(sa_type=AutoString)
  However it led to an unexpected error when trying to post one of these objects into my database : sqlalchemy.exc.DBAPIError: (sqlalchemy.dialects.postgresql.asyncpg.Error) <class 'asyncpg.exceptions.DataError'>: invalid input for query argument $2: IPv4Address('12.34.56.78') (expected str, got IPv4Address)

• The second solution was to use sa_column=Column(String) leading to this line of code : management_ip: IPvAnyAddress = Field(sa_column=Column(String))
  However it led to the same exact error when trying to post data.

• It looks like the solutions aren't being properly applied looking at the logs of the insertion in the SQL database :

  [SQL: INSERT INTO device (name, management_ip) VALUES ($1::VARCHAR, $2::VARCHAR) RETURNING device.id]
  [parameters: ('hostname', IPv4Address('12.34.56.78'))]
  

Do you have any idea about that ?
Does anyone manage to use successfully this feature to do type validation ?

Operating System

Linux

Operating System Details

Ubuntu 22.04.4 LTS

SQLModel Version

0.0.16

Python Version

Python 3.11.4

Additional Context

Relevant dependencies :

• asyncpg 0.29.0
• databases 0.9.0
• fastapi 0.110.0
• pip 24.0
• pydantic 2.6.3
• SQLAlchemy 2.0.28
• sqlmodel 0.0.16
• uvicorn 0.27.1

[Sulray]

Any idea ? Does anyone do type validation on IP address like that ?

[SamuelGuillemet]

FYI the issue encountered with the insertion is not directly linked to SQLModel but rather to SQLAlchemy dialect which can't convert IPv4Address or IPv6Address to a DB element.

What could be done to fix the issue is creating a custom SQLAlchemy type which insert the pydantic IPvAnyAddress (which is by the way an union of IPv4Address and IPv6Address) as a String and retreive the object as a IPvAnyAddress.

from typing import Optional

from sqlmodel import AutoString, Field, SQLModel
from pydantic import IPvAnyAddress


class IPvAnyAddressType(AutoString):
    def process_bind_param(self, value, dialect) -> Optional[str]:
        if value is None:
            return None

        if isinstance(value, str):
            # Test if value is a valid IP address to avoid process result value failling
            try:
                IPvAnyAddress(value)
            except ValueError as e:
                raise ValueError(f"{value} is not a valid IP address") from e

        return str(value)

    def process_result_value(self, value, dialect) -> Optional[IPvAnyAddress]:
        if value is None:
            return None

        return IPvAnyAddress(value)

class DeviceBase(SQLModel, table=True):
    # ...
    management_ip: IPvAnyAddress = Field(sa_type=IPvAnyAddressType)
    # ...

This approach will allow you to use the IPvAnyAddress type in your models and have the correct type in the database, i.e. VARCHAR or TEXT depending on the database you are using.

[SamuelGuillemet]

And this function could be used to generate SQLAlchemy type :

from typing import Optional, Type, TypeVar

from sqlmodel import AutoString, Field, SQLModel
from pydantic import IPvAnyAddress

T = TypeVar("T")


def build_custom_type(internal_type: Type[T]) -> Type[AutoString]:
    class CustomType(AutoString):
        def process_bind_param(self, value, dialect) -> Optional[str]:
            if value is None:
                return None

            if isinstance(value, str):
                # Test if value is valid to avoid `process_result_value` failling
                try:
                    internal_type(value)  # type: ignore[call-arg]
                except ValueError as e:
                    raise ValueError(f"Invalid value for {internal_type.__name__}: {e}") from e

            return str(value)

        def process_result_value(self, value, dialect) -> Optional[T]:
            if value is None:
                return None

            return internal_type(value)  # type: ignore[call-arg]

    return CustomType


class DeviceBase(SQLModel):
    management_ip: IPvAnyAddress


class DeviceTable(DeviceBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    management_ip: IPvAnyAddress = Field(sa_type=build_custom_type(IPvAnyAddress))