For some reason, in this example, the optional is_active attribute is not getting set to default value.
from pydantic import BaseModel, EmailStr
from datetime import datetime
# Pydantic schemas
# Shared properties
class UserBase(BaseModel):
email: Optional[EmailStr] = None
is_active: Optional[bool] = True
is_superuser: bool = False
username: Optional[str] = None
# Properties to receive via API on creation
class UserCreate(UserBase):
email: EmailStr
password: str
# sqlalchemy model
class User(Base):
id = Column(Integer, primary_key=True, index=True)
username = Column(String(25), index=True, unique=True, nullable=False)
email = Column(String(50), unique=True, index=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
is_active = Column(Boolean(), default=True, nullable=False)
is_superuser = Column(Boolean(), default=False, nullable=False)
__mapper_args__ = {"eager_defaults": True}
I was expecting the default value of is_active
, being optional input, to be True
. But I get None
if not explicitly passed.
obj_in = UserCreate(email=email, password=password, username=username)
print(obj_in.is_active)
# True
db_obj = User(
email=obj_in.email,
hashed_password=get_password_hash(obj_in.password),
username=obj_in.username,
is_superuser=obj_in.is_superuser,
# is_active=obj_in.is_active,
)
print(db_obj.is_active)
# None
# I have to add the is_active flag explicitly
db_obj = User(
email=obj_in.email,
hashed_password=get_password_hash(obj_in.password),
username=obj_in.username,
is_superuser=obj_in.is_superuser,
is_active=obj_in.is_active,
)
print(db_obj.is_active)
# True
CodePudding user response:
In hindsight, super obvious. (isn't everything). There were two issues:
I was checking the value of db_obj before actually inserting it into the database. For whatever reason I though object returned by sqlalchemy Model
db_obj = User(...)
would have also the default value assigned. Turns out the default value is assigned only after inserting the object into the db.The second issue that affected my use case was that I didn't flush the session.
So, fixing these two points, we get :
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
engine = create_async_engine(
SQLALCHEMY_DATABASE_URI,
echo=True,
)
# expire_on_commit=False will prevent attributes from being expired
# after commit.
async_session = sessionmaker(
engine, expire_on_commit=False, class_=AsyncSession
)
obj_in = UserCreate(email=email, password=password, username=username)
db_obj = User(
email=obj_in.email,
hashed_password=get_password_hash(obj_in.password),
username=obj_in.username,
is_superuser=obj_in.is_superuser,
# is_active=obj_in.is_active,
)
async with async_session() as session, session.begin():
session.add(db_obj)
await session.flush()
# this assertion fails if above flush is removed.
assert db_obj.is_active == obj_in.is_active