Migrate PostgresDsn.build from pydentic v1 to pydantic v2

Question

I have simple Config class from FastAPI tutorial. But it seems like it uses old pydantic version. I run my code with pydantic v2 version and get a several errors. I fix almost all of them, but the last one I cannot fix yet. This is part of code which does not work:

from pydantic import AnyHttpUrl, HttpUrl, PostgresDsn, field_validator
from pydantic_settings import BaseSettings
from pydantic_core.core_schema import FieldValidationInfo

load_dotenv()


class Settings(BaseSettings):
    ...
    POSTGRES_SERVER: str = 'localhost:5432'
    POSTGRES_USER: str = os.getenv('POSTGRES_USER')
    POSTGRES_PASSWORD: str = os.getenv('POSTGRES_PASSWORD')
    POSTGRES_DB: str = os.getenv('POSTGRES_DB')
    SQLALCHEMY_DATABASE_URI: Optional[PostgresDsn] = None

    @field_validator("SQLALCHEMY_DATABASE_URI", mode='before')
    @classmethod
    def assemble_db_connection(cls, v: Optional[str], info: FieldValidationInfo) -> Any:
        if isinstance(v, str):
            return v
        postgres_dsn = PostgresDsn.build(
            scheme="postgresql",
            username=info.data.get("POSTGRES_USER"),
            password=info.data.get("POSTGRES_PASSWORD"),
            host=info.data.get("POSTGRES_SERVER"),
            path=f"{info.data.get('POSTGRES_DB') or ''}",
        )
        return str(postgres_dsn)

That is the error which I get:

sqlalchemy.exc.ArgumentError: Expected string or URL object, got MultiHostUrl('postgresql://user:password@localhost:5432/database')

I check a lot of places, but cannot find how I can fix that, it looks like build method pass data to the sqlalchemy create_engine method as a MultiHostUrl instance instead of string. How should I properly migrate this code to use pydantic v2?

UPDATE

I have fixed that issue by changing typing for SQLALCHEMY_DATABASE_URI: Optional[PostgresDsn] = None to SQLALCHEMY_DATABASE_URI: Optional[str] = None. Because pydantic makes auto conversion of result for some reason. But I am not sure if that approach is the right one, maybe there are better way to do that?

Answer 1

You can use unicode_string() to stringify your URI as follow:

from sqlalchemy.ext.asyncio import create_async_engine

create_async_engine(settings.POSTGRES_URI.unicode_string())

Check documentation page here for additional explanation.