将 PostgresDsn.build 从 pydentic v1 迁移到 pydantic v2

Question

我有 FastAPI 教程中的简单 Config 类。但它似乎使用旧的 pydantic 版本。我使用 pydantic v2 版本运行我的代码并收到几个错误。我几乎修复了所有这些，但最后一个我还无法修复。这是不起作用的代码的一部分：

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)

这就是我得到的错误：

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

我检查了很多地方，但找不到如何解决这个问题，它看起来像方法将数据作为实例而不是字符串build传递给 sqlalchemycreate_engine方法。MultiHostUrl我应该如何正确迁移此代码以使用 pydantic v2？

更新

SQLALCHEMY_DATABASE_URI: Optional[PostgresDsn] = None我通过将键入更改为to解决了这个问题SQLALCHEMY_DATABASE_URI: Optional[str] = None。因为 pydantic 由于某种原因会自动转换结果。但我不确定这种方法是否正确，也许有更好的方法可以做到这一点？

Answer 1

我遇到了同样的问题，最终提出了以下迁移：

派丹蒂克 V1：

from pydantic import BaseSettings, PostgresDsn, validator


class Settings(BaseSettings):
    POSTGRES_SERVER: Optional[str]
    POSTGRES_USER: Optional[str]
    POSTGRES_PASSWORD: Optional[str]
    POSTGRES_DB: Optional[str]
    SQLALCHEMY_DATABASE_URI: Union[Optional[PostgresDsn], Optional[str]] = None

    @validator("SQLALCHEMY_DATABASE_URI", pre=True)
    def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any:
        if isinstance(v, str):
            print("Loading SQLALCHEMY_DATABASE_URI from docker.env file ...")
            return v
        print("Creating SQLALCHEMY_DATABASE_URI from .env file ...")
        return PostgresDsn.build(
            scheme="postgresql",
            user=values.get("POSTGRES_USER"),
            password=values.get("POSTGRES_PASSWORD"),
            host=values.get("POSTGRES_SERVER"),
            path=f"/{values.get('POSTGRES_DB') or ''}",
        )

    class Config:
        env_file = ".env"
        case_sensitive = True

派丹蒂克 V2：

from pydantic import PostgresDsn, field_validator, ValidationInfo
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", case_sensitive=True)

    POSTGRES_SERVER: Optional[str] = None
    POSTGRES_USER: Optional[str] = None
    POSTGRES_PASSWORD: Optional[str] = None
    POSTGRES_DB: Optional[str] = None
    SQLALCHEMY_DATABASE_URI: Union[Optional[PostgresDsn], Optional[str]] = None

    @field_validator("SQLALCHEMY_DATABASE_URI", mode="before")
    @classmethod
    def assemble_db_connection(cls, v: Optional[str], values: ValidationInfo) -> Any:
        if isinstance(v, str):
            print("Loading SQLALCHEMY_DATABASE_URI from .docker.env file ...")
            return v
        print("Creating SQLALCHEMY_DATABASE_URI from .env file ...")
        return PostgresDsn.build(
            scheme="postgresql",
            username=values.data.get("POSTGRES_USER"),
            password=values.data.get("POSTGRES_PASSWORD"),
            host=values.data.get("POSTGRES_SERVER"),
            path=f"{values.data.get('POSTGRES_DB') or ''}",
        )          

另外，为了调用settings实例对象，您需要在 V2 中转换为字符串：

settings.SQLALCHEMY_DATABASE_URI.unicode_string()
# or
f"{settings.SQLALCHEMY_DATABASE_URI}

Answer 2

您可以按如下方式unicode_string()对您的字符串进行字符串化：URI

from sqlalchemy.ext.asyncio import create_async_engine

create_async_engine(settings.POSTGRES_URI.unicode_string())

请查看此处的文档页面以获取更多说明。