FastAPI+SQLAlchemy连接AWS RDS PostgreSQL时类型转换错误求助

问题背景 #

连接AWS RDS PostgreSQL时触发如下错误：

sqlalchemy.exc.DataError: (psycopg2.errors.InvalidTextRepresentation) invalid input syntax for type integer: cvb@example.com

具体报错栈：

in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.InvalidTextRepresentation: invalid input syntax for type integer: "cvb@example.com"
LINE 3: WHERE "User".id = 'cvb@example.com'

但连接SQLite时数据可正常插入。

问题根源 #

1. 主键类型配置冲突：SQLAlchemy模型中id字段定义为String类型却添加了autoincrement=True，PostgreSQL仅支持整数类型主键使用自增属性，会自动将该列转为整数类型；而SQLite对类型限制宽松，允许字符串自增，导致跨数据库环境出现类型不匹配。
2. 查询逻辑错误：服务层使用query.get()方法查询用户，但get()默认按主键字段查询，代码中却传入email值，导致用字符串类型的email匹配整数类型的主键，触发语法错误。
3. 字段类型不匹配：phone字段定义为Integer，但经过格式化后的号码包含+、-等特殊字符，无法存入整数类型，SQLite隐式转换未报错，但PostgreSQL严格校验触发错误。

解决方案 #

1. 修正SQLAlchemy User模型 #

# Model.py 修改后
class User(Base):
    __tablename__ = "User"

    id = Column(Integer, primary_key=True, autoincrement=True)  # 改为自增整数主键
    email = Column(String(255), unique=True, nullable=False)  # 保留email唯一性约束
    password = Column(String(255))
    first_name = Column(String(255))
    last_name = Column(String(255))
    phone = Column(String(255))  # 改为String类型存储格式化后的号码
    is_active = Column(Boolean, default=True)
    is_superuser = Column(Boolean(), default=False)

    def __init__(self, first_name, last_name, email, phone, password, *args, **kwargs):
        self.first_name = first_name
        self.last_name = last_name
        self.email = email
        self.phone = phone
        self.password = hashing.get_password_hash(password)

    def check_password(self, password):
        return hashing.verify_password(self.password, password)

2. 修正服务层查询逻辑 #

将get()改为按email字段查询的filter_by()：

# Services 修改后
class UserServices(BaseService[User, UserCreate, UserUpdate]):
    def __init__(self, db_session: Session):
        super(UserServices, self).__init__(User, db_session)

    def create(self, obj: UserCreate) -> User:
        # 按email字段查询用户是否存在
        user = self.db_session.query(User).filter_by(email=obj.email).first()
        if user:
            raise HTTPException(
                status_code=400,
                detail=f"User with email = {obj.email} already exists.",
            )
        return super(UserServices, self).create(obj)

    def get_user_email(self, obj: str) -> User:
        # 按email字段查询用户
        user = self.db_session.query(User).filter_by(email=obj).first()
        if user is None:
            raise HTTPException(
                status_code=400,
                detail=f"User with email = {obj} does not exist.",
            )
        return user  # 原代码返回obj，修正为返回查询到的user对象

3. 同步Pydantic模型字段类型 #

# db/users.py 修改后
class User(BaseModel):
    first_name: str
    last_name: str
    email: EmailStr
    phone: Optional[str] = None  # 改为String类型
    user_type: str  # 修正原代码语法错误（缺失冒号）
    is_active: Optional[bool] = True
    is_superuser: bool = False

    @validator('phone')
    def check_phone_number(cls, v):
        if v is None:
            return v

        try:
            n = parse_phone_number(v, 'GB')
        except NumberParseException as e:
            raise ValueError('Please provide a valid mobile phone number') from e

        if not is_valid_number(n) or number_type(n) not in MOBILE_NUMBER_TYPES:
            raise ValueError('Please provide a valid mobile phone number')

        return format_number(n, PhoneNumberFormat.NATIONAL if n.country_code == 44 else PhoneNumberFormat.INTERNATIONAL)

    class Config:
        orm_mode = True
        alias_generator = to_camel
        allow_population_by_field_name = True

4. 重建数据库表 #

修改模型后，需重新生成数据库表：

• 若使用Alembic迁移工具，执行alembic revision --autogenerate生成迁移脚本，再执行alembic upgrade head应用迁移。
• 开发环境可直接删除旧表，重启服务后自动创建新表。

内容的提问来源于stack exchange，提问作者Rohit Jondhale