[Python] Filter Parent class object based on list of child class values in Python | FastAPI |...

Entities

class ParentEntity(Base):
__tablename__ = "parent_entity"

id = Column(Integer, primary_key=True)
status = Column(String)
config_id = Column(Integer, ForeignKey("config_table.id"), nullable=False)
updated_at = Column(DateTime)
is_active = Column(Boolean, default=True)

config = relationship("ConfigTable", back_populates="parent_entities", lazy="selectin")
child_entities = relationship(
"ChildEntity",
back_populates="parent_entity",
primaryjoin="and_(ParentEntity.id == ChildEntity.parent_id, ChildEntity.is_active == true())",
lazy="selectin"
)

class ChildEntity(Base):
__tablename__ = "child_entity"

id = Column(Integer, primary_key=True)
parent_id = Column(Integer, ForeignKey("parent_entity.id"))
client_id = Column(Integer) # added to match filtering use case
name = Column(String(255))
is_active = Column(Boolean, default=True)

parent_entity = relationship("ParentEntity", back_populates="child_entities", lazy="select")



class ConfigTable(Base):
__tablename__ = "config_table"

id = Column(Integer, primary_key=True)
entity_type = Column(String(255))
parent_entities = relationship("ParentEntity", back_populates="config", lazy="selectin")


Parent entitiy repository method.

async def get_all_parent_entities_with_children(
db: AsyncSession,
entity_type: str,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
status_filter: Optional[str] = None,
skip: int = 0,
limit: int = 10
) -> Tuple[List[Optional[ParentEntity]], int]:
"""Retrieve all ParentEntities with ChildEntities and Config details."""
try:
# Dynamic conditions
conditions = [ParentEntity.is_active == True]

if status_filter:
conditions.append(ParentEntity.status == status_filter)
if start_date:
conditions.append(ParentEntity.updated_at >= start_date)
if end_date:
conditions.append(ParentEntity.updated_at <= end_date)
if entity_type:
conditions.append(ConfigTable.entity_type == entity_type)

# Main query
query = (
select(ParentEntity)
.join(ConfigTable, ParentEntity.config_id == ConfigTable.id)
.options(
selectinload(ParentEntity.child_entities),
selectinload(ParentEntity.config)
)
.where(and_(*conditions))
.order_by(ParentEntity.updated_at.desc())
.limit(limit)
.offset(skip)
)

result = await db.execute(query)
parent_entities = result.scalars().all()

# Count query
count_query = (
select(func.count(ParentEntity.id))
.join(ConfigTable, ParentEntity.config_id == ConfigTable.id)
.where(and_(*conditions))
)

total_count = (await db.execute(count_query)).scalar_one()

return parent_entities, total_count

except Exception as e:
logger.error(f"Error fetching entities: {str(e)}")
raise


I am using below logic in my service method calling above repository method

# Filter child_entities by client_id if provided
filtered_children = (
list(filter(
lambda c: c.client_id == given_client_id,
parent_entity.child_entities
))
if given_client_id is not None
else parent_entity.child_entities
)


here you can see when I get like 100 parent client objects from repository method based on my pagination logic, Then if given_client_id is not none, then I have to filter child_entities for a particular parent that has client_id equals to given_client_id only. if a parent entity does not have any child with client_id same as given_client_id, still that parent entity I am returning from my service method which I don't want. I tried multiple approaches but failing again and again. Please help me here.

I want only those parents which has atleast one child entity with its client_id == given_client_id in my 100 objects. I am not able to figure out how to do it in python using fastapi and SQLAlchemy.

Continue reading...