Firstly, I have seen question related to this multiple times. However, I did not understand the solution and some of them did not work for me. I used lazy='subquery' and that did not work as well.
This is the table
class Company(Base):
id: uuid.UUID = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4())
owner_id: uuid.UUID = Column(UUID(as_uuid=True), ForeignKey('user.id', ondelete='CASCADE'))
name: str = Column(String(200), nullable=False, index=True)
is_active: bool = Column(Boolean, default=False)
addresses: List[Address] = relationship('Address', back_populates='company')
posts: List[Post] = relationship('Post', back_populates='company')
class Address(Base):
id: uuid.UUID = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4())
address_1: str = Column(String(100))
address_2: str = Column(String(100))
country: str = Column(String(100))
city: str = Column(String(100))
zipcode: str = Column(String(20))
# profile_id = Column(UUID(as_uuid=True), ForeignKey('profile.id'))
company_id: uuid.UUID = Column(UUID(as_uuid=True), ForeignKey('company.id'))
company = relationship('Company', back_populates='addresses')
In my case, I need to insert in 3 tables while creating a company.
async def company_create(info: Info, input: CompanyCreateInput) -> CompanyCreateResult:
try:
address = input.address
input_without_address = dict(filter(lambda x: x[0] != 'address', input.__dict__.items())) # needs improvement
data = CompanyCreate(owner_id=user.id, **input_without_address, address=address.__dict__)
except ValidationError as err:
error_msg = [Error(loc=str(e['loc']), msg = str(e['msg'])) for e in json.loads(err.json())]
return CompanyCreateError(error=error_msg)
async with async_session() as session:
if await company_by_name(session=session, name=data.name) is not None:
return CompanyCreateError(error=[Error(loc="name", msg="Name already exists.")])
company, address = await create_company(input, owner_id=user.id, address=input.address, session=session)
print("dir company", company.addresses)
print('company, address', company.id)
# company.address = address
return CompanyCreateSuccess(company=cast(CompanyType, company))
async def create_company(company_input: CompanyCreate, owner_id, address, session: AsyncSession) -> Optional[Company]:
print("company_input", company_input, owner_id, address)
input_without_address = dict(filter(lambda x: x[0] != 'address', company_input.__dict__.items()))
company = Company(**input_without_address, owner_id=owner_id)
session.add(instance=company)
await session.flush()
company_user = CompanyUser(user_id=owner_id, company_id=company.id)
session.add(instance=company_user)
await session.flush()
address = Address(**address.__dict__, company_id=company.id)
session.add(instance=address)
await session.commit()
return (company, address)
I cannot able to access addresses because of which None is returned while casting company object with CompanyType. Can anyone explain me what this problem is saying based on my code scenario?
Note: I am using sqlalchemy 1.4