Getting error
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'
while connecting to aws rds postgres however when connected to sqlite the data succesfully injected in database.
Could please me with the possible solution.
Please find the code below.
Model.py
class User(Base):
__tablename__ = "User"
id = Column(String, primary_key=True, autoincrement=True)
email = Column(String(255), unique=True)
password = Column(String(255))
first_name = Column(String(255))
last_name = Column(String(255))
phone = Column(Integer)
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)
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:
user = self.db_session.query(User).get(obj.email)
if user:
raise HTTPException(
status_code=400,
detail=f"User with email = {obj.email} is already exist.",
)
return super(UserServices, self).create(obj)
def get_user_email(self, obj: str) -> User:
user = self.db_session.query(User).get(obj)
if user is None:
raise HTTPException(
status_code=400,
detail=f"User with email = {obj} not exist.",
)
return obj
Session.py
SQLALCHEMY_DATABASE_URL = "postgresql://postgres:password2@hostname:5432/postgres"
engine = create_engine(
SQLALCHEMY_DATABASE_URL)
@lru_cache
def create_session() -> scoped_session:
Session = scoped_session(
sessionmaker(autocommit=False, autoflush=False, bind=engine)
)
return Session
# Dependency
def get_session() -> Generator[scoped_session, None, None]:
Session = create_session()
try:
yield Session
finally:
Session.remove()
db/users.py
from typing import List, Optional
from pydantic import BaseModel, EmailStr, constr, validator
from .utils import to_camel
from phonenumbers import (
NumberParseException,
PhoneNumberFormat,
PhoneNumberType,
format_number,
is_valid_number,
number_type,
parse as parse_phone_number,
)
MOBILE_NUMBER_TYPES = PhoneNumberType.MOBILE, PhoneNumberType.FIXED_LINE_OR_MOBILE
class UserEmail(BaseModel):
email: EmailStr
class User(BaseModel):
first_name: str
last_name: str
email: EmailStr
phone: constr(max_length=50, strip_whitespace=True) = None
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
orm_mode = True
alias_generator = to_camel
allow_population_by_field_name = True
class UserCreate(User):
password: str
class Config:
orm_mode = True
alias_generator = to_camel
allow_population_by_field_name = True
class UserUpdate(User):
pass
I am new to the Fastapi and trying to connect aws rds postgres
